第七章-Broker-创建topic

本文主要是介绍第七章-Broker-创建topic，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

概念就不讲了，直接上操作和源码，这里就用rocketmq自带的dashboard来创建topic，如下图：

clusterName：Broker的集群，可以选择多个

BROKER_NAME：Broker 名字，可以选择多个

topicName：topic 名字

writeQueueNums：写队列数

readQueueNums：读队列数

perm：设置topic的读写模式

最终创建topic的方法，由DefaultMQAdminExt.createAndUpdateTopicConfig实现，那么我们顺着这个调用链走下去

DefaultMQAdminExt.createAndUpdateTopicConfig

->DefaultMQAdminExtImpl.createAndUpdateTopicConfig

->MQClientAPIImpl.createTopic

MQClientAPIImpl.createTopic

/**
* 先讲下各个参数代表什么
* addr：选择的单个broker的地址，注意这里是单个，当选择多个broker时，会遍历一个一个的调用创建
* defaultTopic：默认topic => TBW102
* topicConfig：读写队列数、权限等信息
* timeoutMillis：超过毫秒，默认20000
*/
public void createTopic(final String addr, final String defaultTopic, final TopicConfig topicConfig,final long timeoutMillis)throws RemotingException, MQBrokerException, InterruptedException, MQClientException {// 组装请求头CreateTopicRequestHeader requestHeader = new CreateTopicRequestHeader();requestHeader.setTopic(topicConfig.getTopicName());requestHeader.setDefaultTopic(defaultTopic);requestHeader.setReadQueueNums(topicConfig.getReadQueueNums());requestHeader.setWriteQueueNums(topicConfig.getWriteQueueNums());requestHeader.setPerm(topicConfig.getPerm());requestHeader.setTopicFilterType(topicConfig.getTopicFilterType().name());requestHeader.setTopicSysFlag(topicConfig.getTopicSysFlag());requestHeader.setOrder(topicConfig.isOrder());// 主要就看这一行，将请求组装成远程命令对象，请求码为 RequestCode.UPDATE_AND_CREATE_TOPIC，在RocketMQ中，所有的网络请求最终都会转化成对应的请求码，所以我们找到处理该请求码的代码就行，翻看源码得知，这里就得转到broker模块，并由AdminBrokerProcessor.processRequest中处理，且该方法中对应 RequestCode.UPDATE_AND_CREATE_TOPIC 请求码的方法是 updateAndCreateTopic(ctx, request)，好了，这下我们直接进入该方法就行。RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UPDATE_AND_CREATE_TOPIC, requestHeader);RemotingCommand response = this.remotingClient.invokeSync(MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(), addr),request, timeoutMillis);assert response != null;switch (response.getCode()) {case ResponseCode.SUCCESS: {return;}default:break;}throw new MQClientException(response.getCode(), response.getRemark());
}

AdminBrokerProcessor.updateAndCreateTopic

private synchronized RemotingCommand updateAndCreateTopic(ChannelHandlerContext ctx,RemotingCommand request) throws RemotingCommandException {// 先创建响应命令对象final RemotingCommand response = RemotingCommand.createResponseCommand(null);// 这里就是转换成当前能识别的，有兴趣的可以进去看看final CreateTopicRequestHeader requestHeader =(CreateTopicRequestHeader) request.decodeCommandCustomHeader(CreateTopicRequestHeader.class);log.info("updateAndCreateTopic called by {}", RemotingHelper.parseChannelRemoteAddr(ctx.channel()));// 检查 topic 名是否与 broker 集群名字冲突，有冲突就要返回错误if (requestHeader.getTopic().equals(this.brokerController.getBrokerConfig().getBrokerClusterName())) {String errorMsg = "the topic[" + requestHeader.getTopic() + "] is conflict with system reserved words.";log.warn(errorMsg);response.setCode(ResponseCode.SYSTEM_ERROR);response.setRemark(errorMsg);return response;}try {// 上面检查通过了，设置返回码为成功response.setCode(ResponseCode.SUCCESS);// 唯一标识当前请求的idresponse.setOpaque(request.getOpaque());response.markResponseType();response.setRemark(null);ctx.writeAndFlush(response);// 将响应信息通过netty写回客户端} catch (Exception e) {log.error("Failed to produce a proper response", e);}// 设置topicConfigTopicConfig topicConfig = new TopicConfig(requestHeader.getTopic());topicConfig.setReadQueueNums(requestHeader.getReadQueueNums());topicConfig.setWriteQueueNums(requestHeader.getWriteQueueNums());topicConfig.setTopicFilterType(requestHeader.getTopicFilterTypeEnum());topicConfig.setPerm(requestHeader.getPerm());topicConfig.setTopicSysFlag(requestHeader.getTopicSysFlag() == null ? 0 : requestHeader.getTopicSysFlag());// 正式存放 topic的操作在这里，看 TopicConfigManager.updateTopicConfigthis.brokerController.getTopicConfigManager().updateTopicConfig(topicConfig);
// 将 broker 的信息再注册到 namesrvthis.brokerController.registerIncrementBrokerData(topicConfig,this.brokerController.getTopicConfigManager().getDataVersion());return null;
}

TopicConfigManager.updateTopicConfig

public void updateTopicConfig(final TopicConfig topicConfig) {// 先在缓存 map中设置 topic的配置TopicConfig old = this.topicConfigTable.put(topicConfig.getTopicName(), topicConfig);if (old != null) {log.info("update topic config, old:[{}] new:[{}]", old, topicConfig);} else {log.info("create new topic [{}]", topicConfig);}// 记录操作版本，其实就是id自增this.dataVersion.nextVersion();// 将topic配置信息持久化到文件this.persist();
}

将topic配置信息持续化到文件

持久化操作在ConfigManager类中执行

public synchronized void persist() {// encode是一个模板方法，由子类实现，在这里指的是 TopicConfigManagerString jsonString = this.encode(true);if (jsonString != null) {// 拿到topic文件存放目录，默认情况下是：{user.home}/store/config/topics.jsonString fileName = this.configFilePath();try {// 将 json 字符串写入 topic 配置文件中，格式如`图7-1`MixAll.string2File(jsonString, fileName);} catch (IOException e) {log.error("persist file " + fileName + " exception", e);}}
}public abstract String encode(final boolean prettyFormat);

TopicConfigManager.encode

public String encode(final boolean prettyFormat) {// 这个方法挺简单，就是把刚才在 updateTopicConfig 方法中存放在本地缓存 map中的值设置到 topicConfigSerializeWrapper对象，同时，将 dataVersion 也设置进去，然后再将对象序列化成 json 字符串TopicConfigSerializeWrapper topicConfigSerializeWrapper = new TopicConfigSerializeWrapper();topicConfigSerializeWrapper.setTopicConfigTable(this.topicConfigTable);topicConfigSerializeWrapper.setDataVersion(this.dataVersion);// 生成 json，格式如`图7-1`return topicConfigSerializeWrapper.toJson(prettyFormat);
}

图7-1

这篇关于第七章-Broker-创建topic的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---