RocketMQ源码分析之Broker概述与同步消息发送原理与高可用设计及思考

1、Broker概述

Broker 在 RocketMQ 架构中的角色，就是存储消息，核心任务就是持久化消息，生产者发送消息给 Broker,消费者从 Broker 消费消息，其物理部署架构图如下：

备注：以上摘录自官方 RocketMQ 设计文档。

上述基本描述了消息中间件的架构设计，不仅限于 RocketMQ,不同消息中间件的最大区别之一在消息的存储上。

2、Broker存储设计概要

接下来从配置文件的角度来窥探 Broker 存储设计的关注点，对应代码（MessageStoreConfig）。

• storePathRootDir
  设置Broker的存储根目录，默认为 $Broker_Home/store。
• storePathCommitLog
  设置commitlog的存储目录，默认为$Broker_Home/store/commitlog。
• mapedFileSizeCommitLog
  commitlog 文件的大小，默认为1G。
• mapedFileSizeConsumeQueue
  consumeQueueSize，ConsumeQueue 存放的是定长的信息（20个字节，偏移量、size、tagscode）,默认30w * ConsumeQueue.CQ_STORE_UNIT_SIZE。
• enableConsumeQueueExt
  是否开启 consumeQueueExt,默认为 false,就是如果消费端消息消费速度跟不上，是否创建一个扩展的 ConsumeQueue文件，如果不开启，应该会阻塞从 commitlog 文件中获取消息，并且 ConsumeQueue,应该是按topic独立的。
• mappedFileSizeConsumeQueueExt
  扩展consume文件的大小，默认为48M。
• flushIntervalCommitLog
  刷写 CommitLog 的间隔时间，RocketMQ 后台会启动一个线程，将消息刷写到磁盘，这个也就是该线程每次运行后等待的时间，默认为500毫秒。flush 操作，调用文件通道的force()方法。
• commitIntervalCommitLog
  提交消息到 CommitLog 对应的文件通道的间隔时间，原理与上面类似；将消息写入到文件通道（调用FileChannel.write方法）得到最新的写指针，默认为200毫秒。
• useReentrantLockWhenPutMessage
  在put message( 将消息按格式封装成msg放入相关队列时实用的锁机制：自旋或ReentrantLock)。
• flushIntervalConsumeQueue
  刷写到ConsumeQueue的间隔，默认为1s。
• flushCommitLogLeastPages
  每次 flush commitlog 时最小发生变化的页数。
• commitCommitLogLeastPages
  每一次 commitlog 提交任务至少需要的页数。
• flushLeastPagesWhenWarmMapedFile
  用字节0填充整个文件，每多少页刷盘一次，默认4096，异步刷盘模式生效。
• flushConsumeQueueLeastPages
  一次刷盘至少需要的脏页数量，默认为2，针对 consuequeue 文件。
• putMsgIndexHightWater
  当前版本未使用。

接下来从如下方面去深入其实现：

1）生产者发送消息

2）消息协议（格式）

3）消息存储、检索

4）消费队列维护

5）消息消费、重试等机制

2.1 消息发送

org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl sendDefaultImpl方法源码分析
rprivate SendResult sendDefaultImpl(//Message msg, //    final CommunicationMode communicationMode, //final SendCallback sendCallback, //final long timeout//) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {

2.1.1 消息发送参数详解：

1、Message msg

2、communicationMode communicationMode

发送方式，SYNC(同步)、ASYNC（异步）、ONEWAY(单向，不关注返回)

3、SendCallback sendCallback

异步消息发送回调函数。

4、long timeout

消息发送超时时间。

2.2.2 消息发送流程

默认消息发送实现：

private SendResult sendDefaultImpl(//Message msg, //final CommunicationMode communicationMode, //final SendCallback sendCallback, //final long timeout//) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {this.makeSureStateOK();Validators.checkMessage(msg, this.defaultMQProducer);final long invokeID = random.nextLong();long beginTimestampFirst = System.currentTimeMillis();long beginTimestampPrev = beginTimestampFirst;long endTimestamp = beginTimestampFirst;TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic()); // @1if (topicPublishInfo != null && topicPublishInfo.ok()) {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];for (; times < timesTotal; times++) {String lastBrokerName = null == mq ? null : mq.getBrokerName();MessageQueue tmpmq = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName); // @2if (tmpmq != null) {mq = tmpmq;brokersSent[times] = mq.getBrokerName();try {beginTimestampPrev = System.currentTimeMillis();sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout);  // @3endTimestamp = 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) {if (this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) {continue;}}return sendResult;default:break;}} catch (RemotingException e) {endTimestamp = System.currentTimeMillis();this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true); // @4log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);log.warn(msg.toString());exception = e;continue;} catch (MQClientException e) {endTimestamp = System.currentTimeMillis();this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true); log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);log.warn(msg.toString());exception = e;continue;} catch (MQBrokerException e) {endTimestamp = System.currentTimeMillis();this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);log.warn(msg.toString());exception = e;switch (e.getResponseCode()) {case ResponseCode.TOPIC_NOT_EXIST:case ResponseCode.SERVICE_NOT_AVAILABLE:case ResponseCode.SYSTEM_ERROR:case ResponseCode.NO_PERMISSION:case ResponseCode.NO_BUYER_ID:case ResponseCode.NOT_IN_CURRENT_UNIT:continue;default:if (sendResult != null) {return sendResult;}throw e;}} catch (InterruptedException e) {endTimestamp = System.currentTimeMillis();this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);log.warn(String.format("sendKernelImpl exception, throw exception, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mq), e);log.warn(msg.toString());log.warn("sendKernelImpl exception", e);log.warn(msg.toString());throw e;}} else {break;}}if (sendResult != null) {return sendResult;}String 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 += FAQUrl.suggestTodo(FAQUrl.SEND_MSG_FAILED);MQClientException mqClientException = new MQClientException(info, exception);if (exception instanceof MQBrokerException) {mqClientException.setResponseCode(((MQBrokerException) exception).getResponseCode());} else if (exception instanceof RemotingConnectException) {mqClientException.setResponseCode(ClientErrorCode.CONNECT_BROKER_EXCEPTION);} else if (exception instanceof RemotingTimeoutException) {mqClientException.setResponseCode(ClientErrorCode.ACCESS_BROKER_TIMEOUT);} else if (exception instanceof MQClientException) {mqClientException.setResponseCode(ClientErrorCode.BROKER_NOT_EXIST_EXCEPTION);}throw mqClientException;}List<String> nsList = this.getmQClientFactory().getMQClientAPIImpl().getNameServerAddressList();if (null == nsList || nsList.isEmpty()) {throw new MQClientException("No name server address, please set it." + FAQUrl.suggestTodo(FAQUrl.NAME_SERVER_ADDR_NOT_EXIST_URL), null).setResponseCode(ClientErrorCode.NO_NAME_SERVER_EXCEPTION);}throw new MQClientException("No route info of this topic, " + msg.getTopic() + FAQUrl.suggestTodo(FAQUrl.NO_TOPIC_ROUTE_INFO),null).setResponseCode(ClientErrorCode.NOT_FOUND_TOPIC_EXCEPTION);

主要的核心步骤如下：

代码@1：获取topic的路由信息。

代码@2：根据topic负载均衡算法选择一个MessageQueue。

代码@3：向 MessageQueue 发送消息。

代码@4：更新失败策略,主要用于规避发生故障的 broker，下文会详细介绍。

代码@5：如果是同步调用方式(SYNC),则执行失败重试策略，默认重试两次。

2、2.2.1 获取topic的路由信息

首先我们来思考一下，topic 的路由信息包含哪些内容。

消息的发布与订阅基于topic,路由发布信息以 topic 维度进行描述。

Broker 负载消息存储，一个 topic 可以分布在多台 Broker 上(负载均衡)，每个 Broker 包含多个 Queue。队列元数据基于Broker来描述（QueueData：所在 BrokerName、读队列个数、写队列个数、权限、同步或异步）。

接下来从源码分析 tryToFindTopicPublishInfo方法，详细了解获取 Topic 的路由信息。

DefaultMQProducerImpl#tryToFindTopicPublishInfo

private TopicPublishInfo tryToFindTopicPublishInfo(final String topic) {TopicPublishInfo topicPublishInfo = this.topicPublishInfoTable.get(topic);        // @1if (null == topicPublishInfo || !topicPublishInfo.ok()) {this.topicPublishInfoTable.putIfAbsent(topic, new TopicPublishInfo());this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);          // @2topicPublishInfo = this.topicPublishInfoTable.get(topic);}if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) {            //@3return topicPublishInfo;} else {this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer);      //@4topicPublishInfo = this.topicPublishInfoTable.get(topic);return topicPublishInfo;}

代码@1：从本地缓存(ConcurrentMap< String/* topic */, TopicPublishInfo>)中尝试获取，第一次肯定为空，走代码@2的流程。

代码@2：尝试从 NameServer 获取配置信息并更新本地缓存配置。

代码@3：如果找到可用的路由信息并返回。

代码@4：如果未找到路由信息，则再次尝试使用默认的 topic 去找路由配置信息。

接下来我们重点关注updateTopicRouteInfoFromNameServer方法。

MQClientInstance#updateTopicRouteInfoFromNameServer

public boolean updateTopicRouteInfoFromNameServer(final String topic, boolean isDefault, DefaultMQProducer defaultMQProducer) {try {if (this.lockNamesrv.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {     // @1try {TopicRouteData topicRouteData;if (isDefault && defaultMQProducer != null) {      //@2topicRouteData = this.mQClientAPIImpl.getDefaultTopicRouteInfoFromNameServer(defaultMQProducer.getCreateTopicKey(),1000 * 3);if (topicRouteData != null) {for (QueueData data : topicRouteData.getQueueDatas()) {int queueNums = Math.min(defaultMQProducer.getDefaultTopicQueueNums(), data.getReadQueueNums());data.setReadQueueNums(queueNums);data.setWriteQueueNums(queueNums);}}} else {topicRouteData = this.mQClientAPIImpl.getTopicRouteInfoFromNameServer(topic, 1000 * 3);    //@3}if (topicRouteData != null) {TopicRouteData old = this.topicRouteTable.get(topic);     //@4boolean changed = topicRouteDataIsChange(old, topicRouteData);    //@5if (!changed) {changed = this.isNeedUpdateTopicRouteInfo(topic);                        //@6} else {log.info("the topic[{}] route info changed, old[{}] ,new[{}]", topic, old, topicRouteData);}if (changed) {    //@7TopicRouteData cloneTopicRouteData = topicRouteData.cloneTopicRouteData();for (BrokerData bd : topicRouteData.getBrokerDatas()) {this.brokerAddrTable.put(bd.getBrokerName(), bd.getBrokerAddrs());}// Update Pub info     //@8{TopicPublishInfo publishInfo = topicRouteData2TopicPublishInfo(topic, topicRouteData);publishInfo.setHaveTopicRouterInfo(true);Iterator<Entry<String, MQProducerInner>> it = this.producerTable.entrySet().iterator();while (it.hasNext()) {Entry<String, MQProducerInner> entry = it.next();MQProducerInner impl = entry.getValue();if (impl != null) {impl.updateTopicPublishInfo(topic, publishInfo);}}}// Update sub info    //@9{Set<MessageQueue> subscribeInfo = topicRouteData2TopicSubscribeInfo(topic, topicRouteData);Iterator<Entry<String, MQConsumerInner>> it = this.consumerTable.entrySet().iterator();while (it.hasNext()) {Entry<String, MQConsumerInner> entry = it.next();MQConsumerInner impl = entry.getValue();if (impl != null) {impl.updateTopicSubscribeInfo(topic, subscribeInfo);}}}log.info("topicRouteTable.put. Topic = {}, TopicRouteData[{}]", topic, cloneTopicRouteData);this.topicRouteTable.put(topic, cloneTopicRouteData);return true;}} else {log.warn("updateTopicRouteInfoFromNameServer, getTopicRouteInfoFromNameServer return null, Topic: {}", topic);}} catch (Exception e) {if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX) && !topic.equals(MixAll.DEFAULT_TOPIC)) {log.warn("updateTopicRouteInfoFromNameServer Exception", e);}} finally {this.lockNamesrv.unlock();}} else {log.warn("updateTopicRouteInfoFromNameServer tryLock timeout {}ms", LOCK_TIMEOUT_MILLIS);}} catch (InterruptedException e) {log.warn("updateTopicRouteInfoFromNameServer Exception", e);}return false;

代码@1：为了避免重复从 NameServer 获取配置信息，在这里使用了ReentrantLock,并且设有超时时间。固定为3000s。

代码@2，@3的区别，一个是获取默认 topic 的配置信息，一个是获取指定 topic 的配置信息，该方法在这里就不跟踪进去了，具体的实现就是通过与 NameServer 的长连接 Channel 发送 GET_ROUTEINTO_BY_TOPIC (105)命令，获取配置信息。注意，次过程的超时时间为3s，由此可见，NameServer的实现要求高效。

代码@4、@5、@6：从这里开始，拿到最新的 topic 路由信息后，需要与本地缓存中的 topic 发布信息进行比较，如果有变化，则需要同步更新发送者、消费者关于该 topic 的缓存。

代码@7：更新发送者的缓存。

代码@8：更新订阅者的缓存（消费队列信息）。

至此 tryToFindTopicPublishInfo 运行完毕，从 NameServe r获取 TopicPublishData，继续消息发送的第二个步骤，选取一个消息队列。