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

本文实例讲述了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 (empty($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']);
63.  

然后，我们写一个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. }
13.  

更多关于PHP相关内容感兴趣的读者可查看本站专题：《php+redis数据库程序设计技巧总结》、《php面向对象程序设计入门教程》、《PHP基本语法入门教程》、《PHP数组(Array)操作技巧大全》、《php字符串(string)用法总结》、《php+mysql数据库操作入门教程》及《php常见数据库操作技巧汇总》

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