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

本文实例讲述了php使用redis的有序集合zset实现延迟队列。分享给大家供大家参考，具体如下：

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

延迟队列的应用场景：

1、新用户注册，10分钟后发送邮件或站内信。

2、用户下单后，30分钟未支付，订单自动作废。

我们通过redis的有序集合zset来实现简单的延迟队列，将消息数据序列化，作为zset的value，把消息处理时间作为score，每次通过zRangeByScore获取一条消息进行处理。

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62

<?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脚本，用来处理队列中的任务。

?

1 2 3 4 5 6 7 8 9 10 11 12

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

希望本文所述对大家PHP程序设计有所帮助。

原文链接：https://www.cnblogs.com/jkko123/p/10614696.html