RocketMQ 顺序消费机制

柔光的暖阳◎ 2023-10-11 22:57 18阅读 0赞

顺序消息是指对于一个指定的 Topic ，消息严格按照先进先出（FIFO）的原则进行消息发布和消费，即先发布的消息先消费，后发布的消息后消费。

顺序消息分为**分区顺序消息**和**全局顺序消息**。

**1、分区顺序消息**

对于指定的一个 Topic ，所有消息根据 Sharding Key 进行区块分区，同一个分区内的消息按照严格的先进先出（FIFO）原则进行发布和消费。同一分区内的消息保证顺序，不同分区之间的消息顺序不做要求。

*  适用场景：适用于性能要求高，以 Sharding Key 作为分区字段，在同一个区块中严格地按照先进先出（FIFO）原则进行消息发布和消费的场景。
 *  示例：电商的订单创建，以订单 ID 作为 Sharding Key ，那么同一个订单相关的创建订单消息、订单支付消息、订单退款消息、订单物流消息都会按照发布的先后顺序来消费。

**2、全局顺序消息**

对于指定的一个 Topic ，所有消息按照严格的先入先出（FIFO）的顺序来发布和消费。

*  适用场景：适用于性能要求不高，所有的消息严格按照 FIFO 原则来发布和消费的场景。
 *  示例：在证券处理中，以人民币兑换美元为 Topic，在价格相同的情况下，先出价者优先处理，则可以按照 FIFO 的方式发布和消费全局顺序消息。

> 全局顺序消息实际上是一种特殊的分区顺序消息，即 Topic 中只有一个分区，因此**全局顺序和分区顺序的实现原理相同**。
> 
> 因为分区顺序消息有多个分区，所以**分区顺序消息比全局顺序消息的并发度和性能更高**。

![format_png][]

消息的顺序需要由两个阶段保证：

*  **消息发送**
    
    如上图所示，A1、B1、A2、A3、B2、B3 是订单 A 和订单 B 的消息产生的顺序，业务上要求同一订单的消息保持顺序，例如订单 A 的消息发送和消费都按照 A1、A2、A3 的顺序。
    
    如果是普通消息，订单A 的消息可能会被轮询发送到不同的队列中，不同队列的消息将无法保持顺序，而顺序消息发送时 RocketMQ 支持将 Sharding Key 相同（例如同一订单号）的消息序路由到一个队列中。
    
    ![format_png 1][]
 *  **消息消费**
    
    消费者消费消息时，使用单线程消费重平衡已分配的消息队列，保证消息消费顺序和存储顺序一致，最终实现消费顺序和发布顺序的一致。
    
    ![format_png 2][]

我们知道**负载均衡服务是客户端开始消费的起点**。在负载均衡阶段，并发消费和顺序消费并没有什么大的差别，最大的差别在于：**向 Borker 申请锁** 。

消费者根据分配的队列 messageQueue ，向 Borker 申请锁 ，如果申请成功，则会拉取消息，如果失败，则定时任务每隔20秒会重新尝试。

![format_png 3][]

见上图，顺序消费核心流程如下：

**1、 组装成消费对象**

**2、 将请求对象提交到消费线程池**

![format_png 4][]

和并发消费不同的是，这里的消费请求包含消费快照 processQueue ，消息队列 messageQueue 两个对象，并不对消息列表做任何处理。

**3、 消费线程内，对消费队列加锁**

![format_png 5][]

**4、 从消费快照中取得待消费的消息列表**

![format_png 6][]

消费快照 processQueue 对象里，创建了一个红黑树对象 consumingMsgOrderlyTreeMap 用于临时存储的待消费的消息。

**5、 执行消息监听器**

![format_png 7][]

执行监听器逻辑容易理解，消费快照的**消费锁 consumeLock**的作用是：防止 Rebalance 线程把当前消费的 MessageQueue 对象移除掉。

**6、 处理消费结果**

消费成功时，首先获取需要提交的偏移量，然后更新本地消费进度。

![format_png 8][]

消费失败时，分两种场景：

*  假如已消费次数小于最大重试次数，则将放入对象 consumingMsgOrderlyTreeMap 用例临时存储的待消费的消息，重新加入到消费快照**红黑树 msgTreeMap**中，然后使用定时任务尝试重新消费。
 *  假如已消费次数大于等于最大重试次数，则将失败消息发送到 Broker ，Broker 接收到消息后，会加入到死信队列里 , 获取需要提交的偏移量，然后更新本地消费进度。

我们做一个关于顺序消费的总结：

1.  顺序消费需要由两个阶段**消息发送**和**消息消费**协同配合，底层支撑依靠的是 RocketMQ 的存储模型；
2.  顺序消费服务启动后，通过三把锁的机制，使得消费者实例单线程的消费重平衡分配的消费队列；
3.  假如发生扩容，消费者重启，或者 Broker 宕机 ，顺序消费也会有一定几率较短时间内乱序，所以消费者的业务逻辑还是要**保障幂等**。

[format_png]: https://img-blog.csdnimg.cn/img_convert/16d35e24053593bec164ca0fa10d5638.webp?x-oss-process=image/format,png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/068e361f8af7ddcac260592488b37a2e.webp?x-oss-process=image/format,png
[format_png 2]: https://img-blog.csdnimg.cn/img_convert/f5f32559f135ed3402a6a1c0e64d3469.webp?x-oss-process=image/format,png
[format_png 3]: https://img-blog.csdnimg.cn/img_convert/80d7dd3a5c7e23ce338111fcc365e460.webp?x-oss-process=image/format,png
[format_png 4]: https://img-blog.csdnimg.cn/img_convert/55ea3a80a2b4359cc5ef7104441a1a0b.webp?x-oss-process=image/format,png
[format_png 5]: https://img-blog.csdnimg.cn/img_convert/825cd643cc900e5f26e1fe0165661ee3.webp?x-oss-process=image/format,png
[format_png 6]: https://img-blog.csdnimg.cn/img_convert/34a60afcfd940e5ae404da15589b1c1b.webp?x-oss-process=image/format,png
[format_png 7]: https://img-blog.csdnimg.cn/img_convert/6ac8a178f107e5e59fa5851d57347a18.webp?x-oss-process=image/format,png
[format_png 8]: https://img-blog.csdnimg.cn/img_convert/2e4d5591aefc91b3c7887277c5b4f5db.webp?x-oss-process=image/format,png