page contents

php使用redis的有序集合zset实现延迟队列

延迟队列就是个带延迟功能的消息队列,相对于普通队列,它可以在指定时间消费掉消息。

attachments-2020-06-efjKdtLQ5ed5b44931511.png

延迟队列就是个带延迟功能的消息队列,相对于普通队列,它可以在指定时间消费掉消息。

延迟队列的应用场景:

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



attachments-2020-06-CMkduagP5ed5b398e2dae.jpg

  • 发表于 2020-06-02 10:07
  • 阅读 ( 435 )
  • 分类:PHP开发

你可能感兴趣的文章

相关问题

0 条评论

请先 登录 后评论
Pack
Pack

1135 篇文章

作家榜 »

  1. 轩辕小不懂 2403 文章
  2. 小柒 1316 文章
  3. Pack 1135 文章
  4. Nen 576 文章
  5. 王昭君 209 文章
  6. 文双 71 文章
  7. 小威 64 文章
  8. Cara 36 文章