您现在的位置是：亿华云 > 人工智能

顺序消息的实现-RocketMQ知识体系(五）

亿华云2025-10-03 02:20:30【人工智能】7人已围观

简介我们知道，kafka 如果要保证顺序消费，必须保证消息保存到同一个patition上，而且为了有序性，只能有一个消费者进行消费。这种情况下，Kafka 就退化成了单一队列，毫无并发性可言，极大降低系统

我们知道，顺序识体kafka 如果要保证顺序消费，消息现R系必须保证消息保存到同一个patition上，顺序识体而且为了有序性，消息现R系只能有一个消费者进行消费。顺序识体这种情况下，消息现R系Kafka 就退化成了单一队列，顺序识体毫无并发性可言，消息现R系极大降低系统性能。顺序识体那么对于对业务比较友好的消息现R系RocketMQ 是如何实现的呢?首先，我们循序渐进的顺序识体来了解下顺序消息的实现。

顺序消息业务使用场景

1、消息现R系电商场景中传递订单状态。顺序识体

2、消息现R系同步mysql 的顺序识体binlong 日志，数据库的操作是有顺序的。

3、其他消息之间有先后的依赖关系，后一条消息需要依赖于前一条消息的处理结果的情况。

等等。。。服务器托管

消息中间件中的顺序消息

顺序消息(FIFO 消息)是 MQ 提供的一种严格按照顺序进行发布和消费的消息类型。顺序消息由两个部分组成：顺序发布和顺序消费。

顺序消息包含两种类型：

分区顺序：一个Partition(queue)内所有的消息按照先进先出的顺序进行发布和消费

全局顺序：一个Topic内所有的消息按照先进先出的顺序进行发布和消费.但是全局顺序极大的降低了系统的吞吐量，不符合mq的设计初衷。

那么折中的办法就是选择分区顺序。

【局部顺序消费】

如何保证顺序

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

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

发送时保持顺序意味着对于有顺序要求的消息，用户应该在同一个线程中采用同步的方式发送。存储保持和发送的顺序一致则要求在同一线程中被发送出来的消息A和B，存储时在空间上A一定在B之前。而消费保持和存储一致则要求消息A、B到达Consumer之后必须按照先A后B的源码库顺序被处理。

第一点，消息顺序发送，多线程发送的消息无法保证有序性，因此，需要业务方在发送时，针对同一个业务编号(如同一笔订单)的消息需要保证在一个线程内顺序发送，在上一个消息发送成功后，在进行下一个消息的发送。对应到mq中，消息发送方法就得使用同步发送，异步发送无法保证顺序性。

第二点，消息顺序存储，mq的topic下会存在多个queue，要保证消息的顺序存储，同一个业务编号的消息需要被发送到一个queue中。对应到mq中，需要使用MessageQueueSelector来选择要发送的queue，即对业务编号进行hash，然后根据队列数量对hash值取余，将消息发送到一个queue中。

第三点，消息顺序消费，要保证消息顺序消费，同一个queue就只能被一个消费者所消费，高防服务器因此对broker中消费队列加锁是无法避免的。同一时刻，一个消费队列只能被一个消费者消费，消费者内部，也只能有一个消费线程来消费该队列。即，同一时刻，一个消费队列只能被一个消费者中的一个线程消费。

RocketMQ中顺序的实现

【Producer端】

Producer端确保消息顺序唯一要做的事情就是将消息路由到特定的分区，在RocketMQ中，通过MessageQueueSelector来实现分区的选择。

/** * 消息队列选择器 */ public interface MessageQueueSelector { /** * 选择消息队列 * * @param mqs 消息队列 * @param msg 消息 * @param arg 参数 * @return 消息队列 */ MessageQueue select(final List<MessageQueue> mqs, final Message msg, final Object arg); } List mqs：消息要发送的Topic下所有的分区 Message msg：消息对象额外的参数：用户可以传递自己的参数

比如如下实现就可以保证相同的订单的消息被路由到相同的分区：

long orderId = ((Order) object).getOrderId; return mqs.get(orderId % mqs.size());

【Consumer端】

尝试锁定锁定MessageQueue。

首先我们如何保证一个队列只被一个消费者消费?

消费队列存在于broker端，如果想保证一个队列被一个消费者消费，那么消费者在进行消息拉取消费时就必须向mq服务器申请队列锁，消费者申请队列锁的代码存在于RebalanceService消息队列负载的实现代码中。

消费者重新负载，并且分配完消费队列后，需要向mq服务器发起消息拉取请求，代码实现在RebalanceImpl#updateProcessQueueTableInRebalance中，针对顺序消息的消息拉取，mq做了如下判断：

// 增加不在processQueueTable && 存在于mqSet 里的消息队列。 List<PullRequest> pullRequestList = new ArrayList<>(); // 拉消息请求数组 for (MessageQueue mq : mqSet) { if (!this.processQueueTable.containsKey(mq)) { if (isOrder && !this.lock(mq)) { // 顺序消息锁定消息队列 log.warn("doRebalance, { }, add a new mq failed, { }, because lock failed", consumerGroup, mq); continue; } this.removeDirtyOffset(mq); ProcessQueue pq = new ProcessQueue(); long nextOffset = this.computePullFromWhere(mq); if (nextOffset >= 0) { ProcessQueue pre = this.processQueueTable.putIfAbsent(mq, pq); if (pre != null) { log.info("doRebalance, { }, mq already exists, { }", consumerGroup, mq); } else { log.info("doRebalance, { }, add a new mq, { }", consumerGroup, mq); PullRequest pullRequest = new PullRequest(); pullRequest.setConsumerGroup(consumerGroup); pullRequest.setNextOffset(nextOffset); pullRequest.setMessageQueue(mq); pullRequest.setProcessQueue(pq); pullRequestList.add(pullRequest); changed = true; } } else { log.warn("doRebalance, { }, add new mq failed, { }", consumerGroup, mq); } } } // 发起消息拉取请求 this.dispatchPullRequest(pullRequestList);

核心思想就是，消费客户端先向broker端发起对messageQueue的加锁请求，只有加锁成功时才创建pullRequest进行消息拉取，下面看下lock加锁请求方法：

/** * 请求Broker获得指定消息队列的分布式锁 * * @param mq 队列 * @return 是否成功 */ public boolean lock(final MessageQueue mq) { FindBrokerResult findBrokerResult = this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(), MixAll.MASTER_ID, true); if (findBrokerResult != null) { LockBatchRequestBody requestBody = new LockBatchRequestBody(); requestBody.setConsumerGroup(this.consumerGroup); requestBody.setClientId(this.mQClientFactory.getClientId()); requestBody.getMqSet().add(mq); try { // 请求Broker获得指定消息队列的分布式锁 Set<MessageQueue> lockedMq = this.mQClientFactory.getMQClientAPIImpl().lockBatchMQ(findBrokerResult.getBrokerAddr(), requestBody, 1000); // 设置消息处理队列锁定成功。锁定消息队列成功，可能本地没有消息处理队列，设置锁定成功会在lockAll()方法。 for (MessageQueue mmqq : lockedMq) { ProcessQueue processQueue = this.processQueueTable.get(mmqq); if (processQueue != null) { processQueue.setLocked(true); processQueue.setLastLockTimestamp(System.currentTimeMillis()); } } boolean lockOK = lockedMq.contains(mq); log.info("the message queue lock { }, { } { }", lockOK ? "OK" : "Failed", this.consumerGroup, mq); return lockOK; } catch (Exception e) { log.error("lockBatchMQ exception, " + mq, e); } } return false; }

代码实现逻辑比较清晰，就是调用lockBatchMQ方法发送了一个加锁请求，那么broker端收到加锁请求后的处理逻辑又是怎么样?

【broker端实现】

broker端收到加锁请求的处理逻辑在RebalanceLockManager#tryLockBatch方法中，RebalanceLockManager中关键属性如下：

/** * 消息队列锁过期时间，默认60s */ private final static long REBALANCE_LOCK_MAX_LIVE_TIME = Long.parseLong(System.getProperty( "rocketmq.broker.rebalance.lockMaxLiveTime", "60000")); /** * 锁 */ private final Lock lock = new ReentrantLock(); /** * 消费分组的消息队列锁映射 */ private final ConcurrentHashMap<String/* group */, ConcurrentHashMap<MessageQueue, LockEntry>> mqLockTable = new ConcurrentHashMap<>(1024);

LockEntry对象中关键属性如下：

/** * 锁定记录 */ static class LockEntry { /** * 客户端编号 */ private String clientId; /** * 最后锁定时间 */ private volatile long lastUpdateTimestamp = System.currentTimeMillis(); public String getClientId() { return clientId; } public void setClientId(String clientId) { this.clientId = clientId; } public long getLastUpdateTimestamp() { return lastUpdateTimestamp; } public void setLastUpdateTimestamp(long lastUpdateTimestamp) { this.lastUpdateTimestamp = lastUpdateTimestamp; } /** * 是否锁定 * * @param clientId 客户端编号 * @return 是否 */ public boolean isLocked(final String clientId) { boolean eq = this.clientId.equals(clientId); return eq && !this.isExpired(); } /** * 锁定是否过期 * * @return 是否 */ public boolean isExpired() { boolean expired = (System.currentTimeMillis() - this.lastUpdateTimestamp) > REBALANCE_LOCK_MAX_LIVE_TIME; return expired; } }

broker端通过对ConcurrentMap> mqLockTable的维护来达到messageQueue加锁的目的，使得同一时刻，一个messageQueue只能被一个消费者消费。

【再次回到Consumer端，拿到锁后】

消费者对messageQueue的加锁已经成功，那么就进入到了第二个步骤，创建pullRequest进行消息拉取，消息拉取部分的代码实现在PullMessageService中，消息拉取完后，需要提交到ConsumeMessageService中进行消费，顺序消费的实现为ConsumeMessageOrderlyService，提交消息进行消费的方法为ConsumeMessageOrderlyService#submitConsumeRequest，具体实现如下：

@Override public void submitConsumeRequest(// final List<MessageExt> msgs, // final ProcessQueue processQueue, // final MessageQueue messageQueue, // final boolean dispathToConsume) { if (dispathToConsume) { ConsumeRequest consumeRequest = new ConsumeRequest(processQueue, messageQueue); this.consumeExecutor.submit(consumeRequest); } }

构建了一个ConsumeRequest对象，并提交给了ThreadPoolExecutor来并行消费，看下顺序消费的ConsumeRequest的run方法实现：

public void run() { if (this.processQueue.isDropped()) { log.warn("run, the message queue not be able to consume, because its dropped. { }", this.messageQueue); return; } // 获得 Consumer 消息队列锁 final Object objLock = messageQueueLock.fetchLockObject(this.messageQueue); synchronized (objLock) { // (广播模式) 或者 (集群模式 && Broker消息队列锁有效) if (MessageModel.BROADCASTING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel()) || (this.processQueue.isLocked() && !this.processQueue.isLockExpired())) { final long beginTime = System.currentTimeMillis(); // 循环 for (boolean continueConsume = true; continueConsume; ) { if (this.processQueue.isDropped()) { log.warn("the message queue not be able to consume, because its dropped. { }", this.messageQueue); break; } // 消息队列分布式锁未锁定，提交延迟获得锁并消费请求 if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel()) && !this.processQueue.isLocked()) { log.warn("the message queue not locked, so consume later, { }", this.messageQueue); ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 10); break; } // 消息队列分布式锁已经过期，提交延迟获得锁并消费请求 if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel()) && this.processQueue.isLockExpired()) { log.warn("the message queue lock expired, so consume later, { }", this.messageQueue); ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 10); break; } // 当前周期消费时间超过连续时长，默认：60s，提交延迟消费请求。默认情况下，每消费1分钟休息10ms。 long interval = System.currentTimeMillis() - beginTime; if (interval > MAX_TIME_CONSUME_CONTINUOUSLY) { ConsumeMessageOrderlyService.this.submitConsumeRequestLater(processQueue, messageQueue, 10); break; } // 获取消费消息。此处和并发消息请求不同，并发消息请求已经带了消费哪些消息。 final int consumeBatchSize = ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize(); List<MessageExt> msgs = this.processQueue.takeMessags(consumeBatchSize); if (!msgs.isEmpty()) { final ConsumeOrderlyContext context = new ConsumeOrderlyContext(this.messageQueue); ConsumeOrderlyStatus status = null; // Hook：before ConsumeMessageContext consumeMessageContext = null; if (ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.hasHook()) { consumeMessageContext = new ConsumeMessageContext(); consumeMessageContext .setConsumerGroup(ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumerGroup()); consumeMessageContext.setMq(messageQueue); consumeMessageContext.setMsgList(msgs); consumeMessageContext.setSuccess(false); // init the consume context type consumeMessageContext.setProps(new HashMap<String, String>()); ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.executeHookBefore(consumeMessageContext); } // 执行消费 long beginTimestamp = System.currentTimeMillis(); ConsumeReturnType returnType = ConsumeReturnType.SUCCESS; boolean hasException = false; try { this.processQueue.getLockConsume().lock(); // 锁定队列消费锁 if (this.processQueue.isDropped()) { log.warn("consumeMessage, the message queue not be able to consume, because its dropped. { }", this.messageQueue); break; } status = messageListener.consumeMessage(Collections.unmodifiableList(msgs), context); } catch (Throwable e) { log.warn("consumeMessage exception: { } Group: { } Msgs: { } MQ: { }", // RemotingHelper.exceptionSimpleDesc(e), // ConsumeMessageOrderlyService.this.consumerGroup, // msgs, // messageQueue); hasException = true; } finally { this.processQueue.getLockConsume().unlock(); // 锁定队列消费锁 } if (null == status // || ConsumeOrderlyStatus.ROLLBACK == status// || ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT == status) { log.warn("consumeMessage Orderly return not OK, Group: { } Msgs: { } MQ: { }", // ConsumeMessageOrderlyService.this.consumerGroup, // msgs, // messageQueue); } // 解析消费结果状态 long consumeRT = System.currentTimeMillis() - beginTimestamp; if (null == status) { if (hasException) { returnType = ConsumeReturnType.EXCEPTION; } else { returnType = ConsumeReturnType.RETURNNULL; } } else if (consumeRT >= defaultMQPushConsumer.getConsumeTimeout() * 60 * 1000) { returnType = ConsumeReturnType.TIME_OUT; } else if (ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT == status) { returnType = ConsumeReturnType.FAILED; } else if (ConsumeOrderlyStatus.SUCCESS == status) { returnType = ConsumeReturnType.SUCCESS; } if (ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.hasHook()) { consumeMessageContext.getProps().put(MixAll.CONSUME_CONTEXT_TYPE, returnType.name()); } if (null == status) { status = ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT; } // Hook：after if (ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.hasHook()) { consumeMessageContext.setStatus(status.toString()); consumeMessageContext .setSuccess(ConsumeOrderlyStatus.SUCCESS == status || ConsumeOrderlyStatus.COMMIT == status); ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.executeHookAfter(consumeMessageContext); } ConsumeMessageOrderlyService.this.getConsumerStatsManager() .incConsumeRT(ConsumeMessageOrderlyService.this.consumerGroup, messageQueue.getTopic(), consumeRT); // 处理消费结果 continueConsume = ConsumeMessageOrderlyService.this.processConsumeResult(msgs, status, context, this); } else { continueConsume = false; } } } else { if (this.processQueue.isDropped()) { log.warn("the message queue not be able to consume, because its dropped. { }", this.messageQueue); return; } ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 100); } } }

获取到锁对象后，使用synchronized尝试申请线程级独占锁。

如果加锁成功，同一时刻只有一个线程进行消息消费。

如果加锁失败，会延迟100ms重新尝试向broker端申请锁定messageQueue，锁定成功后重新提交消费请求

至此，第三个关键点的解决思路也清晰了，基本上就两个步骤。

创建消息拉取任务时，消息客户端向broker端申请锁定MessageQueue，使得一个MessageQueue同一个时刻只能被一个消费客户端消费。

消息消费时，多线程针对同一个消息队列的消费先尝试使用synchronized申请独占锁，加锁成功才能进行消费，使得一个MessageQueue同一个时刻只能被一个消费客户端中一个线程消费。

【顺序消费问题拆解】

broke 上要保证一个队列只有一个进程消费，即一个队列同一时间只有一个consumer 消费 broker 给consumer 的消息顺序应该保持一致，这个通过 rpc传输，序列化后消息顺序不变，所以很容易实现 consumer 上的队列消息要保证同一个时间只有一个线程消费

通过问题的拆分，问题变成同一个共享资源串行处理了，要解决这个问题，通常的做法都是访问资源的时候加锁，即broker 上一个队列消息在被consumer 访问的必须加锁，单个consumer 端多线程并发处理消息的时候需要加锁;这里还需要考虑broker 锁的异常情况，假如一个broke 队列上的消息被consumer 锁住了，万一consumer 崩溃了，这个锁就释放不了，所以broker 上的锁需要加上锁的过期时间。

实际上 RocketMQ 消费端也就是照着上面的思路做：

RocketMQ中顺序消息注意事项

实际项目中并不是所有情况都需要用到顺序消息，但这也是设计方案的时候容易忽略的一点

顺序消息是生产者和消费者配合协调作用的结果，但是消费端保证顺序消费，是保证不了顺序消息的

消费端并行方式消费，只设置一次拉取消息的数量为 1(即配置参数 consumeBatchSize )，是否可以实现顺序消费 ?这里实际是不能的，并发消费在消费端有多个线程同时消费，consumeBatchSize 只是一个线程一次拉取消息的数量，对顺序消费没有意义，这里大家有兴趣可以看 ConsumeMessageConcurrentlyService 的代码，并发消费的逻辑都在哪里。

在使用顺序消息时，一定要注意其异常情况的出现，对于顺序消息，当消费者消费消息失败后，消息队列 RocketMQ 版会自动不断地进行消息重试(每次间隔时间为 1 秒)，重试最大值是Integer.MAX_VALUE.这时，应用会出现消息消费被阻塞的情况。因此，建议您使用顺序消息时，务必保证应用能够及时监控并处理消费失败的情况，避免阻塞现象的发生。

重要的事再强调一次：在使用顺序消息时，一定要注意其异常情况的出现!防止资源不释放!