uoload

.gitignore

@@ -0,0 +1,6 @@
+docker/
+.idea
+/vendor/
+composer.lock
+/Tests/config.php
+runtime

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2021 sunnywoon
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1 @@
+# workerman-queue

Tests/TestHandler.php

@@ -0,0 +1,41 @@
+<?php
+/**
+ *
+ * @date 2021/6/24 16:40
+ */
+
+namespace Tests;
+
+class TestHandler extends \WorkerManQueue\Handler\JobHandler
+{
+    /**
+     * 失败回调方法
+     * @param \WorkerManQueue\Job $job 任务
+     * @param string $func 执行的方法
+     * @param array $data 参数
+     * @return mixed
+     */
+    public function failed(\WorkerManQueue\Job $job, $func, $data)
+    {
+        \WorkerManQueue\Helpers\Log::info('failed run handler -- func: ' . $func . ' -- params: ' . json_encode($data));
+    }
+
+    /**
+     * 任务成功回调
+     * @param \WorkerManQueue\Job $job 任务
+     * @param string $func 执行的方法
+     * @param array $data 参数
+     * @return mixed
+     */
+    public function success(\WorkerManQueue\Job $job, $func, $data)
+    {
+        \WorkerManQueue\Helpers\Log::info('success run handler -- func: ' . $func . ' -- params: ' . json_encode($data));
+    }
+
+
+    public function test(\WorkerManQueue\Job $job, $data)
+    {
+        $res = 'success';
+        \WorkerManQueue\Helpers\Log::info('run handler -- func: test -- params: ' . json_encode($data) . '; result : ' . var_export($res, true));
+    }
+}

Tests/bootstrap.php

@@ -0,0 +1,2 @@
+<?php
+require dirname(__DIR__)."/vendor/autoload.php";

Tests/config_example.php

@@ -0,0 +1,24 @@
+<?php
+
+return [
+    'log' => [
+        'logRoot'  => __DIR__ . '/../runtime/log',
+        'fileName' => '\q\u\e\u\e_Y-m-d.\l\o\g',
+    ],
+
+    'connectList' => [
+        'Redis' => [
+            'class'  => '\\WorkerManQueue\\Connection\\Redis\\Redis',
+            'config' => [
+                'popTimeout'  => 3,               // pop阻塞的超时时长 s
+                'host'        => '127.0.0.1',     // 数据库地址
+                'port'        => 6379,            // 数据库端口
+                'db'          => 0,               // 库
+                'password'    => null,            // 密码
+                'connTimeout' => 1,               // 链接超时
+            ],
+        ]
+    ],
+
+    'currentConnect' => 'Redis',
+];

Tests/push.php

@@ -0,0 +1,13 @@
+<?php
+require __DIR__ . "/bootstrap.php";
+require __DIR__ . "/TestHandler.php";
+
+$config = include __DIR__ . '/config.php';
+
+$queue = \WorkerManQueue\Queue::getInstance('Redis', $config);
+
+use Tests\TestHandler;
+
+for ($i = 0; $i <= 100; $i++) {
+    $r = $queue->pushOn(new TestHandler(), 'test', ['test' => $i], 'queue');
+}

Tests/test.php

@@ -0,0 +1,9 @@
+<?php
+
+require __DIR__."/bootstrap.php";
+
+$config = include __DIR__.'/config.php';
+
+$worker = new \WorkerManQueue\Worker($config);
+
+$worker->run();

composer.json

@@ -0,0 +1,12 @@
+{
+    "name": "woon/workerman-queue",
+    "require": {
+        "workerman/workerman": "^4.0"
+    },
+    "autoload": {
+        "psr-4": {
+            "WorkerManQueue\\": "src/",
+            "Tests\\": "Tests/"
+        }
+    }
+}

src/Connection/Connection.php

@@ -0,0 +1,166 @@
+<?php
+/**
+ *
+ * @date 2021/6/24 15:02
+ */
+
+namespace WorkerManQueue\Connection;
+
+use WorkerManQueue\Exception;
+use WorkerManQueue\Job;
+
+/**
+ * 连接类
+ * Class Connection
+ * @package WorkerManQueue\Connection
+ */
+abstract class Connection
+{
+    /**
+     * config
+     * @var array
+     */
+    protected $config = [];
+
+    /**
+     * singleton
+     * @var Connection
+     */
+    protected static $instance = null;
+
+    /**
+     * pop阻塞超时时长
+     * @var int
+     */
+    public $popTimeOut = 3;
+
+    /**
+     * 处理程序
+     * @var \Closure
+     */
+    protected $handler = null;
+
+    /**
+     * Connection constructor.
+     * @param array $config 配置参数
+     */
+    protected function __construct(array $config = [])
+    {
+        $this->config = $config;
+        if (isset($config['popTimeout']) && $config['popTimeout'] > 0) {
+            $this->popTimeOut = $config['popTimeout'];
+        }
+    }
+
+    /**
+     * 设置处理程序
+     * @param \Closure $handler
+     */
+    public function setHandler(\Closure $handler)
+    {
+        $this->handler = $handler;
+    }
+
+    /**
+     * Connection destruct.
+     */
+    public function __destruct()
+    {
+        $this->close();
+        static::$instance = null;
+    }
+
+    /**
+     * 不允许被克隆
+     * @throws Exception
+     */
+    protected function __clone()
+    {
+        throw new Exception("This class cannot be cloned", -101);
+    }
+
+    /**
+     * 获取单例
+     * @param array $config 配置参数
+     * @return Connection|null
+     */
+    public static function getInstance($config = [])
+    {
+        if (!(static::$instance instanceof Connection)) {
+            static::$instance = new static($config);
+        }
+        return static::$instance;
+    }
+
+    /**
+     * 关闭连接
+     * @return boolean
+     */
+    abstract public function close();
+
+    /**
+     * 执行pop出来的任务(阻塞方法)
+     * @param string $queueName
+     */
+    public function popRun($queueName, $popTimeOut = 0)
+    {
+        $extends = [];
+        $job     = $this->pop($queueName, $popTimeOut, $extends);
+        if ($job instanceof Job) {
+            // 执行任务
+            $this->runJob($job, $queueName);
+
+            // 确认任务
+            $this->ack($queueName, $job, $extends);
+        }
+    }
+
+    /**
+     * 执行任务
+     * @param Job $job
+     * @param $queueName
+     */
+    public function runJob(Job $job, $queueName)
+    {
+        // 执行任务
+        $handler = $this->handler;
+        $handler($job, $queueName);
+    }
+
+    /**
+     * 弹出队头任务(blocking)
+     * @param string $queueName 队列名称
+     * @param int $popTimeOut 阻塞时间
+     * @param array & $extends 额外需要传递给ack方法的参数
+     * @return Job|null
+     */
+    abstract protected function pop($queueName, $popTimeOut = 0, &$extends = []);
+
+    /**
+     * 确认任务
+     * @param string $queueName
+     * @param Job $job
+     * @param array $extends
+     */
+    abstract protected function ack($queueName, Job $job = null, $extends = []);
+
+    /**
+     * 压入队列
+     * @param Job $job
+     * @param String $queueName 队列名
+     * @return boolean
+     */
+    abstract public function push(Job $job, $queueName);
+
+
+    /**
+     * 添加一条延迟任务
+     * @param int $delay 延迟的秒数
+     * @param Job $job 任务
+     * @param String $queueName 队列名
+     * @return boolean
+     */
+    abstract public function later($delay, Job $job, $queueName);
+
+
+}

src/Connection/ConnectionFactory.php

@@ -0,0 +1,55 @@
+<?php
+/**
+ *
+ * @date 2021/6/24 15:12
+ */
+
+namespace WorkerManQueue\Connection;
+
+
+use WorkerManQueue\Helpers\Log;
+
+class ConnectionFactory
+{
+    /**
+     * @var array 链接配置列表
+     * example:
+     *
+     */
+    public static $connectList = [
+        'Redis' => [
+            'class'  => '\\WorkerManQueue\\Connection\\Redis\\Redis',
+            'config' => [
+                'popTimeout'  => 3,               // pop阻塞的超时时长 s
+                'host'        => '127.0.0.1',     // 数据库地址
+                'port'        => 6379,            // 数据库端口
+                'db'          => 0,               // 库
+                'password'    => null,            // 密码
+                'connTimeout' => 1,               // 链接超时
+            ],
+        ]
+    ];
+
+    /**
+     * @var string 当前默认使用的链接
+     */
+    public static $currentConnect = 'Redis';
+
+    /**
+     * 获取链接对象
+     * @param string $currentName 当前链接方式
+     * @return Connection
+     */
+    public static function getInstance($currentName)
+    {
+        $connect = isset(self::$connectList[$currentName]) ? self::$connectList[$currentName] : [];
+        if (empty($connect) || !isset($connect['class']) || empty($connect['class'])) {
+            Log::error('There is no connection available type');
+            return null;
+        } else {
+            $class  = $connect['class'];
+            $config = isset($connect['config']) ? $connect['config'] : [];
+            return $class::getInstance($config);
+        }
+    }
+}

src/Connection/Redis/Redis.php

@@ -0,0 +1,273 @@
+<?php
+/**
+ *
+ * @date 2021/6/24 15:16
+ */
+
+namespace WorkerManQueue\Connection\Redis;
+
+
+use WorkerManQueue\Connection\Connection;
+use WorkerManQueue\Job;
+
+class Redis extends Connection
+{
+    /**
+     * redis单例对象
+     * @var Redis
+     */
+    protected static $instance = null;
+
+    /**
+     * redis host
+     * @var string
+     */
+    private $host = '127.0.0.1';
+
+    /**
+     * redis port
+     * @var int
+     */
+    private $port = 6379;
+
+    /**
+     * redis database
+     * @var int
+     */
+    private $database = 0;
+
+    /**
+     * redis password
+     * @var string
+     */
+    private $password = '';
+
+    /**
+     * redis timeout
+     * @var int
+     */
+    private $connTimeout = 0;
+
+    /**
+     * redis驱动
+     * @var \Redis
+     */
+    private $connect = null;
+
+    /**
+     * 本类单例
+     * @var Redis
+     */
+    /**
+     * Connection constructor.
+     * @param array $config 配置参数
+     */
+    protected function __construct(array $config = [])
+    {
+        parent::__construct($config);
+        $this->host        = (isset($config['host']) && !empty($config['host'])) ? $config['host'] : $this->host;
+        $this->port        = (isset($config['port']) && !empty($config['port'])) ? $config['port'] : $this->port;
+        $this->database    = (isset($config['db']) && !empty($config['db'])) ? $config['db'] : $this->database;
+        $this->password    = (isset($config['password']) && !empty($config['password'])) ? $config['password'] : $this->password;
+        $this->connTimeout = (isset($config['connTimeout']) && !empty($config['connTimeout'])) ? $config['connTimeout'] : $this->connTimeout;
+    }
+
+    /**
+     * get connect
+     * @return \Redis
+     */
+    private function getConnect()
+    {
+        if (empty($this->connect)) {
+            $this->connect = new \Redis();
+            $this->open();
+        } else {
+            if (@$this->connect->ping() !== '+PONG') {
+                $this->open();
+            }
+        }
+        return $this->connect;
+    }
+
+    /**
+     * open redis connect
+     */
+    private function open()
+    {
+        $this->connect->connect($this->host, $this->port, $this->connTimeout);
+        if (!empty($this->password)) {
+            $this->connect->auth($this->password);
+        }
+        $this->connect->select($this->database);
+    }
+
+    /**
+     * 关闭连接
+     * @return boolean
+     */
+    public function close()
+    {
+        if (!empty($this->connect)) {
+            $this->connect->close();
+        }
+        return true;
+    }
+
+    /**
+     * 弹出队头任务(blocking)
+     * @param string $queueName 队列名称
+     * @param int $popTimeOut 阻塞时间
+     * @param array & $extends 额外需要传递给ack方法的参数
+     * @return Job|null
+     */
+    protected function pop($queueName, $popTimeOut = 0, &$extends = [])
+    {
+        //从延迟集合中合并到主执行队列
+        $this->migrateAllExpiredJobs($queueName);
+
+        if ($popTimeOut) {
+            $jobStr = $this->getConnect()->blPop($queueName, $popTimeOut);
+            if (isset($jobStr[1]) && !empty($jobStr[1])) {
+                $jobStr = $jobStr[1];
+            } else {
+                return null;
+            }
+        } else {
+            $jobStr = $this->getConnect()->lPop($queueName);
+        }
+
+        if (empty($jobStr)) {
+            return null;
+        } else {
+            return Job::Decode($jobStr);
+        }
+    }
+
+    /**
+     * 确认任务
+     * @param string $queueName
+     * @param Job $job
+     * @param array $extends
+     */
+    public function ack($queueName, Job $job = null, $extends = [])
+    {
+        // redis不需要确认任务
+    }
+
+
+    /**
+     * 压入队列(直接压入主执行队列)
+     *
+     * @param Job $job
+     * @param string $queueName 队列名称
+     *
+     * @return boolean
+     */
+    public function push(Job $job, $queueName)
+    {
+        //命令：rpush 队列名 任务
+        $length = $this->getConnect()->rPush($queueName, Job::Encode($job));
+        if ($length) {
+            return true;
+        } else {
+            return false;
+        }
+    }
+
+    /**
+     * 添加一条延迟任务
+     * 放入等待执行任务的有序集合中
+     *
+     * @param int $delay 延迟的秒数
+     * @param Job $job 任务
+     * @param string $queueName 队列名称
+     *
+     * @return boolean
+     */
+    public function later($delay, Job $job, $queueName)
+    {
+        //命令：zadd  主队列名:delayed   当前时间戳+延迟秒数  任务
+        $result = $this->getConnect()->zAdd($queueName . ':delayed', time() + $delay, Job::Encode($job));
+        if ($result) {
+            return true;
+        } else {
+            return false;
+        }
+    }
+
+
+    /**
+     * 合并等待执行的任务
+     * @param string $queueName
+     * @return void
+     */
+    protected function migrateAllExpiredJobs($queueName)
+    {
+        $this->migrateExpiredJobs($queueName . ':delayed', $queueName);
+    }
+
+
+    /**
+     * 当延时任务到大执行时间时，将延时任务从延时任务集合中移动到主执行队列中
+     * @param string $from 集合名称
+     * @param string $to 队列名称
+     * @return void
+     */
+    protected function migrateExpiredJobs($from, $to)
+    {
+        $time = time();
+        $jobs = $this->getExpiredJobs($from, $time);
+        if (count($jobs) > 0) {
+            $connect = $this->getConnect();
+            //开始redis事物
+            $connect->watch($from);
+            $connect->multi();
+
+            $this->removeExpiredJobs($from, $time);
+            $this->pushExpiredJobsOntoNewQueue($to, $jobs);
+
+            $connect->exec();
+        }
+    }
+
+
+    /**
+     * 从指定集合中获取所有超时的任务
+     * @param String $name 集合名称
+     * @param int $time 超时时间(集合中小于该时间为超时)
+     * @return mixed
+     */
+    public function getExpiredJobs($name, $time)
+    {
+        return $this->getConnect()->zRangeByScore($name, '-inf', $time);
+    }
+
+
+    /**
+     * 从指定集合删除过期任务
+     * @param string $from
+     * @param int $time
+     * @return void
+     */
+    protected function removeExpiredJobs($from, $time)
+    {
+        $this->getConnect()->zRemRangeByScore($from, '-inf', $time);
+    }
+
+
+    /**
+     * 将多个任务从添加到队列
+     *
+     * 场景：将有序集合中的延迟任务入主队列
+     * @param string $to
+     * @param array $jobs
+     * @return void
+     */
+    protected function pushExpiredJobsOntoNewQueue($to, $jobs)
+    {
+        //等价于  $connect->rPush($to,$jobs[0],$jobs[1]... );
+        $connect = $this->getConnect();
+        call_user_func_array([$connect, 'rPush'], array_merge([$to], $jobs));
+    }
+
+}

src/Exception/Exception.php

@@ -0,0 +1,13 @@
+<?php
+/**
+ *
+ * @date 2021/6/24 15:01
+ */
+
+namespace WorkerManQueue;
+
+
+class Exception extends \Exception
+{
+
+}

src/Handler/JobHandler.php

@@ -0,0 +1,63 @@
+<?php
+/**
+ *
+ * @date 2021/6/24 15:07
+ */
+
+namespace WorkerManQueue\Handler;
+
+use WorkerManQueue\Job;
+
+abstract class JobHandler
+{
+    /**
+     * 回调执行任务方法
+     * @param Job $job 任务
+     * @param String $func 执行的方法
+     * @param array $data 参数
+     * @return void
+     */
+    public function handler(Job $job, $func, $data)
+    {
+        try {
+            if (method_exists($this, $func)) {
+                $this->$func($job, $data);
+            } else {
+                $job->setForceFailure('method "' . $func . '" does not exist');
+            }
+        } catch (\Exception $e) {
+            $job->setOnceFailure($e->getMessage());
+        }
+    }
+
+
+    /**
+     * 失败回调方法
+     * @param Job $job 任务
+     * @param string $func 执行的方法
+     * @param array $data 参数
+     * @return mixed
+     */
+    abstract public function failed(Job $job, $func, $data);
+
+
+    /**
+     * 任务成功回调
+     * @param Job $job 任务
+     * @param string $func 执行的方法
+     * @param array $data 参数
+     * @return mixed
+     */
+    abstract public function success(Job $job, $func, $data);
+
+
+    /**
+     * 回调方法
+     * @param $job
+     * @param $data
+     */
+    /**
+     * public function func($job,$data){}
+     */
+
+}

src/Helpers/LoadConfig.php

@@ -0,0 +1,33 @@
+<?php
+/**
+ *
+ * @date 2021/6/24 15:14
+ */
+
+namespace WorkerManQueue\Helpers;
+
+trait LoadConfig
+{
+    /**
+     * 允许配置的变量名
+     * @var array
+     */
+    protected $configNameList = [];
+
+    /**
+     * 加载配置
+     * @param array $config
+     * @return $this
+     */
+    public function setConfig(array $config)
+    {
+        foreach ($config as $k => $v) {
+            if (in_array($k, $this->configNameList)) {
+                if (!is_null($v)) {
+                    $this->$k = $v;
+                }
+            }
+        }
+        return $this;
+    }
+}

src/Helpers/Log.php

@@ -0,0 +1,131 @@
+<?php
+/**
+ *
+ * @date 2021/6/24 15:13
+ */
+
+namespace WorkerManQueue\Helpers;
+
+class Log
+{
+    use LoadConfig;
+
+    /**
+     * 文件路径
+     * @var string
+     */
+    protected $logRoot = __DIR__.'/../../runtime/log';
+
+    /**
+     * 文件名
+     * @var string
+     */
+    protected $fileName = '\q\u\e\u\e_Y-m-d.\l\o\g';
+
+    /**
+     * @var self
+     */
+    protected static $instance = null;
+
+    protected function __construct() {
+        $this->configNameList = ['logRoot', 'fileName'];
+    }
+
+    /**
+     * @return Log
+     */
+    public static function getInstance()
+    {
+        if (!(self::$instance instanceof self)) {
+            self::$instance = new Log();
+        }
+        return self::$instance;
+    }
+
+    /**
+     * @param $level
+     * @param string $message
+     * @param array $extends
+     * @return bool
+     */
+    protected function write($level, $message = '', $extends = [])
+    {
+        if (!is_dir($this->logRoot)) {
+            $mkdir = mkdir($this->logRoot, 0777, true);
+            if (!$mkdir) {
+                return false;
+            }
+        }
+
+        $time = time();
+
+        $filePath = rtrim($this->logRoot, '/') . '/' . date($this->fileName, $time);
+
+        // 信息
+        $string = sprintf(
+            "[%s][%s]: %s; extends=%s \n",
+            $level, date('Y-m-d m:d:s', $time), $message, json_encode($extends)
+        );
+
+        if (file_put_contents($filePath, $string, FILE_APPEND)) {
+            return true;
+        } else {
+            return false;
+        }
+    }
+
+    /**
+     * notice
+     * @param string $message
+     * @param array $extends
+     * @return bool
+     */
+    public static function notice($message = '', $extends = [])
+    {
+        return self::getInstance()->write('notice', $message, $extends);
+    }
+
+    /**
+     * info
+     * @param string $message
+     * @param array $extends
+     * @return bool
+     */
+    public static function info($message = '', $extends = [])
+    {
+        return self::getInstance()->write('info', $message, $extends);
+    }
+
+    /**
+     * warning
+     * @param string $message
+     * @param array $extends
+     * @return bool
+     */
+    public static function warning($message = '', $extends = [])
+    {
+        return self::getInstance()->write('warning', $message, $extends);
+    }
+
+    /**
+     * error
+     * @param string $message
+     * @param array $extends
+     * @return bool
+     */
+    public static function error($message = '', $extends = [])
+    {
+        return self::getInstance()->write('error', $message, $extends);
+    }
+
+    /**
+     * fatal
+     * @param string $message
+     * @param array $extends
+     * @return bool
+     */
+    public static function fatal($message = '', $extends = [])
+    {
+        return self::getInstance()->write('fatal', $message, $extends);
+    }
+}

src/Job.php

@@ -0,0 +1,243 @@
+<?php
+/**
+ *
+ * @date 2021/6/24 15:07
+ */
+
+namespace WorkerManQueue;
+
+
+use WorkerManQueue\Handler\JobHandler;
+
+class Job
+{
+    /**
+     * @var string 工作ID(唯一)
+     */
+    public $id = '';
+
+    /******************* handler ********************/
+    /**
+     * @var JobHandler job handler
+     */
+    protected $handler;
+    /**
+     * @var String 执行的方法
+     */
+    protected $func;
+    /**
+     * @var array 执行方法的参数
+     */
+    protected $param;
+
+
+    /******************* run ********************/
+    /**
+     * 最后一次执行状态 true成功  false失败
+     * @var bool
+     */
+    protected $lastStatus = true;
+    /**
+     * @var bool 是否强制失败(强制失败不会重试)
+     */
+    protected $forceFailed = false;
+    /**
+     * @var string[] 异常信息数组
+     */
+    protected $errorArr = [];
+
+    /**
+     * @var boolean 当前是否执行成功
+     */
+    protected $isExec;
+    /**
+     * @var int 当前已经执行次数
+     */
+    protected $attempts;
+
+    /**
+     * @param JobHandler $handler 回调类
+     * @param String $func 回调类中的回调方法名
+     * @param mixed $param 该回调方法需要的参数
+     */
+    public function __construct(JobHandler $handler, $func, $param)
+    {
+        $this->resetId();
+
+        $this->handler = $handler;
+        $this->func    = $func;
+        $this->param   = $param;
+
+        $this->init();
+    }
+
+    /**
+     * 生成id
+     */
+    public function resetId()
+    {
+        $this->id = md5(uniqid(rand(0, 9999) . microtime(true), true));
+    }
+
+    /**
+     * 初始化默认任务参数
+     */
+    public function init()
+    {
+        $this->isExec   = false;
+        $this->attempts = 0;
+    }
+
+    /**
+     * 该任务已经执行的次数
+     * @return int
+     */
+    public function getAttempts()
+    {
+        return $this->attempts;
+    }
+
+    /**
+     * 任务失败回调
+     * @return void
+     */
+    public function failed()
+    {
+        $this->handler->failed($this, $this->func, $this->param);
+    }
+
+    /**
+     * 任务成功回调
+     * @return void
+     */
+    public function success()
+    {
+        $this->handler->success($this, $this->func, $this->param);
+    }
+
+    /**
+     * 执行任务
+     * @return void
+     */
+    public function execute()
+    {
+        $this->attempts++;
+
+        $this->resetStatus();
+
+        //执行handler回调
+        $this->handler->handler($this, $this->func, $this->param);
+
+        if ($this->lastStatus) {
+            $this->isExec = true;
+        }
+
+    }
+
+    /**
+     * 重制执行状态
+     */
+    protected function resetStatus()
+    {
+        $this->lastStatus  = true;
+        $this->forceFailed = false;
+    }
+
+    /**
+     * 设置本次执行失败(会重试)
+     * @param string $message 错误信息
+     */
+    public function setOnceFailure($message = "")
+    {
+        $this->lastStatus  = false;
+        $this->forceFailed = false;
+        $this->errorArr[]  = $message;
+    }
+
+    /**
+     * 设置本次任务为强制失败(不会重试)
+     * @param string $message 错误信息
+     */
+    public function setForceFailure($message = "")
+    {
+        $this->lastStatus  = false;
+        $this->forceFailed = true;
+        $this->errorArr[]  = $message;
+    }
+
+    /**
+     * 任务是否执行成功
+     * @return boolean
+     */
+    public function isExec()
+    {
+        return $this->isExec;
+    }
+
+
+    /**
+     * 获取任务失败的异常信息数组
+     * @return string[]
+     */
+    public function getErrors()
+    {
+        return $this->errorArr;
+    }
+
+    /**
+     * 是否需要重试该任务
+     * @param int $maxAttempt
+     * @return bool
+     */
+    public function isRetry(int $maxAttempt)
+    {
+        if ($this->isExec()) {
+            // 执行成功不需要重试
+            return false;
+        } else {
+            if (
+                // 判断是否强制失败，如果强制失败，则任务不需要重试
+                ($this->forceFailed) ||
+                // 判断当前是否超出指定执行次数，如果超过最大限制，则任务不需要重试
+                ($maxAttempt > 0 && $this->getAttempts() >= $maxAttempt)
+            ) {
+                return false;
+            } else {
+                return true;
+            }
+        }
+    }
+
+    /**
+     * 重试该任务
+     * @param Queue $queue 队列
+     * @param string $queueName 队列名
+     * @return boolean
+     */
+    public function reTry(Queue $queue, $queueName)
+    {
+        return $queue->push($this, $queueName);
+    }
+
+
+    /**
+     * 序列化对象
+     * @param Job $job
+     * @return string
+     */
+    public static function Encode(Job $job)
+    {
+        return base64_encode(serialize($job));
+    }
+
+
+    /**
+     * 反序列化对象
+     * @param string $jobStr
+     * @return Job
+     */
+    public static function Decode($jobStr)
+    {
+        return unserialize(base64_decode($jobStr));
+    }
+}

src/Load.php

@@ -0,0 +1,36 @@
+<?php
+/**
+ *
+ * @date 2021/6/24 15:35
+ */
+
+namespace WorkerManQueue;
+
+
+use WorkerManQueue\Connection\ConnectionFactory;
+use WorkerManQueue\Helpers\Log;
+
+class Load
+{
+    /**
+     * 加载queue模块依赖的配置
+     * @param array $config
+     */
+    public static function Queue(array $config)
+    {
+        // 加载log
+        if (isset($config['log'])) {
+            Log::getInstance()->setConfig($config['log']);
+        }
+
+        // 加载链接列表
+        if (isset($config['connectList'])) {
+            ConnectionFactory::$connectList = $config['connectList'];
+        }
+
+        // 加载当前链接
+        if (isset($config['currentConnect'])) {
+            ConnectionFactory::$currentConnect = $config['currentConnect'];
+        }
+    }
+}

src/Queue.php

@@ -0,0 +1,154 @@
+<?php
+/**
+ *
+ * @date 2021/6/24 14:54
+ */
+
+namespace WorkerManQueue;
+
+use WorkerManQueue\Connection\Connection;
+use WorkerManQueue\Connection\ConnectionFactory;
+use WorkerManQueue\Handler\JobHandler;
+
+class Queue
+{
+    /**
+     * @var array [Queue] 单例对象数组
+     */
+    protected static $instances;
+
+    /**
+     * @var Connection 连接对象
+     */
+    protected $connection;
+
+    /**
+     * Queue constructor.
+     * @param Connection $connection
+     */
+    protected function __construct(Connection $connection)
+    {
+        $this->connection = $connection;
+    }
+
+    public function __destruct()
+    {
+        static::$instances = null;
+    }
+
+    /**
+     * 不允许被克隆
+     * @throws Exception
+     */
+    protected function __clone()
+    {
+        throw new Exception("This class cannot be cloned", -101);
+    }
+
+    /**
+     * @param string $connectName 链接类型(默认走配置)
+     * @return Queue|null
+     */
+    public static function getInstance($connectName = '', $config = [])
+    {
+        if (!empty($config)) {
+            Load::Queue($config);
+        }
+
+        if (empty($connectName)) {
+            $connectName = ConnectionFactory::$currentConnect;
+        }
+        if (!isset(static::$instances[$connectName]) || !(static::$instances[$connectName] instanceof Queue)) {
+            $connect = ConnectionFactory::getInstance($connectName);
+            if ($connect) {
+                static::$instances[$connectName] = new static($connect);
+            } else {
+                return null;
+            }
+        }
+        return static::$instances[$connectName];
+    }
+
+    /**
+     * 设置处理程序
+     * @param \Closure $handler
+     */
+    public function setHandler(\Closure $handler)
+    {
+        $this->connection->setHandler($handler);
+    }
+
+    /**
+     * 执行pop出来的任务(阻塞方法)
+     * @param string $queueName
+     */
+    public function popRun($queueName)
+    {
+        $this->connection->popRun($queueName);
+    }
+
+    /**
+     * 入队列
+     * @param Job $job
+     * @param string $queueName 队列名
+     * @return boolean
+     */
+    public function push(Job $job, $queueName)
+    {
+        return $this->connection->push($job, $queueName);
+    }
+
+    /**
+     * 延迟入队列
+     * @param int $delay 延迟的秒数
+     * @param Job $job
+     * @param string $queueName 队列名
+     * @return boolean
+     */
+    public function later($delay, Job $job, $queueName)
+    {
+        if ($delay <= 0) {
+            return $this->push($job, $queueName);
+        } else {
+            return $this->connection->later($delay, $job, $queueName);
+        }
+    }
+
+    /**
+     * 入队列  (对外)
+     * @param JobHandler $handler 回调类
+     * @param String $func 方法名
+     * @param mixed $param 参数
+     * @param String $queueName 队列名
+     * @return boolean
+     */
+    public function pushOn(JobHandler $handler, $func, $param, $queueName)
+    {
+        $job = new Job($handler, $func, $param);
+        return $this->push($job, $queueName);
+    }
+
+    /**
+     * 延迟入队列 (对外)
+     * @param Int $delay 延迟时间/秒
+     * @param JobHandler $handler 回调类
+     * @param String $func 方法名
+     * @param mixed $param 参数
+     * @param String $queueName 队列名
+     * @return boolean
+     */
+    public function laterOn($delay, JobHandler $handler, $func, $param, $queueName)
+    {
+        $job = new Job($handler, $func, $param);
+        return $this->later($delay, $job, $queueName);
+    }
+
+
+    /**
+     * 关闭数据库连接
+     */
+    public function close()
+    {
+        $this->connection->close();
+    }
+}

src/Worker.php

@@ -0,0 +1,70 @@
+<?php
+/**
+ *
+ * @date 2021/6/24 15:55
+ */
+
+namespace WorkerManQueue;
+
+
+use Workerman\Timer;
+use WorkerManQueue\Helpers\LoadConfig;
+
+class Worker
+{
+    use LoadConfig;
+
+    protected $config = [];
+
+    /**
+     * Worker constructor.
+     * @param array $config
+     */
+    public function __construct(array $config = [])
+    {
+        $this->config = $config;
+    }
+
+    public function run()
+    {
+        $worker        = new \Workerman\Worker();
+        $worker->count = 10;
+        $worker->name  = 'queue';
+
+        $queueName = $worker->name;
+        $config    = $this->config;
+
+        $worker->onWorkerStart = function () use ($config, $queueName) {
+            Load::Queue($config);
+            $queue   = Queue::getInstance();
+            $attempt = 2;
+
+            $queue->setHandler(function (Job $job, $queueName) use ($queue, $attempt) {
+                // 执行任务
+                $job->execute();
+
+                // 判断任务是否执行成功
+                if ($job->isExec()) {
+                    //任务成功，触发回调
+                    $job->success();
+                } else {
+                    // 是否需要重试该任务
+                    if ($job->isRetry($attempt)) {
+                        // 需要重试，则重新将任务放入队尾
+                        $job->reTry($queue, $queueName);
+                    } else {
+                        // 不需要重试，则任务失败，触发回调
+                        $job->failed();
+                    }
+                }
+            });
+
+            Timer::add(0.1, function () use ($queue, $queueName) {
+                // 消费一次队列任务
+                $queue->popRun($queueName);
+            });
+        };
+
+        \Workerman\Worker::runAll();
+    }
+}