Redis专题-队列

首先，想一想 Redis 适合做消息队列吗？

1、消息队列的消息存取需求是什么？redis中的解决方案是什么？

无非就是下面这几点：

0、数据可以顺序读取
1、支持阻塞等待拉取消息
2、支持发布/订阅模式
3、重新消费
4、消息不丢失
5、消息可堆积

那我们来看看redis怎么满足这些需求

1.1、基于 List 的消息队列解决方案

1.1.1、数据保证顺序

List 本身就是按先进先出的顺序对数据进行存取的，底层的实现就是一个「链表」，在头部和尾部操作元素，时间复杂度都是 O(1)，这意味着它非常符合消息队列的模型。

生产者使用 LPUSH 发布消息：

127.0.0.1:6379> LPUSH mq 5
(integer) 1
127.0.0.1:6379> LPUSH mq 3
(integer) 2

消费者使用 RPOP 拉取消息:

127.0.0.1:6379> RPOP mq
5
127.0.0.1:6379> RPOP mq
3

当队列中已经没有消息了，消费者在执行 RPOP 时，会返回 NULL。

127.0.0.1:6379> RPOP mq
(nil) 

消费者读取数据时，有一个潜在的性能风险点：

生产者写入数据时，List 并不会主动通知消费者有新消息写入。
如果消费者想要及时处理消息，需要在程序中不停地调用 RPOP 命令。
如果有新消息写入，RPOP 命令就会返回结果，否则，RPOP 命令返回空值，再继续循环。

// 伪代码
while (true)
{var msg = redis.rpop("mq")if(msg == null)continue;handle(msg)
}

上述代码中如果队列为空，消费者依旧会频繁拉取消息，这会造成「CPU 空转」，不仅浪费 CPU 资源，还会对 Redis 造成压力。

我们处理一下，当队列为空时，我们可以「休眠」一会，再去尝试拉取消息。

// 伪代码
while (true)
{var msg = redis.rpop("mq")if(msg == null){Thread.Sleep(2000);continue;}handle(msg)
}

「CPU 空转」解决了，但是有新的问题发生了：当消费者在休眠等待时有新消息，那么消费者处理新消息就会存在「延迟」。

那如何做，既能及时处理新消息，还能避免 CPU 空转呢？

1.1.2、支持阻塞等待拉取消息

为了解决这个问题，Redis 提供了 BRPOP 命令。BRPOP 命令也称为阻塞式读取，客户端在没有读到队列数据时，自动阻塞，直到有新的数据写入队列，再开始读取新数据。和消费者程序自己不停地调用 RPOP 命令相比，这种方式能节省 CPU 开销。(这里的 B 指的是阻塞（Block）。)

使用 BRPOP 这种阻塞式方式拉取消息时，还支持传入一个「超时时间」，如果设置为 0，则表示不设置超时，直到有新消息才返回，否则会在指定的超时时间后返回 NULL

// 伪代码
while (true)
{// 没消息阻塞等待，0表示不设置超时时间var msg = redis.brpop("mq",0)if(msg == null)continue;handle(msg)
}

注意：如果设置的超时时间太长，这个连接太久没有活跃过，可能会被 Redis Server 判定为无效连接，之后 Redis Server 会强制把这个客户端踢下线。所以，采用这种方案，客户端要有重连机制。

1.1.3、发布/订阅模式

不支持。

1.1.4、重新消费

不支持。

但是在业务使用唯一ID等方式实现，消费ID后做判断是否处理过，使对于同一条消息处理结果都是一致的，保证幂等性。

1.1.5、消息不丢失

仅消费端不丢失。

List 类型提供了 BRPOPLPUSH 命令，这个命令的作用是让消费者程序从一个 List 中读取消息，同时，Redis 会把这个消息再插入到另一个 List（可以叫作备份 List）留存。

如果消费者程序读了消息但没能正常处理，等它重启后，就可以从备份 List 中重新读取消息并进行处理了。

1.1.6、消息堆积

不可堆积。

如果消费较慢，List 中的消息越积越多，redis内存压力会越来越大。
而且List本身也不支持消费组，不能使用多个消费端消费。

1.1.7、小结

需求	LIST
数据保证顺序	支持。使用LPUSH/RPOP
支持阻塞等待拉取消息	支持。使用BRPOP
支持发布 / 订阅模式	不支持
重复消费	不支持。但是可以自行实现全局唯一ID
消息不丢失	不完全。消费端算是不丢失，BRPOPLPUSH
消息堆积	不支持。内存持续增长

简单的业务场景，可以使用list。
但如果想要有多个生产者和消费者，那么可以继续往下看。

1.2、基于 Pub/Sub 的消息队列解决方案

Redis 专门是针对「发布/订阅」( PUBLISH / SUBSCRIBE) 这种队列模型设计的。

可以解决重复消费问题，可以多组生产者、消费者场景。

使用 Pub/Sub 这种方案，既支持阻塞式拉取消息，还很好地满足了多组消费者，消费同一批数据的业务需求。

除此之外，Pub/Sub 还提供了「匹配订阅」模式，允许消费者根据一定规则，订阅「多个」自己希望的队列。

可以看到，Pub/Sub 最大的优势就是，支持多组生产者、消费者处理消息。

缺点就是：丢数据。

Pub/Sub 没有基于任何数据类型，也没有做任何的数据存储（不会写入到 RDB 和 AOF 中），单纯的建立转发通道，把符合规则的数据转发到另外一端，一切都是实时转发的。

如果消费者异常，那么再次上线只能接受新的消息，在此期间生产者找不到消费者就会丢弃数据。
使用 Pub/Sub 时，注意：消费者必须先订阅队列，生产者才能发布消息，否则消息会丢失。

消息积压时消息也可能会消息丢失或者消费失败，Pub/Sub的实现上就是在server的内存上给订阅的消费者分配了一个buffer。

生产者发布消息不断写入buffer中，当消息积压时，buffer占用内存会持续增长，如果突破了buffer配置的上线，那么消费者就会被踢下线，导致消费失败，数据丢失。

缓冲区的默认配置：client-output-buffer-limit pubsub 32mb 8mb 60。
32mb：缓冲区一旦超过 32MB，Redis 直接强制把消费者踢下线.
8mb + 60：缓冲区超过 8MB，并且持续 60 秒，Redis 也会把消费者踢下线

List 拉数据，Pub/Sub推数据。

Pub/Sub 的优缺点：
1、支持发布 / 订阅，支持多组生产者、消费者处理消息
2、消费者下线，数据会丢失
3、不支持数据持久化，Redis 宕机，数据也会丢失
4、消息堆积，缓冲区溢出，消费者会被强制踢下线，数据也会丢失

哨兵集群和 Redis 实例通信时，采用了 Pub/Sub 的方案，因为哨兵正好符合即时通讯的业务场景。

很明显Pub/Sub不是我们想要的消息队列，继续往下看

1.3、基于 Streams 的消息队列解决方案

Streams 是 Redis 专门为消息队列设计的数据类型，它提供了丰富的消息队列操作命令。

XADD:插入消息，保证有序，可以自动生成全局唯一ID
XREAD:用于读取消息，可以按ID读取数据
XREADGROUP:按消费组形式读取消息
XPENDING:可以用来查询每个消费组内所有消费者已读取但尚未确认的消息
XACK:用于向消息队列确认消息处理已完成

生产者推消息：

// *表示让Redis自动生成消息ID
127.0.0.1:6379> XADD queue * name zhangsan
"1618469123380-0"
127.0.0.1:6379> XADD queue * name lisi
"1618469127777-0"

消费者拉消息：
XADD「*」表示让 Redis 自动生成唯一的消息 ID
消息 ID 的格式是「时间戳-自增序号」(自增序号从0开始编号)

// 从开头读取5条消息，0-0表示从开头读取
127.0.0.1:6379> XREAD COUNT 5 STREAMS queue 0-0
1) 1) "queue"2) 1) 1) "1618469123380-0"2) 1) "name"2) "zhangsan"2) 1) "1618469127777-0"2) 1) "name"2) "lisi"

如果想继续拉取消息，需要传入上一条消息的 ID：

127.0.0.1:6379> XREAD COUNT 5 STREAMS queue 1618469127777-0
(nil)

这就是Stream 最简单的生产、消费。

1.3.1、数据保证顺序

支持。
XADD插入消息，保证有序

1.3.2、支持阻塞等待拉取消息

支持。
在读取消息时，只需要增加 BLOCK 参数即可。

// BLOCK 0 表示阻塞等待，不设置超时时间
127.0.0.1:6379> XREAD COUNT 5 BLOCK 0 STREAMS queue 1618469127777-0

这时，消费者就会阻塞等待，直到生产者发布新的消息才会返回。

1.3.3、发布/订阅模式

支持。
Stream 通过以下命令完成发布订阅：
XGROUP：创建消费者组
XREADGROUP：在指定消费组下，开启消费者拉取消息

127.0.0.1:6379> XADD queue * name zhangsan
"1618470740565-0"
127.0.0.1:6379> XADD queue * name lisi
"1618470743793-0"

// 创建消费者组1，0-0表示从头拉取消息
127.0.0.1:6379> XGROUP CREATE queue group1 0-0
OK
// 创建消费者组2，0-0表示从头拉取消息
127.0.0.1:6379> XGROUP CREATE queue group2 0-0
OK

第一个消费组开始消费：

// group1的consumer开始消费，>表示拉取最新数据
127.0.0.1:6379> XREADGROUP GROUP group1 consumer COUNT 5 STREAMS queue >
1) 1) "queue"2) 1) 1) "1618470740565-0"2) 1) "name"2) "zhangsan"2) 1) "1618470743793-0"2) 1) "name"2) "lisi"

同样地，第二个消费组开始消费：

// group2的consumer开始消费，>表示拉取最新数据
127.0.0.1:6379> XREADGROUP GROUP group2 consumer COUNT 5 STREAMS queue >
1) 1) "queue"2) 1) 1) "1618470740565-0"2) 1) "name"2) "zhangsan"2) 1) "1618470743793-0"2) 1) "name"2) "lisi"

我们可以看到，这 2 组消费者，都可以获取同一批数据进行处理了。

通过创建消费组的形式达到订阅的目的。

1.3.4、重新消费

支持。

上面拉取消息时用到了消息 ID，这里为了保证重新消费，也要用到这个消息 ID。
当一组消费者处理完消息后，需要执行 XACK 命令告知 Redis，这时 Redis 就会把这条消息标记为「处理完成」。

// group1下的 1618472043089-0 消息已处理完成
127.0.0.1:6379> XACK queue group1 1618472043089-0

如果消费者异常宕机，肯定不会发送 XACK，那么 Redis 就会依旧保留这条消息。

待这组消费者重新上线后，Redis 就会把之前没有处理成功的数据，重新发给这个消费者。这样一来，即使消费者异常，也不会丢失数据了。

// 消费者重新上线，0-0表示重新拉取未ACK的消息
127.0.0.1:6379> XREADGROUP GROUP group1 consumer1 COUNT 5 STREAMS queue 0-0
// 之前没消费成功的数据，依旧可以重新消费
1) 1) "queue"2) 1) 1) "1618472043089-0"2) 1) "name"2) "zhangsan"2) 1) "1618472045158-0"2) 1) "name"2) "lisi"

1.3.5、消息不丢失

Stream 是新增加的数据类型，它与其它数据类型一样，每个写操作，也都会写入到 RDB 和 AOF 中。

我们只需要配置好持久化策略，这样的话，就算 Redis 宕机重启，Stream 中的数据也可以从 RDB 或 AOF 中恢复回来。

1.3.6、消息堆积

支持，但有长度限制。

当消息队列发生消息堆积时，一般只有 2 个解决方案：
1、生产者限流：避免消费者处理不及时，导致持续积压
2、丢弃消息：中间件丢弃旧消息，只保留固定长度的新消息

Redis 在实现 Stream 时，采用了第 2 个方案。

在发布消息时，你可以指定队列的最大长度，防止队列积压导致内存爆炸。

// 队列长度最大10000
127.0.0.1:6379> XADD queue MAXLEN 10000 * name zhangsan
"1618473015018-0"

当队列长度超过上限后，旧消息会被删除，只保留固定长度的新消息。
这么来看，Stream 在消息积压时，如果指定了最大长度，还是有可能丢失消息的。

除了以上介绍到的命令，Stream 还支持查看消息长度（XLEN）、查看消费者状态（XINFO）等命令

1.3.7、小结

需求	Stream
数据保证顺序	支持
支持阻塞等待拉取消息	支持
支持发布 / 订阅模式	支持
重复消费	支持
消息不丢失	支持
消息堆积	支持

既然它的功能这么强大，这是不是意味着，Redis 真的可以作为专业的消息队列中间件来使用呢？

2、与专业的消息队列对比

一个专业的消息队列，必须要做到两大块：
1、消息不丢
2、消息可堆积

消息队列，其实就分为三大块：生产者、队列中间件、消费者。

2.1、如何保证不丢消息？

2.1.1、生产者会不会丢失数据？

生产者丢失：
1、消息没法出去，网络原因或者其他原因，中间件返回失败
2、不确定是否发送成功：网络原因等导致发布超时，数据可能已经发送成功，但读取响
应超时

第一种情况，重发即可。
第二种情况，因为不知道是否成功，为了避免丢失，就只能也重试发送到成功为止。

生产者一般设定重试次数，超过上限次数需记录日志，发送警报。

是的，为了不丢失，可以接受重复发送，在消费端就需要做一些逻辑判断了，业务可能需要保证幂等性。

所以，redis或者其他中间件队列，都可以在生产者上保证不丢失数据。

2.1.2、消费者会不会丢失数据？

消费者拿到消息后，还没处理完成，就异常宕机了，那消费者还能否重新消费失败的消息？
要解决这个问题，消费者在处理完消息后，必须「告知」队列中间件，队列中间件才会把标记已处理，否则仍旧把这些数据发给消费者。
这种方案需要消费者和中间件互相配合，才能保证消费者这一侧的消息不丢。
无论是 Redis 的 Stream，还是专业的队列中间件，例如 RabbitMQ、Kafka，其实都是这么做的。

所以，从这个角度来看，Redis 也是合格的。

2.1.3、队列中间件会不会丢失数据？

上面的问题只要客户端和服务端配合好，就能保证生产端、消费端都不丢消息。

但是，如果队列中间件本身就不可靠呢？

在这个方面，Redis 其实没有达到要求。

Redis 在以下 2 个场景下，都会导致数据丢失。

1、AOF 持久化配置为每秒写盘，但这个写盘过程是异步的，Redis 宕机时会存在数据丢失的可能

2、主从复制也是异步的，主从切换时，也存在丢失数据的可能（从库还未同步完成主库发来的数据，就被提成主库）

基于以上原因我们可以看到，Redis 本身的无法保证严格的数据完整性。

RabbitMQ 或 Kafka 这类专业的队列中间件，在使用时，一般是部署一个集群，生产者在发布消息时，队列中间件通常会写「多个节点」，以此保证消息的完整性。这样一来，即便其中一个节点挂了，也能保证集群的数据不丢失。

Redis 的定位则不同，它的定位更多是当作缓存来用，它们两者在这个方面肯定是存在差异的。

2.1.4、消息积压怎么办？

Redis 的数据都存储在内存中，这就意味着一旦发生消息积压，则会导致 Redis 的内存持续增长，如果超过机器内存上限，就会面临被 OOM 的风险。
Redis 的 Stream 提供了可以指定队列最大长度的功能，就是为了避免这种情况发生。

但 Kafka、RabbitMQ 这类消息队列就不一样了，它们的数据都会存储在磁盘上，磁盘的成本要比内存小得多，当消息积压时，无非就是多占用一些磁盘空间，相比于内存，在面对积压时也会更加「坦然」。

把 Redis 当作队列来使用时，始终面临的 2 个问题：
1、Redis 本身可能会丢数据,
2、面对消息积压 Redis 内存资源紧张.

如果你的业务场景足够简单，对于数据丢失不敏感，而且消息积压概率比较小的情况下，把 Redis 当作队列是完全可以的。

而且，Redis 相比于 Kafka、RabbitMQ，部署和运维也更加轻量。

如果你的业务场景对于数据丢失非常敏感，而且写入量非常大，消息积压时会占用很多的机器资源，那么我建议你使用专业的消息队列中间件。

3、额外补充

3.1、延迟队列

应用场景：
1、订单超时未支付，关闭订单退还库存
2、订单完成5天后没有评论自动好评
3、用户并发量大，延后发送邮件短信
4、…

3.1.1实现方式

1. ZSET + 定时轮询

   1. zset支持高性能的 score 排序,且去重
   2. 内存上进行操作的，速度非常快
   3. 注意多进程争抢，使用lua将zrangebyscore和zrem进行原子化

2. 监听key（不建议）

   1. WATCH 可以鉴定单个或者多个key的变化情况
   2. 数量较大时，监听会滞后(过期事件是在Redis服务器删除密钥时产生的，而不是在理论上存活时间达到零时产生的)

参考、复制、学习、引用与：

redis官网
请勿过度依赖 Redis 的过期监听
把Redis当作队列来用，真的合适吗?
消息队列的考验：Redis有哪些解决方案？