Redis异步队列与延时队列

什么是消息队列

MQ全称为Message Queue 消息队列（MQ）是一种应用程序对应用程序的通信方法。MQ是消费-生产者模型的一个典型的代表，一端往消息队列中不断写入消息，而另一端则可以读取队列中的消息。消息发布者只管把消息发布到 MQ 中而不用管谁来取，消息使用者只管从 MQ 中取消息而不管是谁发布的。

先说一下消息队列常见的使用场景吧，其实场景有很多，但是比较核心的有 3 个：解耦、异步、削峰

解耦

背景：A 系统发送数据到 BCD 三个系统，通过接口调用发送。如果 E 系统也要这个数据呢？那如果 C 系统现在不需要了呢？A 系统负责开发者几乎崩溃......在这个场景中，A 系统跟其它各种乱七八糟的系统严重耦合，A 系统产生一条比较关键的数据，很多系统都需要 A 系统将这个数据发送过来。A 系统要时时刻刻考虑 BCDE 四个系统如果挂了该咋办？要不要重发，要不要把消息存起来？头发都白了啊！

使用 MQ：A 系统产生一条数据，发送到 MQ 里面去，哪个系统需要数据自己去 MQ 里面消费。如果新系统需要数据，直接从 MQ 里消费即可；如果某个系统不需要这条数据了，就取消对 MQ 消息的消费即可。这样下来，A 系统压根儿不需要去考虑要给谁发送数据，不需要维护这个代码，也不需要考虑人家是否调用成功、失败超时等情况。

削峰

场景：每天 0:00 到 12:00，A 系统风平浪静，每秒并发请求数量就 50 个。结果每次一到 12:00 ~ 13:00 ，每秒并发请求数量突然会暴增到 5k+ 条。但是系统是直接基于 MySQL 的，大量的请求涌入 MySQL，每秒钟对 MySQL 执行约 5k 条 SQL。

一般的 MySQL，扛到每秒 2k 个请求就差不多了，如果每秒请求到 5k 的话，可能就直接把 MySQL 给打死了，导致系统崩溃，用户也就没法再使用系统了。

但是高峰期一过，到了下午的时候，就成了低峰期，可能也就 1w 的用户同时在网站上操作，每秒中的请求数量可能也就 50 个请求，对整个系统几乎没有任何的压力。

使用MQ：每秒 5k 个请求写入 MQ，A 系统每秒钟最多处理 2k 个请求，因为 MySQL 每秒钟最多处理 2k 个。A 系统从 MQ 中慢慢拉取请求，每秒钟就拉取 2k 个请求，不要超过自己每秒能处理的最大请求数量就 ok，这样下来，哪怕是高峰期的时候，A 系统也绝对不会挂掉。而 MQ 每秒钟 5k 个请求进来，就 2k 个请求出去，结果就导致在中午高峰期（1 个小时），可能有几十万甚至几百万的请求积压在 MQ 中。

异步

背景：如果A 系统接收一个请求，需要在自己本地写库，还需要在 BCD 三个系统写库，自己本地写库要 3ms，BCD 三个系统分别写库要 300ms、450ms、200ms。最终请求总延时是 3 + 300 + 450 + 200 = 953ms，接近 1s，用户感觉搞个什么东西，慢死了慢死了。用户通过浏览器发起请求，等待个 1s，这几乎是不可接受的。

使用MQ：一般互联网类的企业，对于用户直接的操作，一般要求是每个请求都必须在 200 ms 以内完成，对用户几乎是无感知的。如果使用 MQ，那么 A 系统连续发送 3 条消息到 MQ 队列中，假如耗时 5ms，A 系统从接受一个请求到返回响应给用户，总时长是 3 + 5 = 8ms，对于用户而言，其实感觉上就是点个按钮，8ms 以后就直接返回了，爽！网站做得真好，真快！

  • lpush和rpush入队列
  • lpop和rpop出队列
  • blpop和brpop阻塞式出队列

业务逻辑生产消息

<?php

$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

//发送消息
$redis->lPush($list, $value);

写一个php脚本用来处理队列中的任务。

<?php

$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
//消费消息
while (true) {
    try {
        $msg = $redis->rPop($list);
        if (!$msg) {
            sleep(1);
        }
        //业务处理
     
    } catch (Exception $e) {
        echo $e->getMessage();
    }
}

以上代码会有个问题，如果队列长时间是空的，pop就不会不断的循环，会导致redis的QPS升高，影响性能。所以我们使用sleep来解决，当没有消息的时候阻塞一段时间。但其实这样还会带来另一个问题，就是sleep会导致消息的处理延迟增加的机率。这个问题我们可以通过blpop/brpop来阻塞读取队列。blpop/brpop
在队列没有数据的时候，会立即进入休眠状态，一旦数据到来，则立刻醒过来。消息的延迟几乎为零。用blpop/brpop替代前面的lpop/rpop，就完美解决了上面的问题。

还有一个需要注意的点是我们需要是用try/catch来进行异常捕获，如果一直阻塞在那里，Redis服务器一般会主动断开掉空链接，来减少闲置资源的占用 可以使用 ping检查redis心跳

延时队列

定义：

延迟队列就是个带延迟功能的消息队列，相对于普通队列，它可以在指定时间消费掉消息。

应用场景：

  • 1、新用户注册，10分钟后发送邮件或站内信。
  • 2、用户下单后，30分钟未支付，订单自动作废。
  • 3、用户下单后，在抢单大厅订单进行补贴 10s 30s 90s 不同下单时长的订单进行不同的补贴策略。[我们公司目前遇到的场景]

实现方案

方式一：数据库实现

最简单的方式，定时扫表。例如对于订单支付失效要求比较高的，每2S扫表一次检查过期的订单进行主动关单操作。优点是简单，缺点是每分钟全局扫表，浪费资源，如果遇到表数据订单量即将过期的订单量很大，会造成关单延迟。。

方式二：redis的有序集合sort set

步骤：

  • 1.产生消息
    • 用时间戳作为score，使用zadd key score1 value1 命令生产消息
  • 2.读取消息
    • withscores limit 0 1消费消息最早的一条消息。
  • 3.消费消息并删除
    • 实现简单的延迟队列，将消息数据序列化，作为zset的value，把消息处理时间作为score，每次通过zRangeByScore获取一条消息进行处理。

redis 客户端命令
ZADD key score1 member1 [score2 member2]
ZRANGEBYSCORE key min max [WITHSCORES] [LIMIT]
ZREM key member [member ...]

rabbitMQ的死信队列

消息在以下情况下会变成死信息，会被DXL（Dead-Letter-Exchane）死信交换机投递到死信队列：

  • 1.消息被拒绝。
  • 2.消息未被及时消费或者消费了不ack，直接过期。
  • 3.队列达到最大长度。
  • 死信队列的实现：
    • 消息(设置ttl)--->交换机-->队列（消息过期）-->死信交换机-->死信队列-->消费
我们使用redis做延迟队列

整体思路：通过redis的有序集合zset来实现简单的延迟队列，将消息数据序列化，作为zset的value，把消息处理时间作为score，每次通过zRangeByScore获取一条消息进行处理。

<?php

class DelayQueue
{
    protected $prefix = 'delay_queue:';
    protected $redis = null;
    protected $key = '';
    private static $_instance = null;

    /**
     * 构造函数
     * DelayQueue constructor.
     * @param $queue
     * @param array $config
     */
    public function __construct($queue,$config = [])
    {
        $this->key = $this->prefix . $queue;
        $this->redis = new Redis();
        $this->redis->connect($config['host'], $config['port'], $config['timeout']);
        $this->redis->auth($config['auth']);
    }

    /**
     * Notes: 获取数据库句柄方法
     * User: jackin.chen
     * Date: 2020/7/20 下午3:55
     * function: getRedis
     * @return null|Redis
     */
    public function getRedis()
    {
        return $this->redis;
    }

    /**
     * Notes:这是获取当前类对象的唯一方式
     * User: jackin.chen
     * Date: 2020/7/20 下午3:55
     * function: getInstance
     * @param string $queue
     * @param array $config
     * @return DelayQueue|null
     * @static
     */
    public static function getInstance($queue, $config = [])
    {
        // 检查对象是否已经存在，不存在则实例化后保存到$instance属性
        if(!(self::$_instance instanceof self)){
            //内部实例化对象
            self::$_instance = new self($queue,$config);
        }
        return self::$_instance;
    }


    /**
     * Notes: 声明成私有方法，禁止克隆对象
     * User: jackin.chen
     * Date: 2020/7/20 下午3:56
     * function: __clone
     */
    private function __clone(){}

    /**
     * Notes: 声明成私有方法，禁止重建对象
     * User: jackin.chen
     * Date: 2020/7/20 下午3:56
     * function: __wakeup
     */
    private function __wakeup(){}


    /**
     * Notes: 删除任务列表
     * User: jackin.chen
     * Date: 2020/7/20 下午4:00
     * function: delTask
     * @param $value
     * @return int
     */
    public function delTask($value)
    {
        return $this->redis->zRem($this->key, $value);
    }

    /**
     * Notes: 获取一条任务
     * User: jackin.chen
     * Date: 2020/7/20 下午4:00
     * function: getTask
     * @return array
     */
    public function getTask()
    {
        //获取任务，以0和当前时间为区间，返回一条记录
        return $this->redis->zRangeByScore($this->key, 0, time(), ['limit' => [0, 1]]);
    }

    /**
     * Notes: 添加任务
     * User: jackin.chen
     * Date: 2020/7/20 下午4:00
     * function: addTask
     * @param $name
     * @param $time
     * @param $data
     * @return int
     */
    public function addTask($name, $time, $data)
    {
        //添加任务，以时间作为score，对任务队列按时间从小到大排序
        return $this->redis->zAdd(
            $this->key,
            $time,
            json_encode([
                'task_name' => $name,
                'task_time' => $time,
                'task_params' => $data,
            ], JSON_UNESCAPED_UNICODE)
        );
    }


    /**
     * Notes: 执行任务
     * User: jackin.chen
     * Date: 2020/7/20 下午4:14
     * function: run
     * @return bool
     */
    public function run()
    {
        //每次只取一条任务
        $task = $this->getTask();
        if (empty($task)) {
            return false;
        }
        $task = isset($task[0]) ? $task[0] : [];
        //有并发的可能，这里通过zrem返回值判断谁抢到该任务
        if ($task && $this->delTask($task)) {
            $task = json_decode($task, true);
            //处理任务
            echo '任务：' . $task['task_name'] . ' 运行时间：' . date('Y-m-d H:i:s') . PHP_EOL;
            return true;
        }
        return false;
    }
}


//生产使用
$Queue = DelayQueue::getInstance('payment_order',[
    'host' => '127.0.0.1',
    'port' => 6379,
    'auth' => '',
    'timeout' => 60,
]);

$Queue->addTask('payment_order_1', time() + 30, ['order_id' => '1']);
$Queue->addTask('payment_order_2', time() + 60, ['order_id' => '2']);
$Queue->addTask('payment_order_3', time() + 90, ['order_id' => '3']);

写一个php脚本，用来处理队列中的任务。

<?php

set_time_limit(0);
$Queue1 = DelayQueue::getInstance('payment_order',[
    'host' => '127.0.0.1',
    'port' => 6379,
    'auth' => '',
    'timeout' => 60,
]);

//处理任务
while (true) {
    $Queue1->run();
    usleep(100000);
}

这里又产生了一个问题，同一个任务可能会被多个进程取到之后再使用 zrem 进行争抢，那些没抢到的进程都是白取了一次任务，这是浪费。解决办法：将 zrangebyscore 和 zrem 使用 lua 脚本进行原子化操作，这样多个进程之间争抢任务时就不会出现这种浪费了。

©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 206,602评论 6赞 481
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 88,442评论 2赞 382
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 152,878评论 0赞 344
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 55,306评论 1赞 279
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 64,330评论 5赞 373
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 49,071评论 1赞 285
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 38,382评论 3赞 400
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 37,006评论 0赞 259
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 43,512评论 1赞 300
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,965评论 2赞 325
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,094评论 1赞 333
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,732评论 4赞 323
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 39,283评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,286评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,512评论 1赞 262
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 45,536评论 2赞 354
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,828评论 2赞 345