Redis有序集合实现延迟队列

  1. Redis有序集合和集合一样也是string类型元素的集合,且不允许重复的成员;

  2. 不同的是每个元素都会关联一个double类型的分数,redis正是通过分数来为集合中的成员进行从小到大的排序;

  3. 有序集合的成员是唯一的,但是分数(score)却可以重复;

  4. 集合是通过哈希表实现的,每个集合可存储40多个亿成员;


<?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;
    }
}

生产任务

<?php
//生产任务
include_once './DelayQueue.php';
$dq = new DelayQueue('close_order', [
    'host' => '127.0.0.1',
    'port' => 6379,
    'auth' => '',
    'timeout' => 60,
]);

$dq->addTask('close_order', time() + 30, ['order_id' => rand(1,9999)]);
$dq->addTask('close_order', time() + 40, ['order_id' => rand(1,9999)]);
$dq->addTask('close_order', time() + 50, ['order_id' => rand(1,9999)]);

消费任务

<?php
//消费任务
include_once './DelayQueue.php';
set_time_limit(0);

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

while (true) {
    $dq->run();
    sleep(3);
}


MyAnswer博客
请先登录后发表评论
  • 最新评论
  • 总共0条评论