当前位置：首页 > news >正文

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

news 文章来源：https://blog.csdn.net/qq_33291299/article/details/129064794 2025/1/31 17:03:06

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，继续消息发送的第二个步骤，选取一个消息队列。

1、Broker概述

2、Broker存储设计概要

2.1 消息发送

2.1.1 消息发送参数详解：

2.2.2 消息发送流程

相关文章：