JAVA面试题分享三百三十三：如何保证RocketMQ消息有序？如何解决RocketMQ消息积压？_rocketmq保证顺序消费

1、为什么需要消息有序

假设准备去银行存取款，对应两个异步短信消息，要保证先存后取：

• M1 存钱

• M2 取钱

而MQ默认发消息到不同Q显然是行不通的，会乱序。因此，需发往同一Q，依赖队列的先进先出机制。

再例如

如果我们有个大数据系统，需要对业务系统的日志进行收集分析，这时候为了减少对业务系统的影响，通常都会通过MQ来做消息中转。而这时候，对消息的顺序就有一定的要求了。例如我们考虑下面这一系列操作：

1. 用户的积分默认是0分，而新注册用户设置为默认的10分。

2. 用户有奖励行为，积分+2分

3. 用户有不正当行为，积分-3分

这样一组操作，正常用户积分要变成9分。但是如果顺序乱了，这个结果就全部对不上。这时，就需要对这一组操作，保证消息都是有序的。

2、基本概念

有序消息，又叫顺序消息(FIFO消息)，指消息的消费顺序和产生顺序相同。

如订单的生成、付款、发货，这串消息必须按序处理。顺序消息又可分为：

2.1 全局顺序

一个Topic内所有的消息都发布到同一Q，按FIFO顺序进行发布和消费：

适用场景

性能要求不高，所有消息严格按照FIFO进行消息发布和消费的场景。

2.2 分区顺序

对于指定的一个Topic，所有消息按sharding key进行区块(queue)分区，同一Q内的消息严格按FIFO发布和消费。

• Sharding key是顺序消息中用来区分不同分区的关键字段，和普通消息的Key完全不同。

适用场景

性能要求高，根据消息中的sharding key去决定消息发送到哪个queue。

2.3 对比

• 发送方式对比

3、如何保证消息顺序？

在MQ模型中，顺序需由3个阶段去保障

1. 消息被发送时保持顺序

2. 消息被存储时保持和发送的顺序一致

3. 消息被消费时保持和存储的顺序一致

如何保证消息有序

MQ的顺序问题分为全局有序和局部有序

• 全局有序：整个MQ系统的所有消息严格按照队列先入先出顺序进行消费

• 局部有序：只保证一部分关键信息的消费顺序

首先，我们需要分析下这个问题，在通常的业务场景中，全局有序和局部有序哪个更重要？其实大部分的MQ业务场景，我们只需要保证局部有序就可以了。例如我们用QQ聊天，只需要保证一个聊天窗口里的消息有序就可以了。而对于电商订单场景，也只要保证一个订单的所有消息是有序的就可以了。至于全局的消息的顺序，并不会太关心。而通常意义下，全局有序都可以压缩程局部有序的问题。例如以前我们常用的聊天室，就是一个典型的需要保证消息全局有序的场景。但是这种场景，通常可以压缩成只有一个聊天窗口的QQ来理解。即整个系统只有一个聊天通道，这样就可以用QQ那种保证一个聊天窗口消息有序的方式来保证整个系统的全局消息有序。

然后，落地到RocketMQ。通常情况下，发送者发送消息时，会通过MessageQueue轮询的方式保证消息尽量均匀分布到所有的MessageQueue上，而消费者也就同样需要从多个MessageQueue上消费消息。而MessageQueue是RocketMQ存储消息的最小单元，他们之间的消息都是互相隔离的，在这种情况下，是无法保证消息全局有序的。

而对于局部有序的要求，只需要将有序的一组消息都存入同一个MessageQueue里，这样MessageQueue的FIFO设计天生就可以保证这一组消息的有序。RocketMQ中，可以在发送者发送消息时指定一个MessageSelector对象，让这个对象来决定消息发入哪一个MessageQueue。这样就可以保证一组有序的消息能够发到同一个MessageQueue里。

另外，通常所谓的保证Topic全局消息有序的方式，就是将Topic配置成只有一个MessageQueue队列（默认是4个）。这样天生就能保证消息全局有序了。这个说法其实就是我们将聊天室场景压缩只有一个聊天窗口的QQ一样的理解方式。而这种方式对整个Topic的消息吞吐影响是非常大的，如果这样用，基本就没有用MQ的必要了。

4、RocketMQ 有序消息实现原理

RocketMQ消费端有两种类型：

• MQPullConsumer

• MQPushConsumer

底层都是通过pull机制实现，pushConsumer是一种API封装而已。

• MQPullConsumer 由用户控制线程，主动从服务端获取消息，每次获取到的是一个MessageQueue中的消息。

  • PullResult中的 List<MessageExt> msgFoundList

• MQPushConsumer由用户注册MessageListener来消费消息，在客户端中需要保证调用MessageListener时消息的顺序性

看源码

• 拉取生产端消息

• 判断是并发的还是有序的,对应不同服务实现类

5、有序消息的缺陷

发送顺序消息无法利用集群的Failover特性，因为不能更换MessageQueue进行重试。

因为发送的路由策略导致的热点问题，可能某一些MessageQueue的数据量特别大

• 消费的并行读依赖于queue数量

• 消费失败时无法跳过

使用RocketMQ如何快速处理积压消息

如何确定RocketMQ有大量的消息积压

在正常情况下，使用MQ都会要尽量保证他的消息生产速度和消费速度整体上是平衡的，但是如果部分消费者系统出现故障，就会造成大量的消息积累。这类问题通常在实际工作中会出现的比较隐蔽。

例如某一天一个数据库突然挂了，大家大概率就会集中处理数据库的问题。

等好不容易把数据库恢复过来了，这时基于数据库服务的消费者程序就会积累大量的消息。或者网络波动等情况，也会导致消息大量的积累。这在一些大型的互联网项目中，消息积压的速度是相当恐怖的。所以消息积压是个需要时刻关注的问题。

对于消息积压，如果是RocketMQ或者kafka还好，他们的消息积压不会对性能造成很大的影响。

而如果是RabbitMQ的话，那就惨了，大量的消息积压可以瞬间造成性能曲线下滑。

对于RocketMQ来说，有个最简单的方式来确定消息是否有积压。那就是使用web控制台，就能直接看到消息的积压情况。

在web控制台的主题页面，可以通过Consumer管理按钮实时看到消息的积压情况。

另外，也可以通过mqadmin指令在后台检查各个Topic的消息延迟情况。

还有RocketMQ也会在他的 ${storePathRootDir}/config 目录下落地一系列的json文件，也可以用来跟踪消息积压情况。

如何处理大量积压消息

其实我们回顾下RocketMQ的负载均衡的内容就不难想到解决方案。

如果Topic下的MessageQueue配置的是足够多的，那每个Consumer实际上会分配多个MessageQueue来进行消费。

这个时候，就可以简单的通过增加Consumer的节点个数设置成跟MessageQueue的个数相同，但是如果此时再继续增加Consumer的服务节点就没有用了。

而如果Topic下的MessageQueue配置的不够多的话，那就不能用上面这种增加Consumer节点个数的方法了。这时怎么办呢？

这时如果要快速处理积压的消息，可以创建一个新的Topic，配置足够多的MessageQueue。

然后把所有消费者节点的目标Topic转向新的Topic，并紧急上线一组新的消费者，只负责转储，就是消费老Topic中的积压消息，并转储到新的Topic中，这个速度是可以很快的。

然后在新的Topic上，就可以通过增加消费者个数来提高消费速度了。之后再根据情况恢复成正常情况。

在官网中，还分析了一个特殊情况。就是如果RocketMQ原来是采用的普通方式搭建主从架构，而现在想要中途改为使用Dledger高可用集群，这时候如果不想历史消息丢失，就需要先将消息进行对齐，也就是要消费者先把所有的消息都消费完，再来切换主从架构。

因为Dledger集群会接管RocketMQ原有的CommitLog日志，所以切换主从架构时，如果有消息没有消费完，这些消息是存在旧的CommitLog中的，就无法再进行消费了。这个场景下也是需要尽快的处理掉积压的消息。