如何使用Redis实现消息队列

朱雀 2023-10-10 23:15 36阅读 0赞

#### 如何使用Redis实现消息队列 ####

*  基于List的LPUSH+BRPOP实现
 *  基于Sorted-Set实现
 *  基于PUB/SUB, 订阅/发布模式实现
 *   *  优点
     *  缺点
 *  基于Stream类型的实现
 *   *  Stream消息太多怎么办
     *  消息如果忘记ACK会怎样
     *  PEL如何避免消息丢失
     *  死信问题
     *  Stream的高可用
     *  分区Partition

## 基于List的LPUSH+BRPOP实现 ##

**足够简单**，消费消息延迟几乎为零，但是需要处理空闲连接的问题。

如果线程一直阻塞在那里，Redis客户端的连接就成了闲置连接，闲置过久，服务器一般会主动断开连接，减少闲置资源占用，这个时候blpop和brpop或抛出异常，所以在编写客户端消费者的时候要小心，如果捕获到异常，还有重试。

其他缺点包括：  
**做消费者确认ACK麻烦**，不能保证消费者消费消息后是否成功处理的问题（宕机或处理异常等），**通常需要维护一个Pending列表**，保证消息处理确认；不能做广播模式，如pub/sub，消息发布/订阅模型；不能重复消费，一旦消费就会被删除；不支持分组消费。

## 基于Sorted-Set实现 ##

多用来实现**延迟队列**，当然也可以实现有序的普通的消息队列，但是消费者**无法阻塞**的获取消息，只能轮询，**不允许重复消息**。

## 基于PUB/SUB, 订阅/发布模式实现 ##

### 优点 ###

典型的**广播模式**，一个消息可以发布到**多个消费者**；多信道订阅，消费者可以**同时订阅多个信道**，从而接收多类消息；消息即时发送，消息不同等待消费者读取，消费者会自动接收到信道发布的消息。

### 缺点 ###

消息一旦分布，不能接收。换句话就是发布时**若客户端不在线，则消息丢失，不能寻回**；不能保证每个消费者接收的时间是一致的；若消费者客户端出现消息积压，到一定程度，会被强制断开，导致消息意外丢失。通常发生在消息的生产远大于消费速度时；可见，**Pub/Sub模式不适合做消息存储，消息积压类的业务**，而是擅长处理广播，即时通讯，即时反馈的业务。

## 基于Stream类型的实现 ##

![在这里插入图片描述][53cdc5a97dde4b15a7c8a55c353b0477.png]

### Stream消息太多怎么办 ###

要是消息积累太多，Stream的链表岂不是很长，内容会不会爆掉？xdel指令又不会删除消息，它只是给消息做了标志位。

Redis自然考虑到了这一点，所以它**提供了一个定长Stream功能**，在xadd的指令提供一个定长长度maxlen，就可以将老的消息干掉，确保最多不超过指定长度。

### 消息如果忘记ACK会怎样 ###

Stream在每个消费者结构中保存了正在处理中的消息ID列表PEL，如果消费者收到了消息处理完了但是没有回复ack，就会导致PEL列表不断增长，如果有很多消费组的话，那么这个PEL占用的内存就会放大。**所以消息要尽可能的快速消费并确认**。

### PEL如何避免消息丢失 ###

在客户端消费者读取Stream消息时，Redis服务器将消息回复给客户端的过程中，客户端突然断开了连接，消息就丢失了。但是PEL里已经保存了发出去的消息ID。待客户端重新连上之后，可以再次收到PEL中的消息ID列表。不过此时xreadgroup的起始消息ID不能为参数，而必须是任意有效的消息ID，一般将参数设为0-0，表示读取所有的PEL消息以及自last\_delivered\_id之后的新消息。

### 死信问题 ###

如果某个消息，不能被消费者处理，也就是不能被XACK，这是要长时间处于Pending列表中，即使被反复的转移给各个消费者也是如此。此时该消息的delivery counter（通过XPENDING可以查询到）就会累加，**当累加到某个我们预设的临界值时，我们就认为是坏消息**（也叫死信，DeadLetter，**无法投递的消息**），由于有了判定条件，我们将坏消息处理即可，**删除即可**。删除一个消息，使用XDEL语法，注意，这个命令并没有删除Pending中的消息，因此查看Pending，消息还会在，可以在执行XDEL之后，XACK这个消息标识其处理完毕。

### Stream的高可用 ###

Stream的高可用是建立主从复制基础上的，它和其他数据结构的复制机制没有区别，也就是说在Sentinel和Cluster集群环境下Stream是可以**支持高可用的**。不过鉴于Redis的指令复制是异步的，在failover发生时，**Redis可能会丢失极小部分数据**，这点Redis的气压数据结构也是一样的。

### 分区Partition ###

Redis的服务器**没有原生支持分区能力**，如果想要使用分区，那就**需要分配多个Stream**，然后在客户端使用一定的策略来生产消息到不同的Stream。

**参考资料**：[Redis刷题13：怎么使用Redis实现消息队列？][Redis_13_Redis]

[53cdc5a97dde4b15a7c8a55c353b0477.png]: https://img-blog.csdnimg.cn/53cdc5a97dde4b15a7c8a55c353b0477.png
[Redis_13_Redis]: https://www.bilibili.com/video/BV17L411X7aT/?spm_id_from=333.337.search-card.all.click&vd_source=a1af903e57688df870625021f7f3a7eb