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 脚本进行原子化操作，这样多个进程之间争抢任务时就不会出现这种浪费了。