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

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

延迟队列的应用场景：

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

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

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

1. <?php 
2. class DelayQueue 
3. { 
4.   protected $prefix = 'delay_queue:'; 
5.   protected $redis = null; 
6.   protected $key = ''; 
7.   public function __construct($queue, $config = []) 
8.   { 
9.     $this->key = $this->prefix . $queue; 
10.     $this->redis = new Redis(); 
11.     $this->redis->connect($config['host'], $config['port'], $config['timeout']); 
12.     $this->redis->auth($config['auth']); 
13.   } 
14.   public function delTask($value) 
15.   { 
16.     return $this->redis->zRem($this->key, $value); 
17.   } 
18.   public function getTask() 
19.   { 
20.     //获取任务，以0和当前时间为区间，返回一条记录 
21.     return $this->redis->zRangeByScore($this->key, 0, time(), ['limit' => [0, 1]]); 
22.   } 
23.   public function addTask($name, $time, $data) 
24.   { 
25.     //添加任务，以时间作为score，对任务队列按时间从小到大排序 
26.     return $this->redis->zAdd( 
27.       $this->key, 
28.       $time, 
29.       json_encode([ 
30.         'task_name' => $name, 
31.         'task_time' => $time, 
32.         'task_params' => $data, 
33.       ], JSON_UNESCAPED_UNICODE) 
34.     ); 
35.   } 
36.   public function run() 
37.   { 
38.     //每次只取一条任务 
39.     $task = $this->getTask(); 
40.     if (emptyempty($task)) { 
41.       return false; 
42.     } 
43.     $task = $task[0]; 
44.     //有并发的可能，这里通过zrem返回值判断谁抢到该任务 
45.     if ($this->delTask($task)) { 
46.       $task = json_decode($task, true); 
47.       //处理任务 
48.       echo '任务：' . $task['task_name'] . ' 运行时间：' . date('Y-m-d H:i:s') . PHP_EOL; 
49.       return true; 
50.     } 
51.     return false; 
52.   } 
53. } 
54. $dq = new DelayQueue('close_order', [ 
55.   'host' => '127.0.0.1', 
56.   'port' => 6379, 
57.   'auth' => '', 
58.   'timeout' => 60, 
59. ]); 
60. $dq->addTask('close_order_111', time() + 30, ['order_id' => '111']); 
61. $dq->addTask('close_order_222', time() + 60, ['order_id' => '222']); 
62. $dq->addTask('close_order_333', time() + 90, ['order_id' => '333']); 

然后，我们写一个php脚本，用来处理队列中的任务。

1. <?php 
2. set_time_limit(0); 
3. $dq = new DelayQueue('close_order', [ 
4.   'host' => '127.0.0.1', 
5.   'port' => 6379, 
6.   'auth' => '', 
7.   'timeout' => 60, 
8. ]); 
9. while (true) { 
10.   $dq->run(); 
11.   usleep(100000); 
12. }

Tags: redis有序集合 zset

分享到：