写点什么

利用 rpush 和 blpop 实现 Redis 消息队列

  • 2022 年 9 月 28 日
    北京
  • 本文字数:1943 字

    阅读完需:约 6 分钟

背景

系统并发比较强的业务中,为了解决并发问题,这个时候就会用到消息队列,异步处理业务。本次就介绍用 rpush 结合 blpop 实现 Redis 的消息队列方案之一。我们此次以取消订单业务为例子(取消订单无业务代码,重点介绍消息队列实现)。

方案介绍

利用 redis 命令 rpush 往 redis 列表尾部插入数据,之后利用 blpop 阻塞式从列表中获取列表做左边数据(先进先出),阻塞式获取数据是写在后端代码中,也就是消费队列数据的代码中。消费函数 consume(app/WorkQueue.php->consume())由定时任务每 5 分钟执行一次,在 consume()代码中程序执行到 290 秒会自动结束,因为 5 分钟后定时任务下一个进程会再次调用 app/WorkQueue.php->consume()。

实现

目录总览

配置

app/config/config.php 配置文件,quueDriver 为存储的每个队列驱动,cancel_order 为取消订单驱动,cancel_order 下 class 值对应 app/logic/ Order.php, app/WorkQueue.php->consume()会自动去根据配置到 app/logic/下加载这个类;配置 method 为 class 对应的类中执行业务逻辑的方法;open 代表是否开启此队列,true 开启 false 关闭,如果为 false 即使定时任务请求进来,也会中止程序不执行业务代码。


Redis 相关

**redis key 前缀:**我们使用 QUEUE_

**redis 队列值得数据方式:**采用 json 串;在 app/WorkQueue.php->consume()中消费队列数据时会转为数组。

入列举例:



执行消费队列数据

消费函数 consume(app/WorkQueue.php->consume())由定时任务每 5 分钟执行一次,在 consume()代码中程序执行到 290 秒会自动结束,因为 5 分钟后定时任务下一个进程会再次调用 app/WorkQueue.php->consume()

定时任务调用消费方法方式

php /app/WorkQueue.php 配置中驱动名称(app/config/config.php 比如(cancel_order)

*/5 * * * * php /app/WorkQueue.php cancel_order<?phprequire_once './config/config.php';
class WorkQueue{ /** * 执行消息队列消费 */ public function consume() { $startTime = time();//程序开始时间 $allowAllExcuteTime = 290;//允许程序总执行时间,因为定时任务会5分钟执行一次,所以到290秒提前结束,因为下一个进程要进来 $configArr = include_once './config/config.php';//配置 $quueDriverConfig = $configArr['quueDriver'];//队列配置 $driverName = $argv[1];//获取定时任务命令传递的驱动名称 比如 php /home/www/app/WorkQueue.php cancel_order 那么此处就是cancelOrder $redisQueueKey = "QUEUE_{$driverName}";//redis队列的前缀为QUEUE_ //获取队列驱动的logic任务类 $logicName = $quueDriverConfig[$driverName]['class']; $method = $quueDriverConfig[$driverName]['method']; //判断此队列驱动是否开启 如果未开启则不消费队列 if($quueDriverConfig[$driverName]['open'] !== true){ return; } //加载相应消息队列类 require_once "./logic/queue/{$logicName}.php"; $logicQueueObj = new $logicName();//实例化 $redis = new redis();//redis对象 $redis->connect('127.0.0.1', 6379); //获取队列任务池列表 while (true) { $specialTime = $allowAllExcuteTime - (time() - $startTime);//取得程序执行时间和290秒相比较,如果超过280秒直接break if($specialTime <= 0){ break; } //获取任务池的一个任务 $content = $redis->BLPOP($redisQueueKey, $specialTime);//取出左边第一个值 如果没有则阻塞式监听;需要RPUSH入列 $contentArr = json_decode($content,true); //执行任务 if (is_array($content)) { $logicQueueObj->$method($contentArr); } } }
}//实例化类执行消费队列数据$workQueueObj = new WorkQueue();$workQueueObj->consume();
复制代码

BLPOP 阻塞式监听队列 key $redisQueueKey,可以指定阻塞监听时间 $specialTime,如果没有队列数据,会再阻塞监听 $specialTime 秒

$redis->BLPOP($redisQueueKey, $specialTime)
复制代码

执行业务代码,$contentArr 为队列数据

$logicQueueObj->$method($contentArr)
复制代码

总结

队列实现方式有很多种,只要选择适合自己业务场景的即可!在实际业务处理中还会涉及到比如从队列取出数据,如果代码业务层处理业务失败,是否会将此队列数据重新压入队列的问题,这里只是稍微提一下。

更多学习资料戳下方!!!

https://qrcode.ceba.ceshiren.com/link?name=article&project_id=qrcode&from=infoQ&timestamp=1662366626&author=xueqi

用户头像

社区:ceshiren.com 2022.08.29 加入

微信公众号:霍格沃兹测试开发 提供性能测试、自动化测试、测试开发等资料、实事更新一线互联网大厂测试岗位内推需求,共享测试行业动态及资讯,更可零距离接触众多业内大佬

评论

发布
暂无评论
利用 rpush 和 blpop 实现 Redis 消息队列_测试_测吧(北京)科技有限公司_InfoQ写作社区