RocketMQ自动创建topic原理-TBW102

自动创建Topic原理介绍

RocketMQ在发送消息的时候，我们一般会先去Broker创建Topic信息，Producer在发送消息的时候会先去nameSrv拉取Topic信息，那么如果拉取不到Topic，会如何处理呢？这里RocketMQ提供一种自动创建Topic的机制。

Producer流程

先从本地缓存topicPublishInfoTable中获取Topic。
如果本地缓存不存在或者失效，则调用**mQClientFactory.updateTopicRouteInfoFromNameServer(topic)**取nameSrv获取Topic。
如果nameSrv也没有获取到Topic，最后调用**mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer)**获取默认TTBW102。

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);topicPublishInfo = this.topicPublishInfoTable.get(topic);}if (topicPublishInfo.isHaveTopicRouterInfo() || topicPublishInfo.ok()) {return topicPublishInfo;} else {// 如果没有Topic,并且自动创建topic，则先投递到TBW102主题，在Broker再创建topic           this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer);topicPublishInfo = this.topicPublishInfoTable.get(topic);return topicPublishInfo;}}

在调用org.apache.rocketmq.client.impl.factory.MQClientInstance#updateTopicRouteInfoFromNameServer(java.lang.String, boolean, org.apache.rocketmq.client.producer.DefaultMQProducer)的时候，第二个参数isDefault为true标识拉取TBW102，这时候如果Broker开启了自动创建Topic，那么TBW102是肯定可以拉取成功的。

 if (isDefault && defaultMQProducer != null) {// 拉取TBW102topicRouteData = 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);}

Broker流程

Broker在启动初始化TopicConfigManager的时候，autoCreateTopicEnable配置为true（默认也是开启的，但是生产上一般为了防止Topic随意创建和安全性，一般会选择关闭），在初始化Broker的时候会自动创建TBW102.

 public TopicConfigManager(BrokerController brokerController) {this.brokerController = brokerController;// 忽略前面的代码{// MixAll.AUTO_CREATE_TOPIC_KEY_TOPICif (this.brokerController.getBrokerConfig().isAutoCreateTopicEnable()) {String topic = MixAll.AUTO_CREATE_TOPIC_KEY_TOPIC;TopicConfig topicConfig = new TopicConfig(topic);this.systemTopicList.add(topic);topicConfig.setReadQueueNums(this.brokerController.getBrokerConfig().getDefaultTopicQueueNums());topicConfig.setWriteQueueNums(this.brokerController.getBrokerConfig().getDefaultTopicQueueNums());int perm = PermName.PERM_INHERIT | PermName.PERM_READ | PermName.PERM_WRITE;topicConfig.setPerm(perm);this.topicConfigTable.put(topicConfig.getTopicName(), topicConfig);}}

在接收到消息的时候会分发到org.apache.rocketmq.broker.processor.SendMessageProcessor#sendMessage处理，其中调用了msgCheck方法来检查Topic信息。

 protected RemotingCommand msgCheck(final ChannelHandlerContext ctx,final SendMessageRequestHeader requestHeader, final RemotingCommand response) {// 省略代码TopicConfig topicConfig =this.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());if (null == topicConfig) {int topicSysFlag = 0;if (requestHeader.isUnitMode()) {if (requestHeader.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {topicSysFlag = TopicSysFlag.buildSysFlag(false, true);} else {topicSysFlag = TopicSysFlag.buildSysFlag(true, false);}}log.warn("the topic {} not exist, producer: {}", requestHeader.getTopic(), ctx.channel().remoteAddress());// 创建TopictopicConfig = this.brokerController.getTopicConfigManager().createTopicInSendMessageMethod(requestHeader.getTopic(),requestHeader.getDefaultTopic(),RemotingHelper.parseChannelRemoteAddr(ctx.channel()),requestHeader.getDefaultTopicQueueNums(), topicSysFlag);}

总结

整个流程总结一下，Producer发送一个不存在的Topic消息时，首先会从NameServer拉取Topic路由数据，第一次拉取必然失败，第二次会直接拉取TBW102的路由数据，基于它创建TopicPublishInfo并缓存到本地，进行正常的消息发送，在Header里将defaultTopic设置为TBW102。Broker接收到消息时，先对消息做Check，检查到Topic不存在，会基于defaultTopic的配置去创建该Topic，然后注册到NameServer上，这样一个全新的Topic就被自动创建了。