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

121 阅读 0 评论 80 点赞

我是靠谱客的博主英俊小馒头，这篇文章主要介绍利用 rpush 和 blpop 实现 Redis 消息队列，现在分享给大家，希望可以做个参考。

背景

系统并发比较强的业务中，为了解决并发问题，这个时候就会用到消息队列，异步处理业务。本次就介绍用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即使定时任务请求进来，也会中止程序不执行业务代码。

1038×299 29.3 KB

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)

复制代码

1
2
3
*/5 * * * * php /app/WorkQueue.php cancel_order

复制代码

<?php
require_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();

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
&lt;?php
require_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-&gt;connect('127.0.0.1', 6379);
        //获取队列任务池列表
        while (true) {
            $specialTime = $allowAllExcuteTime - (time() - $startTime);//取得程序执行时间和290秒相比较，如果超过280秒直接break
            if($specialTime &lt;= 0){
                break;
            }
            //获取任务池的一个任务
            $content = $redis-&gt;BLPOP($redisQueueKey, $specialTime);//取出左边第一个值 如果没有则阻塞式监听；需要RPUSH入列
            $contentArr = json_decode($content,true);
            //执行任务
            if (is_array($content)) {
                $logicQueueObj-&gt;$method($contentArr);
            }
        }
    }

}
//实例化类执行消费队列数据
$workQueueObj = new WorkQueue();
$workQueueObj-&gt;consume();

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

复制代码

1
2
3
$redis-&gt;BLPOP($redisQueueKey, $specialTime)

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

复制代码

1
2
$logicQueueObj-&gt;$method($contentArr)

总结

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

最后

以上就是英俊小馒头最近收集整理的关于利用 rpush 和 blpop 实现 Redis 消息队列的全部内容，更多相关利用内容请搜索靠谱客的其他文章。

本图文内容来源于网友提供，作为学习参考使用，或来自网络收集整理，版权属于原作者所有。

本文分类：redis
浏览次数：121 次浏览
发布日期：2023-10-21 08:36:33
本文链接：https://www.kaopuke.com/article/k-p-k_13_u_23_o_22_fx_14__23_g3.html

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

最后

评论列表共有 0 条评论

发表评论取消回复

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

最后

微信扫一扫：分享

相关文章

评论列表共有 0 条评论

发表评论 取消回复

微信扫一扫：分享

发表评论取消回复