RocketMQ源码：有序消息分析

本文主要分析RocketMQ中如何保证消息有序的。

RocketMQ的版本为：4.2.0 release。

一.时序图

还是老规矩，先把分析过程的时序图摆出来：

1.Producer发送顺序消息

2.Consumer接收顺序消息(一)

3.Consumer接收顺序消息(二)

二.源码分析 - Producer发送顺序消息

1 DefaultMQProducer#send：发送消息，入参中有自定义的消息队列选择器。

// DefaultMQProducer#send
public SendResult send(Message msg, MessageQueueSelector selector, Object arg)
throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
return this.defaultMQProducerImpl.send(msg, selector, arg);
}

1.1 DefaultMQProducerImpl#makeSureStateOK：确保Producer的状态是运行状态-ServiceState.RUNNING。

// DefaultMQProducerImpl#makeSureStateOK
private void makeSureStateOK() throws MQClientException {
if (this.serviceState != ServiceState.RUNNING) {
throw new MQClientException("The producer service state not OK, "+ this.serviceState
+ FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
null);
}
}

1.2 DefaultMQProducerImpl#tryToFindTopicPublishInfo：根据Topic获取发布Topic用到的路由信息。

// DefaultMQProducerImpl#tryToFindTopicPublishInfo
private TopicPublishInfo tryToFindTopicPublishInfo(final String topic) {
TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);
if (null == topicPublishInfo || !topicPublishInfo.ok()) {
this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo());
this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);// 为空则从 NameServer更新获取，false，不传入 defaultMQProducer
topicPublishInfo = this.topicPublishInfoTable.get(topic);
}
if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) {// 有了路由信息而且状态OK，则返回
return topicPublishInfo;
} else {
this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer);
topicPublishInfo = this.topicPublishInfoTable.get(topic);
return topicPublishInfo;
}
}

1.3 调用自定义消息队列选择器的select方法。

// DefaultMQProducerImpl#sendSelectImpl
MessageQueue mq = null;
try {
mq = selector.select(topicPublishInfo.getMessageQueueList(), msg, arg);
} catch (Throwable e) {
throw new MQClientException("select message queue throwed exception.", e);
}
// Producer#main
SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
@Override
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
Integer id = (Integer) arg;
int index = id % mqs.size();
return mqs.get(index);
}
}, orderId);

1.4 DefaultMQProducerImpl#sendKernelImpl：发送消息的核心实现方法。

// DefaultMQProducerImpl#sendKernelImpl
......
switch (communicationMode) {
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;
......

1.4.1 MQClientAPIImpl#sendMessage：发送消息。

// MQClientAPIImpl#sendMessage
......
switch (communicationMode) {// 根据发送消息的模式(同步/异步)选择不同的方式，默认是同步
case SYNC:
long costTimeSync = System.currentTimeMillis() - beginStartTime;
if (timeoutMillis < costTimeSync) {
throw new RemotingTooMuchRequestException("sendMessage call timeout");
}
return this.sendMessageSync(addr, brokerName, msg, timeoutMillis - costTimeSync, request);
......

1.4.1.1 MQClientAPIImpl#sendMessageSync：发送同步消息。

// MQClientAPIImpl#sendMessageSync
private SendResult sendMessageSync(
final String addr,
final String brokerName,
final Message msg,
final long timeoutMillis,
final RemotingCommand request
) throws RemotingException, MQBrokerException, InterruptedException {
RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
assert response != null;
return this.processSendResponse(brokerName, msg, response);
}

1.4.1.1.1 NettyRemotingClient#invokeSync：构造RemotingCommand，调用的方式是同步。

// NettyRemotingClient#invokeSync
RemotingCommand response = this.invokeSyncImpl(channel, request, timeoutMillis - costTime);
if (this.rpcHook != null) {
this.rpcHook.doAfterResponse(RemotingHelper.parseChannelRemoteAddr(channel), request, response);
}
return response;

三.源码分析 - Consumer接收顺序消息(一)

1 DefaultMQPushConsumer#registerMessageListener：把Consumer传入的消息监听器加入到messageListener中。

// DefaultMQPushConsumer#registerMessageListener
public void registerMessageListener(MessageListenerOrderly messageListener) {
this.messageListener = messageListener;
this.defaultMQPushConsumerImpl.registerMessageListener(messageListener);
}

1.1 DefaultMQPushConsumerImpl#registerMessageListener：把Consumer传入的消息监听器加入到messageListenerInner中。

// DefaultMQPushConsumerImpl#registerMessageListener
public void registerMessageListener(MessageListener messageListener) {
this.messageListenerInner = messageListener;
}

2 DefaultMQPushConsumer#start：启动Consumer。

// DefaultMQPushConsumer#start
public void start() throws MQClientException {
this.defaultMQPushConsumerImpl.start();
}

2.1 DefaultMQPushConsumerImpl#start：启动ConsumerImpl。

// DefaultMQPushConsumerImpl#start
switch (this.serviceState) {
case CREATE_JUST:// 刚刚创建
......
if (this.getMessageListenerInner() instanceof MessageListenerOrderly) {// 有序消息服务
this.consumeOrderly = true;
this.consumeMessageService = new ConsumeMessageOrderlyService(this, (MessageListenerOrderly) this.getMessageListenerInner());
} else if (this.getMessageListenerInner() instanceof MessageListenerConcurrently) {// 并发无序消息服务
this.consumeOrderly = false;
this.consumeMessageService = new ConsumeMessageConcurrentlyService(this, (MessageListenerConcurrently) this.getMessageListenerInner());
}
......
this.consumeMessageService.start();// 启动消息服务
......
mQClientFactory.start();// 启动MQClientInstance
......

2.1.1 new ConsumeMessageOrderlyService()：构造顺序消息服务。

// ConsumeMessageOrderlyService#ConsumeMessageOrderlyService
public ConsumeMessageOrderlyService(DefaultMQPushConsumerImpl defaultMQPushConsumerImpl,
MessageListenerOrderly messageListener) {
this.defaultMQPushConsumerImpl = defaultMQPushConsumerImpl;
this.messageListener = messageListener;
this.defaultMQPushConsumer = this.defaultMQPushConsumerImpl.getDefaultMQPushConsumer();
this.consumerGroup = this.defaultMQPushConsumer.getConsumerGroup();
this.consumeRequestQueue = new LinkedBlockingQueue<Runnable>();
this.consumeExecutor = new ThreadPoolExecutor(// 主消息消费线程池，正常执行收到的ConsumeRequest。多线程
this.defaultMQPushConsumer.getConsumeThreadMin(),
this.defaultMQPushConsumer.getConsumeThreadMax(),
1000 * 60,
TimeUnit.MILLISECONDS,
this.consumeRequestQueue,
new ThreadFactoryImpl("ConsumeMessageThread_"));
this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("ConsumeMessageScheduledThread_"));
}

2.1.2 ConsumeMessageOrderlyService#start：启动消息队列客户端实例。

// DefaultMQPushConsumerImpl#start
this.consumeMessageService.start();
// ConsumeMessageOrderlyService#start
public void start() {
if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())) {
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
ConsumeMessageOrderlyService.this.lockMQPeriodically();// 定时向broker发送批量锁住当前正在消费的队列集合的消息
}
}, 1000 * 1, ProcessQueue.REBALANCE_LOCK_INTERVAL, TimeUnit.MILLISECONDS);
}
}

2.1.2.1 ConsumeMessageOrderlyService#lockMQPeriodically：定时向broker发送批量锁住当前正在消费的队列集合的消息。

2.1.2.1.1 RebalanceImpl#lockAll：锁住所有正在消息的队列。

// ConsumeMessageOrderlyService#lockMQPeriodically
if (!this.stopped) {
this.defaultMQPushConsumerImpl.getRebalanceImpl().lockAll();
}
// RebalanceImpl#lockAll
HashMap<String, Set<MessageQueue>> brokerMqs = this.buildProcessQueueTableByBrokerName();// 根据brokerName从processQueueTable获取正在消费的队列集合
......
Set<MessageQueue> lockOKMQSet = this.mQClientFactory.getMQClientAPIImpl().lockBatchMQ(findBrokerResult.getBrokerAddr(), requestBody, 1000);// 向Broker发送锁住消息队列的指令
for (MessageQueue mq : lockOKMQSet) {
ProcessQueue processQueue = this.processQueueTable.get(mq);
if (processQueue != null) {
if (!processQueue.isLocked()) {
log.info("the message queue locked OK, Group: {} {}", this.consumerGroup, mq);
}
processQueue.setLocked(true);
processQueue.setLastLockTimestamp(System.currentTimeMillis());
}
}
......

2.1.3 MQClientInstance#start：启动MQClientInstance。过程较复杂，放到大标题四中分析。

// DefaultMQPushConsumerImpl#start
mQClientFactory.start();

四.源码分析 - Consumer接收顺序消息(二)

1 MQClientInstance#start：启动客户端实例MQClientInstance。

// MQClientInstance#start
synchronized (this) {
switch (this.serviceState) {
case CREATE_JUST:
......
// Start pull service 启动拉取消息服务
this.pullMessageService.start();
// Start rebalance service 启动消费端负载均衡服务
this.rebalanceService.start();
......

1.1 PullMessageService#run：启动拉取消息服务。实际调用的是DefaultMQPushConsumerImpl的pullMessage方法。

// PullMessageService#run
public void run() {
log.info(this.getServiceName() + " service started");
while (!this.isStopped()) {
try {
PullRequest pullRequest = this.pullRequestQueue.take();
this.pullMessage(pullRequest);
} catch (InterruptedException ignored) {
} catch (Exception e) {
log.error("Pull Message Service Run Method exception", e);
}
}
log.info(this.getServiceName() + " service end");
}
// PullMessageService#pullMessage
private void pullMessage(final PullRequest pullRequest) {
final MQConsumerInner consumer = this.mQClientFactory.selectConsumer(pullRequest.getConsumerGroup());
if (consumer != null) {
DefaultMQPushConsumerImpl impl = (DefaultMQPushConsumerImpl) consumer;
impl.pullMessage(pullRequest);// 调用DefaultMQPushConsumerImpl的pullMessage
} else {
log.warn("No matched consumer for the PullRequest {}, drop it", pullRequest);
}
}

1.1.1.1 DefaultMQPushConsumerImpl#pullMessage：拉取消息。提交到ConsumeMessageOrderlyService的线程池consumeExecutor中执行。

// DefaultMQPushConsumerImpl#pullMessage
......
PullCallback pullCallback = new PullCallback() {
@Override
public void onSuccess(PullResult pullResult) {
switch (pullResult.getPullStatus()) {
case FOUND:
long prevRequestOffset = pullRequest.getNextOffset();
pullRequest.setNextOffset(pullResult.getNextBeginOffset());
long pullRT = System.currentTimeMillis() - beginTimestamp;
......
DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(
pullResult.getMsgFoundList(),
processQueue,
pullRequest.getMessageQueue(),
dispatchToConsume);
......

1.1.1.1.1.1.1 ConsumeRequest#run：处理消息消费的线程。

// ConsumeMessageOrderlyService.ConsumeRequest#run
List<MessageExt> msgs = this.processQueue.takeMessags(consumeBatchSize);
......
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 it's dropped. {}",this.messageQueue);
break;
}
status = messageListener.consumeMessage(Collections.unmodifiableList(msgs), context);// 实际消费消息的地方，回调消息监听器的consumeMessage方法
} 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();
}
......

1.2 RebalanceService#run：启动消息端负载均衡服务。

// RebalanceService#run
public void run() {
log.info(this.getServiceName() + " service started");
while (!this.isStopped()) {
this.waitForRunning(waitInterval);
this.mqClientFactory.doRebalance();
}
log.info(this.getServiceName() + " service end");
}
// MQClientInstance#doRebalance
public void doRebalance() {
for (Map.Entry<String, MQConsumerInner> entry : this.consumerTable.entrySet()) {
MQConsumerInner impl = entry.getValue();
if (impl != null) {
try {
impl.doRebalance();
} catch (Throwable e) {
log.error("doRebalance exception", e);
}
}
}
}
// DefaultMQPushConsumerImpl#doRebalance
public void doRebalance() {
if (!this.pause) {
this.rebalanceImpl.doRebalance(this.isConsumeOrderly());
}
}

1.2.1.1.1 RebalanceImpl#doRebalance：负载均衡服务类处理。

// RebalanceImpl#doRebalance
public void doRebalance(final boolean isOrder) {
Map<String, SubscriptionData> subTable = this.getSubscriptionInner();
if (subTable != null) {
for (final Map.Entry<String, SubscriptionData> entry : subTable.entrySet()) {
final String topic = entry.getKey();
try {
this.rebalanceByTopic(topic, isOrder);
} catch (Throwable e) {
if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
log.warn("rebalanceByTopic Exception", e);
}
}
}
}
this.truncateMessageQueueNotMyTopic();
}
// RebalanceImpl#rebalanceByTopic
switch (messageModel) {
case BROADCASTING: {
Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
if (mqSet != null) {
boolean changed = this.updateProcessQueueTableInRebalance(topic, mqSet, isOrder);// 根据Toipc去除queue
if (changed) {
this.messageQueueChanged(topic, mqSet, mqSet);
log.info("messageQueueChanged {} {} {} {}",
consumerGroup,
topic,
mqSet,
mqSet);
}
} else {
......
// RebalanceImpl#updateProcessQueueTableInRebalance
this.dispatchPullRequest(pullRequestList);// RebalancePushImpl分发消息

1.2.1.1.1.1.1.1 RebalancePushImpl#dispatchPullRequest：RebalancePushImpl分发。

// RebalancePushImpl#dispatchPullRequest
public void dispatchPullRequest(List<PullRequest> pullRequestList) {
for (PullRequest pullRequest : pullRequestList) {
this.defaultMQPushConsumerImpl.executePullRequestImmediately(pullRequest);
log.info("doRebalance, {}, add a new pull request {}", consumerGroup, pullRequest);
}
}

五.总结

相比Producer的发送流程，Consumer的接收流程稍微复杂一点。通过上面的源码分析，可以知道RocketMQ是怎样保证消息的有序的：

1.通过ReblanceImp的lockAll方法，每隔一段时间定时锁住当前消费端正在消费的队列。设置本地队列ProcessQueue的locked属性为true。保证broker中的每个消息队列只对应一个消费端；

2.另外，消费端也是通过锁，保证每个ProcessQueue只有一个线程消费。