Redis做消息队列

消息队列（Message Queue）是一种应用间的通信方式，是为了方便业务场景中数据的传输和处理。

• 消息队列流程中有三个角色：生产者、消费者、消息处理中心。生产者就是发啊送消息的，消费者就是接收消息的，消息处理中心就是指消息队列。
• 使用消息队列，生产者将数据发送到消息队列，不用等待消费者的响应消息，实现了数据的异步处理，使整个过程松耦合。
• 作为消息队列，最重要的是它的可靠性。可靠性既要保证数据不能丢失，又要能在功能上满足业务上的种种需求：数据只被消费一次，数据可以被多次消费，数据要保证顺序等等。
  Redis做消息队列
  Redis设计初衷是做缓存的，并不是消息队列。一开始作者的建议也是不要使用redis去做消息队列。但随着越来越多的人使用redis去做去消息队列，redis的作者终于在 Redis 5.0 版本，把 disque 功能移植到了 Redis 中，并给它定义了一个新的数据类型：Stream。
  Redis做消息队列，主要有3种模式来实现。
• List 队列
• 发布/订阅模型：Pub/Sub
• Stream数据

1. 最简单的：使用 List队列

当业务需求比较简单的时候，可以使用List来实现消息队列。
因为 List 底层的实现就是一个链表，在头部和尾部操作元素，时间复杂度都是 O(1)，这意味着它非常符合消息队列的模型。
Redis中也提供了方法来操作数据，可以左进右出，右进左出

• 生产者使用 LPUSH 发布消息：

127.0.0.1:6379> lpush testList msg1
1
127.0.0.1:6379> lpush testList msg2
2

• 消费者使用 RPOP 拉取消息：

127.0.0.1:6379> rpop testList
msg1
127.0.0.1:6379> rpop testList
msg2
127.0.0.1:6379> rpop testList

127.0.0.1:6379>

写代码时，一般会用一个死循环，不停的去拉取。
但此时会有一个问题，当队列为空时，消费者依旧会频繁拉取消息，这会造成CPU 空转，不仅浪费 CPU 资源，还会对 Redis 造成压力。
当然我们可以不用死循环，用个定时任务来拉取数据，但当有连续大量数据进来时，就不能连续消费了，而是要等定时任务才能消费。
其实Redis也给出了其他方法来处理CPU 空转的情况：BRPOP / BLPOP，这里的 B 指的是阻塞（Block），即阻塞式拉取消息。实现原理是：如果队列为空，消费者在拉取消息时就阻塞等待，一旦有新消息过来，就通知消费者立即处理新消息。使用 BRPOP 这种阻塞式方式拉取消息时，还支持传入一个超时时间，如果设置为 0，则表示不设置超时，直到有新消息才返回，否则会在指定的超时时间后返回 NULL。设置了超时时间，可以很好的解决Redis连接超时的问题。
这时用List实现消息队列是这样的：

while (true) {
    try {
        Object object = redisTemplate.opsForList().rightPop("test", Duration.ofSeconds(60));
        if (object == null) {
            // ThreadUtil.sleep(60 * 1000);
            continue;
        }
        handle(object);
    } catch (Exception e) {
        e.printStackTrace();
    }
}

此时，redis的配置中， 连接超时时间timeout: 100000（毫秒） 要大于上面 一个循环的时间，否则有可能报错。
这就是最简单的消息队列模型，简单业务中可以使用，毕竟比搭建专门的消息队列省了很多事情和成本。
List队列模型中，虽然生产者的消息可以持久化的，但还存在着2个问题：

• 1. 不能重复消费，当消费者pop拿到数据后，队列中就不存在了，即不支持多个消费者消费
• 2. 会有数据丢失，当消费者拿到数据后，还没使用时宕机了，这条数据就消失了

2. 发布/订阅模型：Pub/Sub

发布订阅模型，可以满足多个消费者的条件，流程示意图如下（来源于网络）：

image.png

Redis 提供了 PUBLISH / SUBSCRIBE 命令，来完成发布、订阅的操作。但是需要注意：

• 发布订阅流程很快完成，所以要求 消费者先订阅消息，等待生产者发布消息，这样才能不丢失消息，因为消息在redis中并不存储，redis只是根据订阅情况转发给每个订阅者一份数据；

• 上图展示了，一个生产者发布消息主题，多个消费者订阅。同样的，redis也支持多个生产者往同一个消息主题中发布消息，供多个消费者订阅使用，即支持多组生产者、多组消费者。

  
  image.png
  

  订阅/发布的命令是这样的

  
  订阅
  发布
  发布订阅模式也存在着问题：
• 1. 数据丢失，当生产者发布消息时，消费者必须在线，不然没有消费者，数据也不会持久化，就丢失了
• 2. 消息堆积，导致数据丢失
     一个完整的发布、订阅消息处理流程是这样的：

1. 消费者订阅指定队列，Redis 就会记录一个映射关系：队列->消费者。同时Redis会在 Server 上给这个消费者在分配一个缓冲区，这个缓冲区其实就是一块内存，缓冲区大小可调节；
2. 生产者向这个队列发布消息，Redis会先把消息写到对应消费者的缓冲区中，消费者在不断地从缓冲区读取消息，处理消息
   因为有了这个缓冲区，才会发生消息堆积。当消费者拉取消息很慢，就会造成生产者发布到缓冲区的消息开始积压，缓冲区内存持续增长。如果超过了缓冲区配置的上限，此时，Redis 就会强制把这个消费者踢下线。这时消费者就会消费失败，也会丢失数据。
   Redis 的配置文件中，这个缓冲区的默认配置：client-output-buffer-limit pubsub 32mb 8mb 60
   它的参数含义如下：

• 32mb：缓冲区一旦超过 32MB，Redis 直接强制把消费者踢下线
• 8mb + 60：缓冲区超过 8MB，并且持续 60 秒，Redis 也会把消费者踢下线

可以看出，Redis的消费/订阅模式，除了支持多消费者和多生产者以外，其他的还不如List模型呢，不能持久化、消息同样丢失、还多了消息堆积

3. Redis中的Stream

Stream 是新增加的数据类型，它与其它数据类型一样，每个写操作，也都会写入到 RDB 和 AOF 中
我们只需要配置好持久化策略，这样的话，就算 Redis 宕机重启，Stream 中的数据也可以从 RDB 或 AOF 中恢复回来。
Stream是专门为消息队列增加的，它也有命令来实现消息队列的功能。

• 1. 最简单的，像List数据一样，发布/读取

发布

// 使用 XADD 命令发布消息，其中的*表示让 Redis 自动生成唯一的消息 ID
// 这个消息 ID 的格式是时间戳-自增序号
127.0.0.1:6379> XADD testStr * name zhangsan
1670255186379-0
127.0.0.1:6379> XADD testStr * name lisi
1670255207117-0
127.0.0.1:6379>

消费拉取

// 从开头读取5条消息，0-0表示从开头读取
127.0.0.1:6379> XREAD COUNT 5 STREAMS testStr 0-0
testStr
1670255186379-0
name
zhangsan
1670255207117-0
name
lisi
// 继续拉取消息，需要传入上一条消息的 ID：
127.0.0.1:6379> xread count 2 streams testStr 1670255207117-0

127.0.0.1:6379>

阻塞拉取消费
Stream 可以 实现阻塞拉取，添加BLOCK参数即可。

// BLOCK 0 表示阻塞等待，不设置超时时间
127.0.0.1:6379> xread count 2 BLOCK 0 streams testStr 1670255207117-0

• 2. Stream支持发布 / 订阅模式
     发布
     发布同上，使用XADD命令即可。
     订阅消费
     这个和上面不同，此时消费 需要先创建消费者组，然后给每个消费者组下面挂一个消费者，让它们分别处理同一批数据。

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

// group1的consumer开始消费，>表示拉取最新数据
127.0.0.1:6379> XREADGROUP GROUP group1 consumer COUNT 5 STREAMS testStr >
testStr
1670255186379-0
name
zhangsan
1670255207117-0
name
lisi
// group2的consumer开始消费，>表示拉取最新数据
127.0.0.1:6379> XREADGROUP GROUP group2 consumer COUNT 5 STREAMS testStr >
testStr
1670255186379-0
name
zhangsan
1670255207117-0
name
lisi
127.0.0.1:6379>

Stream做消息队列数据固化了，从生产者和组件本身来说不会丢失了，同时它还有其他的设计，弥补了List和Redis发布/订阅的缺点。

1. Stream不会丢失数据了，因为像其他消息队列一样，添加了ACK机制，每个消息都有自己的ID，当消费者处理完数据以后，会执行 XACK 命令告知 Redis，这时 Redis 就会把这条消息标记为处理完成。如果消费者异常宕机，肯定不会发送 XACK，那么 Redis 就会依旧保留这条消息。待这组消费者重新上线后，Redis 就会把之前没有处理成功的数据，重新发给这个消费者。这样一来，即使消费者异常，也不会丢失数据了。
2. 消息堆积的处理，当消息队列发生消息堆积时，一般只有 2 个解决方案：
   • 生产者限流：避免消费者处理不及时，导致持续积压
   • 丢弃消息：中间件丢弃旧消息，只保留固定长度的新消息
     而 Redis 在实现 Stream 时，采用了第 2 个方案。
     在发布消息时，你可以指定队列的最大长度，防止队列积压导致内存爆炸。

// 队列长度最大10000
127.0.0.1:6379> XADD str2 MAXLEN 10000 * name zhangsan
1670257154826-0
127.0.0.1:6379>

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

3. Stream 还支持查看消息长度XLEN、查看消费者状态XINFO等命令，使用比较简单，可以查询官方文档了解一下

---

Redis在做的越来越好，但是就Redis本身的设计来说，比如AOF持久化机制，使Redis组件本身还存在丢数据的可能，消息堆积时，消费者也可能丢失消息。因此Redis和专业消息队列还是有差别的。
但是，Redis是轻量化的组件，作为缓存存储项目中经常用到，此时在业务需求简单且允许的情况下拿它来做消息队列还是很不错的。