五张图带你理解 RocketMQ 顺序消息实现机制

大家好，我是君哥。今天聊一聊 RocketMQ 的顺序消息实现机制。,在有些场景下，使用 MQ 需要保证消息的顺序性，比如在电商系统中，用户提交订单、支付订单、订单出库这 3 个消息应该保证顺序性，如下图：,对于 RocketMQ 来说，主要是通过 Producer 和 Consumer 来保证消息顺序的。,下面代码是 Producer 发送顺序消息的官方示例：,跟发送并发消息不一样的是，发送消息时传入了 MessageQueueSelector，这里可以指定消息发送到固定的 MessageQueue。,注意：上面的代码把 orderId 相同的消息都会发送到同一个 MessageQueue，这样同一个 orderId 的消息是有序的，这也叫做局部有序。对应的另一种是全局有序，这需要把所有的消息都发到同一个 MessageQueue。,下面再来看一下发送的代码：,可以看到，在发送的时候，使用 MessageQueueSelector 选择一个 MessageQueue，然后发送消息到这个 MessageQueue。对于并发消息，这里不传 MessageQueueSelector，如果发送方法没有指定 MessageQueue，就会按照默认的策略选择一个。,以 RocketMQ 推模式为例，消费者会注册一个监听器，进行消息的拉取和消费处理，下面的 UML 类图显示了调用关系：,上图中包含了对顺序消息和对并发消息的处理。其中 MessageListenerOrderly 和 ConsumeMessageOrderlyService 对顺序消息进行处理。跟并发消息不一样的是，顺序消息定义了一个 MessageQueueLock 类，这个类保存了每个 MessageQueue 对应的锁，代码如下：,下面代码是顺序消费的官方示例：,下面看一下顺序消息的消费端处理逻辑。,上面的代码定义了顺序消息监听器 MessageListenerOrderly，并且注册到 DefaultMQPushConsumer，这个注册同时也注册到了 DefaultMQPushConsumerImpl。,在 DefaultMQPushConsumerImpl 类初始化的时候，会判断注册的 MessageListener 是不是 MessageListenerOrderly，如果是，就把 consumeOrderly 变量设置为 true，以此来标记是顺序消息拉取还是并发消息拉取。然后把 ConsumeMessageService 初始化为 ConsumeMessageOrderlyService。,要保证消息的顺序性，就需要保证同一个 MessageQueue 只能被同一个 Consumer 消费。,ConsumeMessageOrderlyService 初始化的时候，会启动一个定时任务，周期性（默认 20s）地向 Broker 发送锁定消息（请求类型 LOCK_BATCH_MQ），Broker 收到后，就会把 MessageQueue、group 和 clientId 进行绑定，这样其他客户端就不能从这个 MessageQueue 拉取消息。,注意：Broker 的锁定是有过期时间的，默认 60s，可以配置，锁定过期后，有可能被其他 Consumer 进行消费。,Broker 端锁结构如下图：,消费者启动时，启动消费拉取线程 PullMessageService，里面死循环不停地从 Broker 拉取消息。这里调用了 DefaultMQPushConsumerImpl 类的 pullMessage 方法。这里拉取消息的逻辑跟并发消息逻辑是一样的。,拉取到消息后，调用 PullCallback 的 onSuccess 方法处理结果，这里调用了 ConsumeMessageOrderlyService 的 submitConsumeRequest 方法，里面用线程池提交了 ConsumeRequest 线程。,上面拉取到消息后，先把消息放到了 ProcessQueue，然后调用了 submitConsumeRequest 方法。跟并发消息处理方式不同的是，submitConsumeRequest 方法并没有处理拉取到的消息，而真正处理的时候是从 ProcessQueue 获取。,处理消息的逻辑在 ConsumeMessageOrderlyService 的内部类 ConsumeRequest，这是一个线程类，run 方法如下：,上面的代码总结一下，Consumer 消费消息的逻辑如下：,注意：ProcessQueue 中的锁是 ReentrantLock。,跟并发消息不一样的是，顺序消息消费失败后并不会把消息发送到 Broker，而是直接在 Consumer 端进行重试，如果重试次数超过了最大重试次数（16 次），则发送到 Broker，Broker 则将消息推入死信队列。如下图：,RocketMQ 顺序消息的原理是在 Producer 端把一批需要保证顺序的消息发送到同一个 MessageQueue，Consumer 端则通过加锁的机制来保证消息消费的顺序性，Broker 端通过对 MessageQueue 进行加锁，保证同一个 MessageQueue 只能被同一个 Consumer 进行消费。,根据实现原理可以看到，RocketMQ 的顺序消息可能存在两个坑：,Broker1 发生故障，把订单出库的消息发送到了 Broker2，由 Consumer2 来进行消费，消息顺序很可能会错乱。