php使用redis的有序集合zset實現延遲隊列

延遲隊列就是個帶延遲功能的消息隊列，相對於普通隊列，它可以在指定時間消費掉消息。

延遲隊列的應用場景：

1、新用戶注冊，10分鍾后發送郵件或站內信。

2、用戶下單后，30分鍾未支付，訂單自動作廢。

 

我們通過redis的有序集合zset來實現簡單的延遲隊列，將消息數據序列化，作為zset的value，把消息處理時間作為score，每次通過zRangeByScore獲取一條消息進行處理。

<?php

class DelayQueue
{
    protected $prefix = 'delay_queue:';
    protected $redis = null;
    protected $key = '';

    public function __construct($queue, $config = [])
    {
        $this->key = $this->prefix . $queue;
        $this->redis = new Redis();
        $this->redis->connect($config['host'], $config['port'], $config['timeout']);
        $this->redis->auth($config['auth']);
    }

    public function delTask($value)
    {
        return $this->redis->zRem($this->key, $value);
    }

    public function getTask()
    {
        //獲取任務，以0和當前時間為區間，返回一條記錄
        return $this->redis->zRangeByScore($this->key, 0, time(), ['limit' => [0, 1]]);
    }

    public function addTask($name, $time, $data)
    {
        //添加任務，以時間作為score，對任務隊列按時間從小到大排序
        return $this->redis->zAdd(
            $this->key,
            $time,
            json_encode([
                'task_name' => $name,
                'task_time' => $time,
                'task_params' => $data,
            ], JSON_UNESCAPED_UNICODE)
        );
    }

    public function run()
    {
        //每次只取一條任務
        $task = $this->getTask();
        if (empty($task)) {
            return false;
        }

        $task = $task[0];
        //有並發的可能，這里通過zrem返回值判斷誰搶到該任務
        if ($this->delTask($task)) {
            $task = json_decode($task, true);

            //處理任務
            echo '任務：' . $task['task_name'] . ' 運行時間：' . date('Y-m-d H:i:s') . PHP_EOL;

            return true;
        }

        return false;
    }
}

$dq = new DelayQueue('close_order', [
    'host' => '127.0.0.1',
    'port' => 6379,
    'auth' => '',
    'timeout' => 60,
]);

$dq->addTask('close_order_111', time() + 30, ['order_id' => '111']);
$dq->addTask('close_order_222', time() + 60, ['order_id' => '222']);
$dq->addTask('close_order_333', time() + 90, ['order_id' => '333']);

然后，我們寫一個php腳本，用來處理隊列中的任務。

<?php

set_time_limit(0);

$dq = new DelayQueue('close_order', [
    'host' => '127.0.0.1',
    'port' => 6379,
    'auth' => '',
    'timeout' => 60,
]);

while (true) {
    $dq->run();
    usleep(100000);
}