likes
comments
collection
share

浅谈Redis消息队列设计

作者站长头像
站长
· 阅读数 21

1. 背景

为了实现系统解耦、流量削峰、功能异步化，企业系统开发人员都会考虑使用消息队列。市面上以高吞吐量著称的Kafka、以消息绝对可靠而被广泛应用在金融或订单系统中的RabbitMQ，这些开源的消息中间件都深受开发者喜欢。今天笔者来和大家谈谈redis实现的消息队列，它主要应用在系统内部，对消息可靠性没有特别大要求的一些业务场景中;它的特点就是轻量、高效、简单。

2. redis实现消息队列的数据结构和命令

redis为实现消息队列提供了多种数据结构和相关命令，使得系统开发中实现一个简单的消息队列非常容易。通过list数据结构，可以实现简单的消息队列；通过zset数据结构可以实现延时消息队列；为了实现消息的多播机制，redis还提供了 publish、subscribe命令来实现消息的订阅与发布；redis5.0新增了stream数据结构，实现了更加专业的消息队列。下边我们先来分别介绍一下它们。

2.1 list实现简单队列

redis提供的list数据结构本身就是一个FIFO(先进先出)队列，业务上通过list实现消息队列可以通过生产者在队尾添加消息，消费者从队首取出消息来实现，如下图描述：

浅谈Redis消息队列设计

消息入队可以通过lpush或rpush将消息体以O(1)的时间复杂度添加到队列中，可以通过启动一个消费者每隔一段时间(例如1s)将消息rpop/lpop出来进行消费，消费者php代码如下：

<?php
$redis = new Redis();

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

while($message = $redis->rpop('queue_list_test')) {
    var_dump($message);
    
    //处理完逻辑后，休眠1s
    sleep(1)
}

消费者代码中的sleep是很有必要的，假如说没有sleep，那么当队列queue_list_test为空时，while循环就会一直与redis服务器建立连接，直接拉高客户端服务器的CPU，也会使redis服务器的流量突增。

在消费者的这种处理方式下，消息被消费会存在一定的延时，为此redis也提供了brpop/blpop命令，它们是rpop/lpop的阻塞版本，也就是说当list中没有数据时,命令会阻塞等待，直到超时或者队列中有消息时结束，改造后的消费者php代码如下：

<?php
$redis = new Redis();

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

while($message = $redis->brpop('queue_list_test',3600)) {
    var_dump($message);
}

上面的例子是，获取key名是queue_list_test的list的数据，没有数据的话会阻塞3600秒。但由于php的redis扩展是基于php的socket方式实现的，如果php本身配置了socket read超时时间，那么超时会报错退出。 socket默认超时配置项在php.ini中的默认配置是：

default_socket_timeout = 60

好的实践方式是通过php的ini_set函数对socket超时时间在进程中重新设置，解决超时报错消费者的php代码如下：

<?php
ini_set('default_socket_timeout', -1); //不超时

$redis = new Redis();

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

while($message = $redis->brpop('queue_list_test',0)) {
    var_dump($message);
}

将brpop的超时时间设置为0意味着如果队列中没有消息的到来，脚本就无限等待。这样就解决了消费者延时消费问题。

2.2 zset实现延时消费队列

redis中的zset和set很像，都是字符串的集合，都不允许重复的成员出现在一个set中。他们的区别在于有序集合中每一个成员都有一个分数（score）与之关联，redis 正是通过分数来对集合里的成员进行从小到大的排序。试想如果将消息添加到zset结构中，并将消息要被消费的时间戳设置为对应的score，是不是就组成了一个时间序列的的消费队列了呢？如下图所示：

浅谈Redis消息队列设计

延时消息队列就是这么实现的：

  • 生产者通过zadd命令将消息添加到时间序列的消费队列中，score为消息需要被消费的时间戳(将来的一个时间点)。
  • 消费者通过zrangebyscore命令取出到目前时间点为止，需要被消费的消息。

浅谈Redis消息队列设计

2.3 PubSub实现消息多播

我们前边说到的使用list实现的简单队列和使用zset实现的延时消费队列都不支持消息多播，如果多个不同的消费者组想要消费队列中的消息，只能将消费者组的逻辑串联起来进行连续消费了：

浅谈Redis消息队列设计

redis单独使用PubSub模块来支持消息多播，也就是PublisherSubscriber(发布者/订阅者模式)。消息多播允许生产者只生产一次消息，由中间件负责将消息复制到多个消息队列，每个消息队列由相应的消费组进行消费：

浅谈Redis消息队列设计

关于PubSub的相关命令和使用方式这里就不做介绍了，读者可以自行查阅资料学习，需要说明的是PubSub的设计有一个致命缺点：不支持消息的持久化

PubSub的生产者传递过来一个消息，Redis会直接找到相应的消费者传递过去。如果一个消费者也没有，那么消息就会被丢弃。比如刚开始有三个消费者，一个消费者突然挂掉了，生产者会继续发送消息，另外两个消费者可以持续收到消息，但是当挂掉的消费者重连上的时候，在断连期间生产者发送的消息，对于这个消费者来说就彻底丢失了。如果Redis 宕机或者重启，相当于一个消费者也没有了，所有的消息都会被直接丢弃，PubSub的消息是不会持久化的。

由于PubSub的这些缺点，在消息队列的领域几乎找不到它的应用场景！

Redis的作者还单独开启了一个项目Disque专门用来做消息队列，不过一直没有成熟，一直处于Beta版本。到了2018年6月，Redis5.0 新增了 Stream数据结构，这个功能给Redis带来了持久化的消息队列，从此PubSub作为消息队列的功能消失在人海，Disque永远也发不出release版本了。

2.4 强大的支持消息多播的可持久化消息队列 —— stream

Redis5.0 最大的特性就是新增了一个数据结构stream做消息队列，它极大的借鉴了Kafka的设计，从此Redis就有了真正意义上专业的消息队列了。结构如下图所示：

浅谈Redis消息队列设计

stream结构本身就是一个时间唯一序列的数组队列，类似与我们上边说到的zset做的时间序列消息队列。每一个消息都有一个消息ID，消息ID的形式是timestampInMillis-sequence,例如1607226700982-5，它表示当前的消息是在毫秒时间戳1607226700982时产生，并且是该毫秒内的第5条消息，消息ID可以由服务器自动生成，也可以由客户端自己指定(由此可以看出Stream支持延时任务的功能)，但是形式必须是“整数-整数”，而且后边加入的消息的ID必须要大于前边的消息ID。

每个Stream都有唯一的名称，它就是Redis的key,在首次使用命令xadd指令时自动创建。从图中可以看出，每个Stream可以挂多个消费组(Consumer Group)，每个消费组会有游标last_deliverred_id在Stream数组队列上向前移动，表示当前消费组已经消费到哪条消息了，每个消费组在stream内名字也是唯一的，消费组需要通过xgroup_create单独创建并指定从某个消息ID开始消费，这个消息ID对应的就是消费组内部的last_deliverred_id变量值。

消费组的状态都是独立的，相互不受影响，每一个Stream消息会投递到每一个消费组中去，每个消费组可以挂载多个消费者(Consumer)，这些消费者之间是竞争关系，任意一个消费者读取了消息都会使游标last_deliverred_id向前移动，每个消费者在组内也都有唯一的名字。

消费者内部的状态变量pending_ids用来记录当前哪些消息被客户端读取，但是还没有被ack的消息。这个pending_ids变量用来确保客户端至少消费了消息一次，不会在网络传输过程中因为丢失了而未被处理。

关于stream的设计不是本节的重点，先介绍这么多，但是非常建议读者自行查阅资料学习一下stream，它是未来redis做消息队列的最佳方案！

3. 传统redis消息队列设计

因为redis5.0刚推出不久，由于历史原因企业系统内部的redis队列还大都采用了传统的消息队列设计，也就是采用list + zset结构封装实现，在传统redis队列设计方案里，很多解决问题的思路非常值得学习，本节我们来谈一谈传统redis消息队列面临的问题以及解决思路是什么。

3.1 整合“延时”队列与“即时”队列

我们前边说过，传统的redis消息队列使用list做即时消费队列，通过lpop/rpop或者blpop/brpop命令取消费数据；使用zset做延时消费队列，通过zrangebyscore命令取出消费数据。但是对于消费者而言，并不在乎通过哪种方式从队列中取出消费数据。一个好的实践是封装一个方法提供给消费者，让它只管每次从队列中取出数据进行消费就可以了，通常的实现方式是当消费者每次通过这个方法从队列中取数据时，系统首先从zset延时队列中将已经到期的任务迁移到list即时队列中，再通过lpop/rpop从队列中取出数据返回给消费者，示意图如下：

浅谈Redis消息队列设计

需要注意的是，我们封装的从队列中取消息的系统方法再将延时队列中到期的消息迁移到即时队列中的过程并不是像lpop/rpop一样是原子性的，为了保证系统的并发安全，需要配合lua脚本来完成这项工作，示例代码如下：

/**
     * Get the Lua script to migrate expired jobs back onto the queue.
     *
     * KEYS[1] - The queue we are removing jobs from, for example: queues:foo:reserved
     * KEYS[2] - The queue we are moving jobs to, for example: queues:foo
     * ARGV[1] - The current UNIX timestamp
     *
     * @return string
     */
    public static function migrateExpiredJobs()
    {
        return <<<'LUA'
-- Get all of the jobs with an expired "score"...
local val = redis.call('zrangebyscore', KEYS[1], '-inf', ARGV[1])

-- If we have values in the array, we will remove them from the first queue
-- and add them onto the destination queue in chunks of 100, which moves
-- all of the appropriate jobs onto the destination queue very safely.
if(next(val) ~= nil) then
    redis.call('zremrangebyrank', KEYS[1], 0, #val - 1)

    for i = 1, #val, 100 do
        redis.call('rpush', KEYS[2], unpack(val, i, math.min(i+99, #val)))
    end
end

return val
LUA;
    }

上述lua脚本将zrangebyscore、zremrangebyrank、rpush三条redis指令打包在一起，保证了消息迁移的原子性，这样我们的系统就可以起安全的启用多个消费者了。

对于生产者来说封装一个系统方法也是很有必要的，得益于zadd、lpush/rpush指令都是原子性的，我们只需要判断消息是否需要延期执行，然后调用指令将延时消息推向延时队列、将即时消息推向即时队列就好了:

浅谈Redis消息队列设计

3.2 实现消息的“ack机制”与“超时重试”

我们前边提到的Kafka和RabbitMQ这些专业的消息队列，都有ack消息确认机制。ack机制保证了队列中的消息至少被消费了一次，通常队列会提供手动ack、自动ack两种机制，对消息可靠性要求比较高的消费者应该采用手动ack机制。redis传统消息队列如果要实现ack机制，首先想到的就是使用一个set数据结构，将消费者取出来的消息添加到set集合中，当消费者处理完成消息后，再将消息从set集合中删除掉，完成ack。

前边我们使用了zset做延时队列，zset也是set，并且提供了一个排名属性值score，如果我们使用zset来完成我们队列系统的ack，还可以利用score完成另外一项很有用的功能：超时重试。如下图所示：

浅谈Redis消息队列设计

增加了“超时重试”和“ack机制”的redis消息系统比之前复杂了很多。消费者通过系统方法从消息系统中取任务时，消息系统首先将延时队列、ack队列中已经到期的和没有ack的消息迁移到即时队列中(为了保证系统的并发安全，这两个迁移消息的过程要lua脚本配合完成);再将消息取出来交给消费者之前，消息系统需要将消息添加到ack队列中，并设置超时重试时间retry_after_time,也就是如果消息过了retry_after_time的时间依然没有进行ack,那么消息就会被重新迁移到即时队列再次消费；正常的处理应该是消费者取到消息以后，通过处理程序消费完成了消息数据，进行ack(也就是从ack队列中删除消息)。

3.3 消费者的“优雅重启”

虽然ack能够保证队列中的消息至少能够被消费一次，但是业务场景中对于大任务的处理往往比较耗时，我们并不想在消息被消费到一半的时候，强行终止消费者，而是渴望在当前消息被消费完成的时候，再重启消费者，我们称之为消费者的优雅重启。

熟悉nginx的读者应该清楚，通过nginx -s reload命令能够保证nginx在对外提供服务的同时，重新加载最新的配置文件并重启子进程。其工作原理就是通过主进程监听信号分发给子进程，在子进程处理完当前的服务后才会退出，然后master重新启动新的子进程。所以通过该命令重启nginx的前后，主进程号并没有发生变化。

对于队列的消费者也是一样的，可以通过监听信号的方式来实现“优雅重启”，不同编程语言提供的处理信号函数也不同，下边还是以php为例，php通过pcntl拓展来实现信号处理，下边是主要代码实现逻辑:

class Worker
{
    public $shouldQuit = false;
    
    public function daemon(string $queueName = '')
    {
        $this->listenForSignals();

        while (!$this->shouldQuit) {
            //这里是业务的处理逻辑
        }
    }
    
    //监听信号
    protected function listenForSignals()
    {
        if ($this->supportsAsyncSignals()) {
            pcntl_async_signals(true);

            pcntl_signal(SIGTERM, function () {
                $this->shouldQuit = true;
            });
            pcntl_signal(SIGUSR2, function () {
                $this->shouldQuit = true;
            });
            pcntl_signal(SIGCONT, function () {
                $this->shouldQuit = true;
            });
        }
    }
}

通过上述代码，当我们使用kill命令杀死消费者进程号的时候，消费者程序就会接收到信号，并将下次循环条件设置为false，处理完当前任务后再退出，实现“优雅重启”。不过在业务开发当中，往往使用supervisor来管理消费者，supervisor也可以通过信号与消费者交互，推荐给读者一篇文章，感兴趣的可以相互学习一下：supervisor在PHP项目中的使用

3.4 失败消息的处理

通过ack机制保证消息至少被消费一次也好，通过消费者监听信号实现优雅重启也罢，总会存在消费失败的消息，这在业务当中是避免不了的。对于redis实现的消息队列来讲，我们可以给消费者设置一个失败处理函数，当消息失败的时候，由消费者程序决定怎样处理它(通常会把错误消息添加到数据库，后续人工排查)。下边是php代码大致实现逻辑:

class Worker
{
    public function daemon(string $queueName = '')
    {
        while (!$this->shouldQuit) {
            try {
                //这里是业务的处理逻辑
            } catch (\Exception $e) {
                //消息失败了，判断是否需要重试，若是不需要重试，若有失败处理函数，记录下来失败信息
                $this->failedHandler($e, $jobInfo);
                die($e->getMessage()) . PHP_EOL;
            } catch (\Throwable $e) {
                die($e->getMessage()) . PHP_EOL;       //系统错误, 退出
            }
        }
    }
    
    public function failedHandler(\Exception $e, array $jobInfo)
    {
        if (class_exists($jobInfo['commandName'])) {
            $consumerInstance = new $jobInfo['commandName']($jobInfo['data']);
            if (method_exists($consumerInstance, $this->failedHandlerMethod)) {
                //调用失败处理函数
                $failedMethod = $this->failedHandlerMethod;
                $consumerInstance->$failedMethod($e);
            }
        }
    }
}

4.小结

实现一个功能完备、健壮高可用的redis消息队列并没有那么简单，需要考虑很多的因素。出于业务需要，redis本身也经历了一段历史的发展，才有了现在stream这个相对可靠的消息队列解决方案。传统的使用list + zset实现的redis消息队列在企业中服务了很多年，并至今也没有完全被取代是有一定道理的。本文分析了传统消息队列实现面临的问题，针对这些问题，采取的解决方案是什么。虽然传统的redis消息队列会被stream取代，但是通过学习本节我们理解了redis实现消息队列的思路，领悟了复杂系统实现的一些方法论，这个过程还是很快乐的。