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 (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. }