EasySwoole 基于Redis组件实现延迟队列

介绍

在用户要支付订单的时候，如果超过30分钟未支付，会把订单关掉。当然我们可以做一个定时任务，每个一段时间来扫描未支付的订单，如果该订单超过支付时间就关闭，但是在数据量小的时候并没有什么大的问题，但是数据量一大轮训数据库的方式就会变得特别耗资源。当面对千万级、上亿级数据量时，本身写入的IO就比较高，导致长时间查询或者根本就查不出来，更别说分库分表以后了。

使用延迟队列解决的痛点无非是

  1. 实现了数据延迟
  2. 数据摊开(仔细去理解)

知识点

  1. redis有序集合
  2. EasySwoole Redis协程客户端

案例

生成订单id —-> 扔到延迟队列 —-> 延迟队列消费进程不停获取30分钟前的订单满足条件的订单 —-> 处理订单

直接上代码

EasySwooleEvent.php 注册redis连接池、注册延迟队列消费进程

  1. <?php
  2. namespace EasySwoole\EasySwoole;
  4. use App\Process\Consumer;
  5. use EasySwoole\EasySwoole\Swoole\EventRegister;
  6. use EasySwoole\EasySwoole\AbstractInterface\Event;
  7. use EasySwoole\Http\Request;
  8. use EasySwoole\Http\Response;
  9. use EasySwoole\Pool\Manager;
  10. use EasySwoole\Redis\Config\RedisConfig;
  11. use App\RedisPool\RedisPool;
  12. use EasySwoole\Pool\Config;
  13. class EasySwooleEvent implements Event
  14. {
  16.     public static function initialize()
  17.     {
  18.         // TODO: Implement initialize() method.
  19.         date_default_timezone_set('Asia/Shanghai');
  20.     }
  22.     public static function mainServerCreate(EventRegister $register)
  23.     {
  25.         //TODO:: 注册redis连接池
  26.         $config = new Config();
  27.         $redisConfig1 = new RedisConfig([
  28.             'host'      => '127.0.0.1',
  29.             'port'      => '6379'
  30.         ]);
  32.         // 这里的redis连接池看文档配吧
  33.         Manager::getInstance()->register(new RedisPool($config,$redisConfig1),'redis');
  35.         //TODO:: 延迟队列消费进程
  36.         $processConfig= new \EasySwoole\Component\Process\Config();
  37.         $processConfig->setProcessName('testProcess');
  39.         \EasySwoole\Component\Process\Manager::getInstance()->addProcess(new Consumer($processConfig));
  40.     }
  42.     public static function onRequest(Request $request, Response $response): bool
  43.     {
  44.         // TODO: Implement onRequest() method.
  46.         return true;
  47.     }
  49.     public static function afterRequest(Request $request, Response $response): void
  50.     {
  51.         // TODO: Implement afterAction() method.
  52.     }
  53. }

扔到延迟队列

  1. <?php
  2. namespace App\HttpController;
  4. use EasySwoole\Http\AbstractInterface\Controller;
  5. use EasySwoole\Pool\Manager;
  7. class Index extends Controller
  8. {
  10.     function index()
  11.     {
  12.         /** @var $redis \EasySwoole\Redis\Redis*/
  13.         $orderId = date('YmdHis', time());
  14.         $redis = Manager::getInstance()->get('redis')->getObj();
  15.         $res = $redis->zAdd('delay_queue_test1', time(), $orderId);
  16.         if ($res) {
  17.             $this->writeJson(200, '订单添加成功:'.$orderId);
  18.         }
  19.     }
  21. }

延迟队列消费进程

  1. <?php
  2. namespace App\Process;
  4. use EasySwoole\Component\Process\AbstractProcess;
  5. use EasySwoole\Pool\Manager;
  6. use Swoole\Coroutine;
  8. class Consumer extends AbstractProcess {
  9.     protected function run($arg)
  10.     {
  11.         go(function (){
  12.             while (true) {
  14.                 //TODO:: 拿到redis
  15.                 /** @var $redis \EasySwoole\Redis\Redis*/
  16.                 $redis = Manager::getInstance()->get('redis')->defer();
  18.                 //TODO:: 从有序集合中拿到三秒(模拟30分钟)以前的订单
  19.                 $orderIds = $redis->zRangeByScore('delay_queue_test1', 0, time()-3, ['withscores' => TRUE]);
  21.                 if (empty($orderIds)) {
  22.                     Coroutine::sleep(1);
  23.                     continue;
  24.                 }
  26.                 //TODO::拿出后立马删除
  27.                 $redis->zRem('delay_queue_test1', ...$orderIds);
  29.                 foreach ($orderIds as $orderId)
  30.                 {
  31.                     var_dump($orderId);
  33.                     //TODO::判断此订单30分钟后，是否仍未完成，做相应处理
  34.                 }
  35.             }
  36.         });
  37.     }
  39. }

测试

请求index/index 投递订单到延迟队列

  1. ➜  ~ curl 127.0.0.1:9501/index/index
  2. {"code":200,"result":"订单添加成功:20200422004046","msg":null}%

等3s看终端是否输出

  1. ➜  easyswoole php easyswoole start
  2.   ______                          _____                              _
  3.  |  ____|                        / ____|                            | |
  4.  | |__      __ _   ___   _   _  | (___   __      __   ___     ___   | |   ___
  5.  |  __|    / _` | / __| | | | |  \___ \  \ \ /\ / /  / _ \   / _ \  | |  / _ \
  6.  | |____  | (_| | \__ \ | |_| |  ____) |  \ V  V /  | (_) | | (_) | | | |  __/
  7.  |______|  \__,_| |___/  \__, | |_____/    \_/\_/    \___/   \___/  |_|  \___|
  8.                           __/ |
  9.                          |___/
  10. main server                   SWOOLE_WEB
  11. listen address                0.0.0.0
  12. listen port                   9501
  13. ip@en0                        192.168.43.57
  14. worker_num                    8
  15. reload_async                  true
  16. max_wait_time                 3
  17. pid_file                      /Users/xx/sites/easyswoole/Temp/pid.pid
  18. log_file                      /Users/xx/sites/easyswoole/Log/swoole.log
  19. user                          xx
  20. daemonize                     false
  21. swoole version                4.4.15
  22. php version                   7.2.18
  23. easy swoole                   3.3.7
  24. develop/produce               develop
  25. temp dir                      /Users/xx/sites/easyswoole/Temp
  26. log dir                       /Users/xx/sites/easyswoole/Log
  28. string(14) "20200422004046"

总结

这只是一个思路，大家可以根据实际业务做不同调整