RocketMQ 顺序消费机制

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

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

1、分区顺序消息

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

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

2、全局顺序消息

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

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

全局顺序消息实际上是一种特殊的分区顺序消息，即 Topic 中只有一个分区，因此全局顺序和分区顺序的实现原理相同。

因为分区顺序消息有多个分区，所以分区顺序消息比全局顺序消息的并发度和性能更高。

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

• 消息发送

  如上图所示，A1、B1、A2、A3、B2、B3 是订单 A 和订单 B 的消息产生的顺序，业务上要求同一订单的消息保持顺序，例如订单 A 的消息发送和消费都按照 A1、A2、A3 的顺序。

  如果是普通消息，订单A 的消息可能会被轮询发送到不同的队列中，不同队列的消息将无法保持顺序，而顺序消息发送时 RocketMQ 支持将 Sharding Key 相同（例如同一订单号）的消息序路由到一个队列中。

  

• 消息消费

  消费者消费消息时，使用单线程消费重平衡已分配的消息队列，保证消息消费顺序和存储顺序一致，最终实现消费顺序和发布顺序的一致。

  

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

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

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

1、 组装成消费对象

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

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

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

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

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

5、 执行消息监听器

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

6、 处理消费结果

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

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

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

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

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