回顾
上文中我们对 RocketMQ 的 顺序消息 进行了 spring cloud 版本的演示，这里再回顾一下：
顺序消息分为 分区顺序消息 和 全局顺序消息。
全局顺序消息其实是分区顺序消息的一种特殊情况，即如果只有一个分区且同一时间只有一个消费者线程进行消费，那么就可以看作是全局顺序消息。
在 RocketMQ 创建 topic 时默认队列（分区）数量是：8 ，是针对所有 topic 的
如果要单独设置一个 topic 的队列（分区）数量，在 spring cloud alibaba 中可以这样配置：
spring: application: name: mq-example cloud: stream: bindings: # 定义 name 为 input 的 binding 消费 input: content-type: application/json destination: test-topic3 group: consumer-group # 定义 name 为 output 的 binding 生产 output-order: content-type: application/json destination: test-topic3 # Producer 配置项，对应 ProducerProperties 类 producer: partitionCount: 1 # 分区数量
注意这里 partitionCount，如将该值设置为 1，则 broker 会将消息发送到同一个分区中。
原理
本文我们重点了解一下，RocketMQ 的顺序消息的实现原理。
在 MQ 的模型中，顺序需要由 3 个阶段去保障：
- 消息被发送时保持顺序
- 消息被存储时保持和发送的顺序一致
- 消息被消费时保持和存储的顺序一致
RocketMQ 要想实现顺序消息，核心就是 Producer 同步发送，确保一组顺序消息被发送到同一个分区队列，然后 Consumer 确保同一个队列只被一个线程消费
发送有序
这里我们串一下代码，看一下 producer 发送消息的时候是怎么实现顺序发送的：
private SendResult sendDefaultImpl(Message msg, CommunicationMode communicationMode, SendCallback sendCallback, long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { this.makeSureStateOK(); Validators.checkMessage(msg, this.defaultMQProducer); long invokeID = this.random.nextLong(); long beginTimestampFirst = System.currentTimeMillis(); long beginTimestampPrev = beginTimestampFirst; TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic()); if (topicPublishInfo != null && topicPublishInfo.ok()) { boolean callTimeout = false; MessageQueue mq = null; Exception exception = null; SendResult sendResult = null; int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1; int times = 0; String[] brokersSent = new String[timesTotal]; while(true) { label122: { String info; if (times < timesTotal) { info = null == mq ? null : mq.getBrokerName(); MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, info); if (mqSelected != null) { mq = mqSelected; brokersSent[times] = mqSelected.getBrokerName(); long endTimestamp; try { beginTimestampPrev = System.currentTimeMillis(); long costTime = beginTimestampPrev - beginTimestampFirst; if (timeout >= costTime) { sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime); endTimestamp = System.currentTimeMillis(); this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false); switch(communicationMode) { case ASYNC: return null; case ONEWAY: return null; case SYNC: if (sendResult.getSendStatus() == SendStatus.SEND_OK || !this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) { return sendResult; } default: break label122; } } callTimeout = true; } catch (RemotingException var26) { endTimestamp = System.currentTimeMillis(); this.updateFaultItem(mqSelected.getBrokerName(), endTimestamp - beginTimestampPrev, true); this.log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mqSelected), var26); this.log.warn(msg.toString()); exception = var26; break label122; } catch (MQClientException var27) { endTimestamp = System.currentTimeMillis(); this.updateFaultItem(mqSelected.getBrokerName(), endTimestamp - beginTimestampPrev, true); this.log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mqSelected), var27); this.log.warn(msg.toString()); exception = var27; break label122; } catch (MQBrokerException var28) { endTimestamp = System.currentTimeMillis(); this.updateFaultItem(mqSelected.getBrokerName(), endTimestamp - beginTimestampPrev, true); this.log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mqSelected), var28); this.log.warn(msg.toString()); exception = var28; switch(var28.getResponseCode()) { case 1: case 14: case 16: case 17: case 204: case 205: break label122; default: if (sendResult != null) { return sendResult; } throw var28; } } catch (InterruptedException var29) { endTimestamp = System.currentTimeMillis(); this.updateFaultItem(mqSelected.getBrokerName(), endTimestamp - beginTimestampPrev, false); this.log.warn(String.format("sendKernelImpl exception, throw exception, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mqSelected), var29); this.log.warn(msg.toString()); this.log.warn("sendKernelImpl exception", var29); this.log.warn(msg.toString()); throw var29; } } } if (sendResult != null) { return sendResult; } info = String.format("Send [%d] times, still failed, cost [%d]ms, Topic: %s, BrokersSent: %s", times, System.currentTimeMillis() - beginTimestampFirst, msg.getTopic(), Arrays.toString(brokersSent)); info = info + FAQUrl.suggestTodo("http://rocketmq.apache.org/docs/faq/"); MQClientException mqClientException = new MQClientException(info, (Throwable)exception); if (callTimeout) { throw new RemotingTooMuchRequestException("sendDefaultImpl call timeout"); } if (exception instanceof MQBrokerException) { mqClientException.setResponseCode(((MQBrokerException)exception).getResponseCode()); } else if (exception instanceof RemotingConnectException) { mqClientException.setResponseCode(10001); } else if (exception instanceof RemotingTimeoutException) { mqClientException.setResponseCode(10002); } else if (exception instanceof MQClientException) { mqClientException.setResponseCode(10003); } throw mqClientException; } ++times; } } else { List<String> nsList = this.getmQClientFactory().getMQClientAPIImpl().getNameServerAddressList(); if (null != nsList && !nsList.isEmpty()) { throw (new MQClientException("No route info of this topic, " + msg.getTopic() + FAQUrl.suggestTodo("http://rocketmq.apache.org/docs/faq/"), (Throwable)null)).setResponseCode(10005); } else { throw (new MQClientException("No name server address, please set it." + FAQUrl.suggestTodo("http://rocketmq.apache.org/docs/faq/"), (Throwable)null)).setResponseCode(10004); } } }
上面是消息发送的代码，下面梳理下主要流程：
消息发送时，先根据 Topic 从 Broker 拉取 TopicPublishInfo 信息，它里面包含了 Topic 下所有的 MessageQueue。
TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
private TopicPublishInfo tryToFindTopicPublishInfo(String topic) { TopicPublishInfo topicPublishInfo = (TopicPublishInfo)this.topicPublishInfoTable.get(topic); if (null == topicPublishInfo || !topicPublishInfo.ok()) { this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo()); this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic); topicPublishInfo = (TopicPublishInfo)this.topicPublishInfoTable.get(topic); } if (!topicPublishInfo.isHaveTopicRouterInfo() && !topicPublishInfo.ok()) { this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer); topicPublishInfo = (TopicPublishInfo)this.topicPublishInfoTable.get(topic); return topicPublishInfo; } else { return topicPublishInfo; } }
public class TopicPublishInfo { private boolean orderTopic = false; private boolean haveTopicRouterInfo = false; private List<MessageQueue> messageQueueList = new ArrayList(); private volatile ThreadLocalIndex sendWhichQueue = new ThreadLocalIndex(); private TopicRouteData topicRouteData; public TopicPublishInfo() { } ...
选取一个目标队列：
MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, info);
接着调用核心发送方法，将消息发送到 broker
sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);
必须使用同步发送，异步/单向发送都无法保证消息被有序写入队列
sendKernelImpl 方法中有同/异步的判断，这里应该是走的 case SYNC
case ASYNC: Message tmpMessage = msg; if (msgBodyCompressed) { tmpMessage = MessageAccessor.cloneMessage(msg); msg.setBody(prevBody); } long costTimeAsync = System.currentTimeMillis() - beginStartTime; if (timeout < costTimeAsync) { throw new RemotingTooMuchRequestException("sendKernelImpl call timeout"); } sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(brokerAddr, mq.getBrokerName(), tmpMessage, requestHeader, timeout - costTimeAsync, communicationMode, sendCallback, topicPublishInfo, this.mQClientFactory, this.defaultMQProducer.getRetryTimesWhenSendAsyncFailed(), context, this); break; case ONEWAY: case SYNC: long costTimeSync = System.currentTimeMillis() - beginStartTime; if (timeout < costTimeSync) { throw new RemotingTooMuchRequestException("sendKernelImpl call timeout"); } sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(brokerAddr, mq.getBrokerName(), msg, requestHeader, timeout - costTimeSync, communicationMode, context, this); break; default: assert false;
Producer 保证发送有序，只要保证相同 ShardingKey 的消息发送到同一队列即可，以 spring cloud stream 的实现为例，可以查看 PartitionHandler 类中的 determinePartition 方法
public int determinePartition(Message<?> message) { Object key = this.extractKey(message); int partition; if (this.producerProperties.getPartitionSelectorExpression() != null) { partition = (Integer)this.producerProperties.getPartitionSelectorExpression().getValue(this.evaluationContext, key, Integer.class); } else { partition = this.partitionSelectorStrategy.selectPartition(key, this.partitionCount); } return Math.abs(partition % this.partitionCount); }
可以看到 partition 的值如果之前配置了分区 key 表达式如：
producer: partition-key-expression: payload['id']
则值是表达式的值，如没有配置，则走默认策略，默认策略的实现取的是消息的 hash:
private static class DefaultPartitionSelector implements PartitionSelectorStrategy { private DefaultPartitionSelector() { } public int selectPartition(Object key, int partitionCount) { int hashCode = key.hashCode(); if (hashCode == -2147483648) { hashCode = 0; } return Math.abs(hashCode); } }
最后队列的选择是利用 partition 和队列（分区）总数取模后得到的结果。 这样就可以保证相同 ShardingKey 的消息发送到同一队列了。
整体的流程如下图：
消息发送后，由于队列本身的 FIFO 特性，它保存到 broker 端也是有序的。
消费有序
Consumer 默认是多线程并发消费同一个 MessageQueue 的，即使消息是顺序到达的，也不能保证消息顺序消费。
那么 RocketMQ 如何保证消息顺序消费呢？
与 producer 一样，我们按照 consumer 的流程串一下代码
consumer 启动时，在 MQClientInstance 类的 start 方法中进行了重平衡操作：
public void start() throws MQClientException { synchronized (this) { switch (this.serviceState) { case CREATE_JUST: this.serviceState = ServiceState.START_FAILED; // If not specified,looking address from name server if (null == this.clientConfig.getNamesrvAddr()) { this.mQClientAPIImpl.fetchNameServerAddr(); } // Start request-response channel this.mQClientAPIImpl.start(); // Start various schedule tasks this.startScheduledTask(); // Start pull service this.pullMessageService.start(); // Start rebalance service this.rebalanceService.start(); // Start push service this.defaultMQProducer.getDefaultMQProducerImpl().start(false); log.info("the client factory [{}] start OK", this.clientId); this.serviceState = ServiceState.RUNNING; break; case START_FAILED: throw new MQClientException("The Factory object[" + this.getClientId() + "] has been created before, and failed.", null); default: break; } } }
就是这一行 rebalanceService.start() ，它的目的是给自己分配 MessageQueue。要保证一个队列被一个消费者消费，那么消费者在进行消息拉取消费时就必须向 mq 服务器申请队列锁。如果申请到琐，则拉取消息，否则放弃消息拉取，等到下一个队列负载周期 (20s) 再试。
申请锁可以参考 RebalanceImpl类 updateProcessQueueTableInRebalance 和 lock 方法中的代码：
//如果是顺序消息，需要向 Broker 申请锁队列，加锁成功才开始消费。 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();
public boolean lock(final MessageQueue mq) { // 查找 Broker Master 主机地址 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());// 客户端实例 ID requestBody.getMqSet().add(mq);// 申请锁哪些队列 try { // 发送请求，Broker 返回锁住的队列集合 Set<MessageQueue> lockedMq = this.mQClientFactory.getMQClientAPIImpl().lockBatchMQ(findBrokerResult.getBrokerAddr(), requestBody, 1000); 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; }
这个锁是 Broker 维护的全局锁。
一旦加锁成功，就会开始构建 PullRequest 对象开始拉取消息，消息拉取部分的代码实现在 PullMessageService 中，拉取成功后，在 PullCallback 里会将拉取到的消息填充到 ProcessQueue，然后提交消费请求，让 ConsumeMessageOrderlyService 开始消费消息。
消费消息时，先获取 MessageQueue 的锁对象，然后通过 synchronized 关键字保证只有一个线程消费
对于顺序消息，当消费者消费消息失败后，消息队列 RocketMQ 会自动不断地进行消息重试（每次间隔时间为 1 秒），重试最大值是 Integer.MAX_VALUE
case SUSPEND_CURRENT_QUEUE_A_MOMENT: this.getConsumerStatsManager().incConsumeFailedTPS(consumerGroup, consumeRequest.getMessageQueue().getTopic(), msgs.size()); // 校验最大重试次数，默认 Integer.MAX_VALUE if (checkReconsumeTimes(msgs)) { // 标记消息等待重新消费 consumeRequest.getProcessQueue().makeMessageToConsumeAgain(msgs); // 提交消费请求，稍后重试 this.submitConsumeRequestLater( consumeRequest.getProcessQueue(), consumeRequest.getMessageQueue(), context.getSuspendCurrentQueueTimeMillis()); continueConsume = false; } else { commitOffset = consumeRequest.getProcessQueue().commit(); } break;
最后补充一点，在消费的过程中，会对处理队列 (ProccessQueue) 进行加锁，保证处理中的消息消费完成，发生队列负载后，其他消费者才能继续消费。
例如队列 q3 目前是分配给消费者 C2 进行消费，已将拉取了 32 条消息在线程池中处理，然后对消费者进行了扩容，分配给 C2 的 q3 队列，被分配给 C3 了，由于 C2 已将处理了一部分，位点信息还没有提交，如果 C3 立马去消费 q3 队列中的消息，那存在一部分数据会被重复消费，故在 C2 消费者在消费 q3 队列的时候，消息没有消费完成，那负载队列就不能丢弃该队列，就不会在 broker 端释放琐，其他消费者就无法从该队列消费，尽最大可能保证了消息的重复消费，保证顺序性语义
消费总结 ：
- 创建消息拉取任务时，消息客户端向 broker 端申请锁定 MessageQueue，使得一个 MessageQueue 同一个时刻只能被一个消费客户端消费
- 消息消费时，多线程针对同一个消息队列的消费先尝试使用 synchronized 申请独占锁，加锁成功才能进行消费，使得一个 MessageQueue 同一个时刻只能被一个消费客户端中一个线程消费
- RocketMQ 中每一个消费组一个单独的线程池并发消费拉取到的消息，即消费端是多线程消费。而顺序消费的并发度等于该消费者分配到的队列数。消费并行度理论上不会有太大问题，因为 MessageQueue 的数量可以调整。
- 在消费的过程中，会对处理队列 (ProccessQueue) 进行加锁，保证处理中的消息消费完成
- 顺序消息一旦消费失败，默认会一直重试，不会跳过，因为一旦跳过就失去顺序消息的语义了
顺序消息可能存在的问题
消息阻塞
在顺序性语义的要求下，如果一条消息没有被成功消费，下一条消息就不能被消费，否则就不是顺序消费了。一条消息失败，如果跳过去消费其他消息，那就违背了顺序消费的语义。
建议在使用顺序消息时，务必保证应用能够及时监控并处理消费失败的情况，避免阻塞现象的发生。可以提供一些策略，由用户根据错误类型来决定是否跳过，并且提供重试队列之类的功能，在跳过之后用户可以在“其他”地方重新消费到这条消息。
failover 失效
发送顺序消息无法利用集群的 Failover 特性，因为不能更换 MessageQueue 进行重试
