Redis 延迟任务

业务场景

1. 订单下单后,30分钟内未支付,则取消订单(类似订单自动退款|自动收货等都一样)
2. 类似支付宝异步通知,在未收到回复前,按此频率 0|30|60|150 通知

---

方案一：定时任务

创建定时任务,每隔一分钟查询数据库,处理超时订单

1. 适用小项目,可参考 数据库动态配置定时任务、秒级定时任务
2. 缺点：
   • 时效性差,有延迟
   • 效率差

方案二：redis psubscribe

订阅过期事件

配置redis.conf

# K    键空间通知,以__keyspace@<db>__为前缀  
# E    键事件通知,以__keysevent@<db>__为前缀  
# g    del, expipre, rename 等类型无关的通用命令的通知...  
# $    String命令  
# l    List命令  
# s    Set命令  
# h    Hash命令  
# z    有序集合命令  
# x    过期事件(每次key过期时生成)
# e    驱逐事件(（)当key在内存满了被清除时生成)
# A    g$lshzxe的别名,因此"AKE"意味着所有的事件
notify-keyspace-events "Ex"

代码实现

$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
$redis->auth('123456') or die("redis password verification failed");
$redis->ping() == '+PONG' or die("redis connection is not available, ping={$redis->ping()}");

$orderList = [];
for ($i = 1; $i < 10; $i++) {
    $orderKey = 'order_' . $i;
    $rand = mt_rand(1, 30);
    $redis->setex($orderKey, $rand, $i);
    $orderList[$rand] = $orderKey;
}
ksort($orderList);
var_export($orderList);

$redis->setOption(Redis::OPT_READ_TIMEOUT, -1);
$redis->psubscribe(['__keyevent@0__:expired'], function ($redis, $pattern, $chan, $msg) {
    echo PHP_EOL . date('Y-m-d H:i:s') . " {$msg}";
});

方案三：redis psubscribe + queue

例：A、B、C 三台服务器
A 订阅过期事件,将过期的订单 rpush 到 queue 里, B 和 C 通过 blpop 从队列获取订单处理

A服务器订阅过期事件

function getRedis(bool $reset = false)
{
    if (!$reset) {
        static $redis;
    }

    $redis = new Redis();
    $redis->connect('127.0.0.1', 6379);
    $redis->auth('123456') or die("redis password verification failed");
    $redis->ping() == '+PONG' or die("redis connection is not available, ping={$redis->ping()}");

    return $redis;
}

$redis = getRedis(true);

$orderList = [];
for ($i = 1; $i < 10; $i++) {
    $orderKey = 'order_' . $i;
    $rand = mt_rand(1, 30);
    $redis->setex($orderKey, $rand, $i);
    $orderList[$rand] = $orderKey;
}

$redis->setOption(Redis::OPT_READ_TIMEOUT, -1);
// 一旦客户端进入订阅状态,客户端就只可接受订阅相关的命令SUBSCRIBE、PSUBSCRIBE、UNSUBSCRIBE、PUNSUBSCRIBE,其他命令一律失效
// http://www.redis.cn/commands/subscribe.html
$redis->psubscribe(['__keyevent@0__:expired'], function ($redis, $pattern, $chan, $msg) {
    echo PHP_EOL . date('Y-m-d H:i:s') . " {$msg}";

    $r = getRedis();
    $listKey = 'order_list';
    $r->rPush($listKey, $msg);
});

B、C服务器处理队列

$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
$redis->auth('123456') or die("redis password verification failed");
$redis->ping() == '+PONG' or die("redis connection is not available, ping={$redis->ping()}");

$redis->setOption(Redis::OPT_READ_TIMEOUT, -1);
while (1) {
    $data = $redis->blPop('order_list', 0);
    echo "pid：" . posix_getpid() . ", order_id：{$data['1']}" . PHP_EOL;
}

方案四：延迟任务 DelayedTask

项目地址 Delayed-task
基于 Redis 的 Sorted Set 实现延迟任务
环境要求：

• Redis
• PHP >=7.0.0
• PHP 需安装 pctnl、posix、redis 模块

整体结构

1. Task Pool 存放所有任务的信息(Redis Hashes)
2. Delay Bucket 存放所有需要延迟的任务ID,以时间戳排序(Redis Sorted Set),以取模的方式将延迟任务分配至多个Bucket
3. 创建多个进程,负责扫描各自的Delay Bucket(取模方式),获取delayTime(延迟执行时间)小于等于当前时间的任务,从Delay Bucket中删除并放入Ready Queue(Redis Lists),按任务优先级存入不同的队列
4. 创建多个进程,负责扫描各自的Ready Queue(先处理优先级高的队列)

流程

1. TaskClient->add() 客户端添加任务
    1.1 初始化任务
        1.1.1 topic 任务类型
        1.1.2 id 任务ID, topic
        1.1.3 url、call 设置一个即可,表示通知URL或call指定方法
        1.1.4 callback 任务执行成功后的回调函数
        1.1.5 任务调用规则
            1.1.5.1 intervalTime = 3, persistent = true 每隔3秒调用一次
            1.1.5.2 intervalTime = 10, persistent = false 10秒后调用一次就结束任务
            1.1.5.3 intervalTime = '0,30,60', persistent = true 0|30|60 秒后调用任务,第三次调用任务后,之后每隔60秒调用一次
            1.1.5.4 intervalTime = '0,30,60', persistent = false 0|30|60 秒后调用任务,第三次调用后结束任务
        1.1.6 priority 任务优先级,不同优先级处理的任务数量也不同,目前支持设置优先级1-3,数字越高优先级越高,优先处理的任务也就越多
        1.1.7 params 额外参数
        1.1.8 status 任务状态,默认为延迟中
    1.2 hSet 保存任务至 Task Pool
    1.3 zAdd 保存任务ID,任务延迟执行时间作为排序值,bucketKey 通过 crc32(topic) % bucketCount 取模,得到任务ID所要保存在哪个Bucket中,便于多进程时的进程一对一监控
2. TaskServer->scanBucket() 服务端扫描延迟任务
    2.1 创建子进程,通过取模获取当前进程所需要监控的Bucket
    2.2 通过 zRangeByScore 获取delayTime小于等于当前时间的任务
        2.2.1 任务状态为延迟中,则将任务从Bucket中删除,并根据任务优先级添加至不同的队列(这里采用分布式锁,保证原子性)
        2.2.2 任务状态不为延迟中,将任务从Bucket中删除
3. TaskServer->scanQueue() 服务端扫描队列
    3.1 创建子进程,先处理优先级较高的队列
    3.2 通过 lPop 获取任务,再次检测一下任务状态及delayTime
    3.3 handleTask 处理任务
        3.3.1 OK 调用 callback
        3.3.2 FAIL 不处理
        3.3.3 DELAY 重新计算任务下次执行时间,将任务放入Bucket

参考：

1. 有赞延迟队列设计
2. chenlinzhong/php-delayqueue
3. 基于redis的延迟消息队列设计

注意：涉及延迟任务的处理的只有 TaskServer.php 的 scanBucket、scanQueue、handleTask 方法, 其它方法的作用参考 任务调度

启动、停止服务 server.php

// 启动：php server.php
// 停止：php server.php stop
<?php
use Sevming\DelayedTask\TaskServer;

require_once 'vendor/autoload.php';

try {
    $server = new TaskServer('127.0.0.1', 6379, '123456');
    $server->debug = true;
    $server->runAll();
} catch (Exception $e) {
    exit($e->getMessage());
}

添加、删除任务 client.php

<?php
use Sevming\DelayedTask\TaskClient;

require_once 'vendor/autoload.php';

try {
    $client = new TaskClient('127.0.0.1', 6379, '123456');

    // 1. 每隔3秒调用一次Order类的return方法
    $client->add([
        'topic' => 'order:return',
        'id' => 1,
        'call' => ['Order', 'return'],
        'intervalTime' => 3,
        'callback' => ['Order', 'success'],
    ]);

    // 2. 10秒后调用Order类的receipt方法,仅调用1次
    $client->add([
        'topic' => 'order:receipt',
        'id' => 2,
        'call' => ['Order', 'receipt'],
        'intervalTime' => 10,
        'persistent' => false,
    ]);

    // 3. 设置优先级,每隔 0|30|60秒 调用一次Order类的timeout,调用三次后结束调用
    $client->add([
        'topic' => 'order:timeout',
        'id' => 3,
        'call' => ['Order', 'timeout'],
        'priority' => 2,
        'intervalTime' => '0,30,60',
        'persistent' => false,
    ]);

    // 4. 每隔 0|10|20秒 调用一次Order类的refund,调用三次后,间隔时间为20秒调用一次
    $client->add([
        'topic' => 'order:refund',
        'id' => 4,
        'call' => ['Order', 'refund'],
        'intervalTime' => '0,10,20',
    ]);

    // 5. 每隔 0|15|30秒 请求一次url,请求三次后,不再请求
    $client->add([
        'topic' => 'order:notify',
        'id' => 5,
        'url' => '192.168.50.77:8080',
        'params' => [
            'content' => 'hello'
        ],
        'intervalTime' => '0,15,30',
        'persistent' => false,
    ]);

    // 6. 删除任务,根据添加任务时的topic和id
    //$client->del('order:return', 1);
} catch (Exception $e) {
    exit($e->getMessage());
}

Order.php

<?php

class Order
{
    public function return(array $task)
    {
        echo __FUNCTION__ . ",id={$task['id']},createTime={$task['createTime']},count={$task['rule']['count']},runTime="  . date('Y-m-d H:i:s') . PHP_EOL;
        // 返回success,则任务结束
        return 'success';
    }

    public function receipt(array $task)
    {
        echo __FUNCTION__ . ",id={$task['id']},createTime={$task['createTime']},count={$task['rule']['count']},runTime=" . date('Y-m-d H:i:s') . PHP_EOL;
    }

    public function timeout(array $task)
    {
        echo __FUNCTION__ . ",id={$task['id']},createTime={$task['createTime']},count={$task['rule']['count']},runTime=" . date('Y-m-d H:i:s') . PHP_EOL;
    }

    public function refund(array $task)
    {
        echo __FUNCTION__ . ",id={$task['id']},createTime={$task['createTime']},count={$task['rule']['count']},runTime=" . date('Y-m-d H:i:s') . PHP_EOL;
    }

    public function success(array $task)
    {
        var_dump($task);
        echo "任务处理成功\n";
    }
}

TaskClient.php

<?php

namespace Sevming\DelayedTask;

use Exception, RedisException;

class TaskClient extends DelayedTask
{
    /**
     * Add task.
     *
     * @param array $params
     *
     * @throws Exception|RedisException
     */
    public function add(array $params)
    {
        $task = $this->initTask($params);
        $redis = $this->getRedisInstance();

        $taskKey = $this->getTaskKey($task['topic'], $task['id']);
        // Save task to task pool.
        $redis->hSet($this->getTaskPoolKey(), $taskKey, json_encode($task));
        // Save the task ID, with the task delay execution time as the sort value
        $redis->zAdd($this->getBucketKey($task['topic']), $task['delayTime'], $taskKey);
    }

    /**
     * Del task.
     *
     * @param string $topic
     * @param mixed  $id
     * @param bool   $softDelete
     *
     * @return void
     * @throws Exception|RedisException
     */
    public function del(string $topic, $id, bool $softDelete = true)
    {
        $taskKey = $this->getTaskKey($topic, $id);
        $redis = $this->getRedisInstance();

        $taskPoolKey = $this->getTaskPoolKey();
        if ($task = $redis->hGet($taskPoolKey, $taskKey)) {
            if (!$softDelete) {
                $redis->hDel($taskPoolKey, $taskKey);
                return;
            }

            $task = json_decode($task, true);
            if ($task['status'] !== static::TASK_STATUS_DELETED) {
                $task['status'] = static::TASK_STATUS_DELETED;
                $redis->hSet($taskPoolKey, $taskKey, json_encode($task));
            }
        }
    }

    /**
     * Init task.
     *
     * @param array $params
     *
     * @return array
     * @throws Exception
     */
    protected function initTask(array $params)
    {
        if (empty($params['topic'])) {
            throw new Exception('topic can not be empty.');
        }

        if (empty($params['id'])) {
            throw new Exception('id can not be empty.');
        }

        if (empty($params['url']) && empty($params['call'])) {
            throw new Exception('url or call can not be empty.');
        }

        if (!empty($params['call'])) {
            list ($class, $action) = $params['call'];
            if (!class_exists($class) || !method_exists($class, $action)) {
                throw new Exception("call class {$class} not exists or action {$action} not exists.");
            }
        }

        if (!empty($params['callback'])) {
            list ($class, $action) = $params['callback'];
            if (!class_exists($class) || !method_exists($class, $action)) {
                throw new Exception("callback class {$class} not exists or action {$action} not exists.");
            }
        }

        $intervalTimeArray = [0];
        if (!empty($params['intervalTime'])) {
            $intervalTimeArray = explode(',', $params['intervalTime']);
            foreach ($intervalTimeArray as $interval) {
                if (!ctype_digit((string)$interval)) {
                    throw new Exception("intervalTime incorrect format");
                }
            }
        }

        $priority = static::QUEUE_PRIORITY_1;
        if (isset($params['priority']) && in_array($params['priority'], array_keys($this->queueTaskPriority))) {
            $priority = $params['priority'];
        }

        $time = time();
        $task = [
            'topic' => $params['topic'],
            'id' => $params['id'],
            'priority' => $priority,
            'status' => static::TASK_STATUS_DELAY,
            'url' => $params['url'] ?? '',
            'method' => $params['method'] ?? 'GET',
            'call' => $params['call'] ?? [],
            'params' => $params['params'] ?? [],
            'callback' => $params['callback'] ?? [],
            'rule' => [
                'interval' => $intervalTimeArray,
                'count' => 0,
                'persistent' => isset($params['persistent']) ? (bool)($params['persistent']) : true,
            ],
            'delayTime' => $params['delayTime'] ?? ($time + $intervalTimeArray[0]),
            'lastRunTime' => '',
            'createTime' => date('Y-m-d H:i:s', $time),
        ];

        return $task;
    }
}

TaskServer.php

<?php

namespace Sevming\DelayedTask;

use Exception;
use Sevming\DelayedTask\Traits\HttpTrait;

class TaskServer extends DelayedTask
{
    use HttpTrait;

    /**
     * Log file.
     *
     * @var string
     */
    public $logFile = '';

    /**
     * The file to store master process PID.
     *
     * @var string
     */
    public $pidFile = '';

    /**
     * Debug mode.
     *
     * @var bool
     */
    public $debug = false;

    /**
     * The PID of master process.
     *
     * @var int
     */
    protected $masterPid = 0;

    /**
     * The PID Map of bucket process.
     *
     * @var array
     */
    protected $bucketPidMap = [];

    /**
     * The PID Map of queue process.
     *
     * @var array
     */
    protected $queuePidMap = [];

    /**
     * Run all.
     *
     * @throws Exception
     */
    public function runAll()
    {
        $this->checkEnv();
        $this->init();
        $this->parseCommand();
        $this->daemonize();
        $this->installSignal();
        $this->saveMasterPid();
        $this->scanBucket();
        $this->scanQueue();
        $this->resetStd();
        $this->monitorProcess();
    }

    /**
     * Check env.
     */
    protected function checkEnv()
    {
        // Only for cli
        if (php_sapi_name() != 'cli') {
            exit("only run in command line mode.\n");
        }

        // Only for linux
        if (strpos(strtolower(PHP_OS), 'win') === 0) {
            exit("Not support windows.\n");
        }

        // Need pcntl extension
        if (!extension_loaded('pcntl')) {
            exit("Please install pcntl extension.\n");
        }

        // Need posix extension
        if (!extension_loaded('posix')) {
            exit("Please install posix extension.\n");
        }

        // Need redis extension
        if (!extension_loaded('redis')) {
            exit("Please install redis extension.\n");
        }
    }

    /**
     * Init.
     */
    protected function init()
    {
        $backtrace = debug_backtrace();
        $startFile = $backtrace[count($backtrace) - 1]['file'];

        if (empty($this->pidFile)) {
            $this->pidFile = __DIR__ . '/' . str_replace('/', '_', $startFile) . '.pid';
        }

        if (empty($this->logFile)) {
            $this->logFile = __DIR__ . '/TaskServer.log';
        }

        $this->logFile = (string)$this->logFile;
        if (!is_file($this->logFile)) {
            touch($this->logFile);
            chmod($this->logFile, 0622);
        }

        $this->setProcessTitle('master');
    }

    /**
     * Parse command.
     */
    protected function parseCommand()
    {
        global $argv;

        $availableCommands = ['start', 'stop'];
        if (!isset($argv[1]) || !in_array($argv[1], $availableCommands)) {
            if (isset($argv[1])) {
                exit("Unknown command: {$argv[1]}\n");
            }
        }

        $command = isset($argv[1]) ? trim($argv[1]) : '';
        // Avoid repeated run
        $masterPid = is_file($this->pidFile) ? file_get_contents($this->pidFile) : 0;
        $masterIsAlive = $masterPid && posix_kill($masterPid, 0) && posix_getpid() != $masterPid;
        if ($masterIsAlive && $command === 'start') {
            $this->log('already running');
            exit;
        }

        switch ($command) {
            case 'stop':
                $this->log('process is stopping ...');
                // Send stop signal to master process.
                $masterPid && posix_kill($masterPid, SIGINT);
                $timeout = 5;
                $startTime = time();
                // Check master process is still alive?
                while (1) {
                    $masterIsAlive = $masterPid && posix_kill($masterPid, 0);
                    if ($masterIsAlive) {
                        if (time() - $startTime >= $timeout) {
                            $this->log('process stop fail.');
                            exit;
                        }

                        usleep(10000);
                        continue;
                    }

                    $this->log('process stop success.');
                    exit(0);
                }

                break;
        }
    }

    /**
     * Run as deamon mode.
     *
     * @throws Exception
     */
    protected function daemonize()
    {
        if ($this->debug) {
            return;
        }

        $pid = pcntl_fork();
        if ($pid === -1) {
            throw new Exception('fork fail');
        } elseif ($pid > 0) {
            exit(0);
        }

        if (posix_setsid() === -1) {
            throw new Exception('setsid fail');
        }

        // Fork again avoid SVR4 system regain the control of terminal.
        $pid = pcntl_fork();
        if ($pid === -1) {
            throw new Exception('fork fail');
        } elseif ($pid > 0) {
            exit(0);
        }

        umask(0);
        $this->setProcessTitle('master');
    }

    /**
     * Install signal.
     */
    protected function installSignal()
    {
        // stop
        pcntl_signal(SIGINT, [$this, 'signalHandler'], false);
        // ignore
        pcntl_signal(SIGPIPE, SIG_IGN, false);
    }

    /**
     * Signal handler.
     *
     * @param $signal
     */
    protected function signalHandler($signal)
    {
        switch ($signal) {
            case SIGINT:
                $this->stopAll();
                break;
        }
    }

    /**
     * Stop all.
     */
    public function stopAll()
    {
        // For master process
        if ($this->masterPid === posix_getpid()) {
            $this->log("master process {$this->masterPid} stopping ...");

            $childPidMap = array_merge($this->bucketPidMap, $this->queuePidMap);
            foreach ($childPidMap as $pid) {
                posix_kill($pid, SIGKILL);
                $this->log("child process {$pid} stopping success.");
            }
        } // For child processes
        else {
            exit(0);
        }
    }

    /**
     * Save master pid.
     *
     * @throws Exception
     */
    protected function saveMasterPid()
    {
        $this->masterPid = posix_getpid();
        if (file_put_contents($this->pidFile, $this->masterPid) === false) {
            throw new Exception('can not save pid to ' . $this->pidFile);
        }
    }

    /**
     * Reset Std.
     */
    protected function resetStd()
    {
        if ($this->debug) {
            return;
        }

        fclose(STDIN);
        fclose(STDOUT);
        fclose(STDERR);
        $stdoutFile = '/dev/null';
        fopen($stdoutFile, 'r');
        fopen($stdoutFile, 'a');
        fopen($stdoutFile, 'a');
    }

    /**
     * Scan Bucket in real time to handle delayed tasks.
     *
     * @throws Exception
     */
    protected function scanBucket()
    {
        for ($i = 0; $i < $this->bucketCount; $i++) {
            $pid = pcntl_fork();
            if ($pid > 0) {
                $this->bucketPidMap[$pid] = $pid;
            } elseif ($pid === -1) {
                throw new Exception('bucket fork fail');
            } else {
                try {
                    // Re-establish a redis connection for each process.
                    static::$redis = null;
                    $redis = $this->getRedisInstance();
                    $taskPoolKey = $this->getTaskPoolKey();
                    $bucketKey = $this->getBucketKey('', $i);
                    $this->setProcessTitle('bucket');

                    while (1) {
                        pcntl_signal_dispatch();
                        // Get a delay task with a delay time less than or equal to the current time
                        $taskKeyArray = $redis->zRangeByScore($bucketKey, '-inf', time(), [
                            'withscores' => true,
                            'limit' => [0, $this->bucketRangeLimit],
                        ]);

                        if (!empty($taskKeyArray)) {
                            foreach ($taskKeyArray as $taskKey => $score) {
                                if ($task = $redis->hGet($taskPoolKey, $taskKey)) {
                                    // Decode task.
                                    $task = json_decode($task, true);
                                    if ($task['status'] === static::TASK_STATUS_DELAY) {
                                        // Add lock.
                                        $lockData = $this->lock($taskKey);

                                        if ($lockData) {
                                            if ($redis->zRem($bucketKey, $taskKey)) {
                                                // Add tasks to different queues based on task priority
                                                if (!$redis->rPush($this->getQueueKey($task['priority']), $taskKey)) {
                                                    $this->log("bucket rPush fail,taskKey={$taskKey}");
                                                }
                                            }

                                            // Unlock.
                                            $this->unLock($lockData);
                                        }
                                    } else {
                                        $redis->zRem($bucketKey, $taskKey);
                                    }
                                }
                            }
                        }
                    }
                } catch (Exception $e) {
                    $this->log($e->getMessage());
                    exit(250);
                }
            }
        }
    }

    /**
     * Scan Queue in real time to handle tasks.
     *
     * @throws Exception
     */
    protected function scanQueue()
    {
        for ($i = 0; $i < $this->queueCount; $i++) {
            $pid = pcntl_fork();
            if ($pid > 0) {
                $this->queuePidMap[$pid] = $pid;
            } elseif ($pid === -1) {
                throw new Exception('queue fork fail');
            } else {
                try {
                    // Re-establish a redis connection for each process.
                    static::$redis = null;
                    $redis = $this->getRedisInstance();
                    $taskPoolKey = $this->getTaskPoolKey();
                    $this->setProcessTitle('queue');
                    $queueTaskPriority = array_reverse($this->queueTaskPriority, true);

                    while (1) {
                        pcntl_signal_dispatch();

                        foreach ($queueTaskPriority as $priority => $nums) {
                            $queueKey = $this->getQueueKey($priority);
                            // Process $nums tasks each time, avoiding the inability to handle higher priority tasks
                            while ($nums--) {
                                $taskKey = $redis->lPop($queueKey);
                                if (empty($taskKey)) {
                                    break;
                                }

                                if ($task = $redis->hGet($taskPoolKey, $taskKey)) {
                                    $task = json_decode($task, true);
                                    if ($task['status'] === static::TASK_STATUS_DELAY && $task['delayTime'] <= time()) {
                                        $newTask = $this->handleTask($task);
                                        $redis->hSet($taskPoolKey, $taskKey, json_encode($newTask));
                                        if ($newTask['status'] === static::TASK_STATUS_DELAY) {
                                            // Save the task ID, with the task delay execution time as the sort value
                                            $redis->zAdd($this->getBucketKey($newTask['topic']), $newTask['delayTime'], $taskKey);
                                        }
                                    }
                                }
                            }
                        }
                    }
                } catch (Exception $e) {
                    $this->log($e->getMessage());
                    exit(250);
                }
            }
        }
    }

    /**
     * Handle task.
     *
     * @param array $task
     *
     * @return array
     */
    protected function handleTask(array $task)
    {
        if (!empty($task['url'])) {
            $result = static::request($task['url'], $task['params'], $task['method']);
        } else {
            list ($class, $action) = $task['call'];
            if (!class_exists($class) || !method_exists($class, $action)) {
                $result = 'fail';
                $this->log("call class {$class} not exists or action {$action} not exists.");
            } else {
                $result = (new $class())->$action($task);
            }
        }

        $task['lastRunTime'] = date('Y-m-d H:i:s');
        $task['rule']['count']++;
        $result = strtolower((string)$result);

        if ($result === 'success') {
            $task['status'] = static::TASK_STATUS_OK;
            if (!empty($task['callback'])) {
                list ($class, $action) = $task['callback'];
                if (!class_exists($class) || !method_exists($class, $action)) {
                    $this->log("callback class {$class} not exists or action {$action} not exists.");
                } else {
                    $result = (new $class())->$action($task);
                }
            }
        } elseif ($result === 'fail') {
            $task['status'] = static::TASK_STATUS_FAIL;
        } else {
            $index = $task['rule']['count'];
            if (!isset($task['rule']['interval'][$index]) && !$task['rule']['persistent']) {
                $task['status'] = static::TASK_STATUS_FAIL;
            } else {
                $intervalTime = $task['rule']['interval'][$index] ?? end($task['rule']['interval']);
                $delayTime = $task['delayTime'] + $intervalTime;
                $nowTime = time();
                if ($delayTime < $nowTime) {
                    $delayTime = $nowTime + $intervalTime;
                }

                $task['delayTime'] = $delayTime;
            }
        }

        return $task;
    }

    /**
     * Monitor process.
     */
    protected function monitorProcess()
    {
        while (1) {
            pcntl_signal_dispatch();
            $status = 0;
            $pid = pcntl_wait($status, WUNTRACED);
            pcntl_signal_dispatch();

            if ($pid > 0) {
                if ($status !== 0) {
                    $this->log("child process {$pid} exit with status {$status}");
                }

                unset($this->bucketPidMap[$pid], $this->queuePidMap[$pid]);
            }

            if (empty($this->bucketPidMap) && empty($this->queuePidMap)) {
                @unlink($this->pidFile);
                exit(0);
            }
        }
    }

    /**
     * Set process title.
     *
     * @param string $title
     */
    protected function setProcessTitle(string $title)
    {
        if (function_exists('cli_set_process_title')) {
            cli_set_process_title($this->prefix . $title);
        }
    }

    /**
     * Log.
     *
     * @param string $msg
     */
    protected function log(string $msg)
    {
        $msg = date('Y-m-d H:i:s') . ' [' . static::class . '] ' . $msg . PHP_EOL;
        file_put_contents((string)$this->logFile, $msg, FILE_APPEND | LOCK_EX);
    }
}

DelayedTask.php

<?php

namespace Sevming\DelayedTask;

use RedisException, Redis;

class DelayedTask
{
    /**
     * @var string
     */
    public $prefix = 'dt_';

    /**
     * Get the number of delayed tasks each time.
     *
     * @var int
     */
    public $bucketRangeLimit = 1;

    /**
     * Number of bucket processes.(Number of storage bucket.)
     *
     * @var int
     */
    public $bucketCount = 4;

    /**
     * Number of queue processes.
     *
     * @var int
     */
    public $queueCount = 4;

    /**
     * Redis config.
     *
     * @var array
     */
    protected $config = [];

    /**
     * Redis instance.
     *
     * @var null
     */
    protected static $redis = null;

    /**
     * Number of task processing corresponding to queue priority.
     *
     * @var array
     */
    protected $queueTaskPriority = [
        self::QUEUE_PRIORITY_1 => 2,
        self::QUEUE_PRIORITY_2 => 3,
        self::QUEUE_PRIORITY_3 => 5
    ];

    /**
     * @var int
     */
    protected const QUEUE_PRIORITY_1 = 1;
    protected const QUEUE_PRIORITY_2 = 2;
    protected const QUEUE_PRIORITY_3 = 3;

    /**
     * Task status delay.
     *
     * @var int
     */
    protected const TASK_STATUS_DELAY = 1;

    /**
     * Task status ok.
     *
     * @var int
     */
    protected const TASK_STATUS_OK = 2;

    /**
     * Task status fail.
     *
     * @var int
     */
    protected const TASK_STATUS_FAIL = 3;

    /**
     * Task status deleted.
     *
     * @var int
     */
    protected const TASK_STATUS_DELETED = 4;

    /**
     * Constructor.
     *
     * @param string $host
     * @param int    $port
     * @param string $auth
     */
    public function __construct(string $host = '127.0.0.1', int $port = 6379, string $auth = '')
    {
        $this->config = [
            'host' => $host,
            'port' => $port,
            'auth' => $auth,
        ];
    }

    /**
     * Get redis.
     *
     * @return Redis
     * @throws RedisException
     */
    protected function getRedisInstance()
    {
        if (!isset(static::$redis)) {
            $redis = new \Redis();
            $redis->connect($this->config['host'], $this->config['port']);

            if (!$redis->auth($this->config['auth'])) {
                throw new RedisException("redis password verification failed, auth={$this->config['auth']}");
            }

            if ($redis->ping() !== '+PONG') {
                throw new RedisException("redis connection is not available, ping={$redis->ping()}");
            }

            static::$redis = $redis;
        }

        return static::$redis;
    }

    /**
     * Get task pool key.
     *
     * @return string
     */
    protected function getTaskPoolKey()
    {
        return $this->prefix . 'task_pool';
    }

    /**
     * Get task key.
     *
     * @param string $topic
     * @param mixed  $id
     *
     * @return string
     */
    protected function getTaskKey(string $topic, $id)
    {
        return $topic . ':' . $id;
    }

    /**
     * Get bucket key.
     *
     * @param string $topic
     * @param bool   $flag
     *
     * @return string
     */
    protected function getBucketKey(string $topic = '', $flag = false)
    {
        $key = crc32($topic) % $this->bucketCount;
        return $this->prefix . 'task_bucket:' . ($flag !== false ? $flag : $key);
    }

    /**
     * Get queue key.
     *
     * @param int $priority
     *
     * @return string
     */
    protected function getQueueKey(int $priority)
    {
        return $this->prefix . 'task_queue:' . $priority;
    }

    /**
     * Lock.
     *
     * @param string $taskKey
     *
     * @return array|bool
     */
    protected function lock(string $taskKey)
    {
        $redis = $this->getRedisInstance();
        $lockKey = $this->prefix . 'task_lock:' . $taskKey;
        $lockValue = md5(uniqid(md5(microtime(true)), true));
        $status = $redis->set($lockKey, $lockValue, ['nx', 'px' => 3000]);
        if ($status) {
            return [$lockKey, $lockValue];
        }

        return false;
    }

    /**
     * Unlock.
     *
     * @param array $lockData
     *
     * @return bool
     */
    protected function unlock(array $lockData)
    {
        $redis = $this->getRedisInstance();
        $script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";

        return $redis->eval($script, $lockData, 1);
    }
}

HttpTrait.php

<?php

namespace Sevming\DelayedTask\Traits;

trait HttpTrait
{
    /**
     * Request.
     *
     * @param string $url
     * @param array  $params
     * @param string $method
     * @param int    $timeout
     * @param array  $headers
     * @param array  $userAgent
     * @return mixed
     */
    public static function request(string $url, array $params = [], string $method = 'GET', int $timeout = 1, array $headers = [], array $userAgent = [])
    {
        if (!preg_match('/^(http|https)/is', $url)) {
            $url = 'http://' . $url;
        }

        $ch = curl_init();
        $method = strtoupper($method);
        switch ($method) {
            case 'GET':
                $url .= '?' . http_build_query($params);
                break;
            case 'POST':
                curl_setopt($ch, CURLOPT_POST, true);
                curl_setopt($ch, CURLOPT_POSTFIELDS, $params);
                break;
        }

        curl_setopt($ch, CURLOPT_URL, $url);
        curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
        curl_setopt($ch, CURLOPT_SSL_VERIFYPEER, false);
        curl_setopt($ch, CURLOPT_SSL_VERIFYHOST, false);

        if ($timeout > 0 && $timeout < 1) {
            $timeout = (int)($timeout * 1000);
            curl_setopt($ch, CURLOPT_NOSIGNAL, true);
            curl_setopt($ch, CURLOPT_TIMEOUT_MS, $timeout);
        } else {
            $timeout = (int)$timeout;
            curl_setopt($ch, CURLOPT_TIMEOUT, $timeout);
        }

        if (!empty($headers)) {
            curl_setopt($ch, CURLOPT_HTTPHEADER, $headers);
        }

        if (!empty($userAgent)) {
            curl_setopt($ch, CURLOPT_USERAGENT, $userAgent);
        }

        $result = curl_exec($ch);
        $errno = curl_errno($ch);
        curl_close($ch);
        if (!$errno) {
            return $result;
        }

        return false;
    }
}