RocketMQ Topic创建【源码笔记】

sh bin/mqadmin updateTopic -c DefaultCluster -n localhost:9876 -t threezto-test -r 12 -w 12

   UpdateTopicSubCommand.execute()->defaultMQAdminExtImpl.createAndUpdateTopicConfig->MQClientAPIImpl.createTopic

//-c指定，集群中每个broker主节点都会创建
Set<String> masterSet =
CommandUtil.fetchMasterAddrByClusterName(defaultMQAdminExt, clusterName);
for (String addr : masterSet) {
    defaultMQAdminExt.createAndUpdateTopicConfig(addr, topicConfig);
    System.out.printf("create topic to %s success.%n", addr);
}

AdminBrokerProcessor.processRequest()->RequestCode.UPDATE_AND_CREATE_TOPIC->updateAndCreateTopic

private RemotingCommand updateAndCreateTopic(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    //省略代码
    //更改broker缓存topicConfigTable信息并落盘存储
    this.brokerController.getTopicConfigManager().updateTopicConfig(topicConfig);
    //向nameSrv上报变更信息，主从同步
    this.brokerController.registerBrokerAll(false, true);

    return null;
}

public void updateTopicConfig(final TopicConfig topicConfig) {
    //更新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);
    }
    //递增版本
    this.dataVersion.nextVersion();
    //存储本地磁盘
    this.persist();
}

/**
 * 持久化磁盘
 */
public synchronized void persist() {
    String jsonString = this.encode(true);
    if (jsonString != null) {
        String fileName = this.configFilePath();
        try {
            MixAll.string2File(jsonString, fileName);
        } catch (IOException e) {
            PLOG.error("persist file Exception, " + fileName, e);
        }
    }
}

"topic_online_test":{
        "order":false,
        "perm":6,
        "readQueueNums":4,
        "topicFilterType":"SINGLE_TAG",
        "topicName":"topic_online_test",
        "topicSysFlag":0,
        "writeQueueNums":4
}

public synchronized void registerBrokerAll(final boolean checkOrderConfig, boolean oneway) {
    TopicConfigSerializeWrapper topicConfigWrapper = this.getTopicConfigManager().buildTopicConfigSerializeWrapper();
    //Broker只有读或者写权限时, Topic设置为与broker相同的权限
    if (!PermName.isWriteable(this.getBrokerConfig().getBrokerPermission())
        || !PermName.isReadable(this.getBrokerConfig().getBrokerPermission())) {
        ConcurrentHashMap<String, TopicConfig> topicConfigTable = new ConcurrentHashMap<String, TopicConfig>();
        for (TopicConfig topicConfig : topicConfigWrapper.getTopicConfigTable().values()) {
            TopicConfig tmp =
                new TopicConfig(topicConfig.getTopicName(), topicConfig.getReadQueueNums(), topicConfig.getWriteQueueNums(),
                    this.brokerConfig.getBrokerPermission());
            topicConfigTable.put(topicConfig.getTopicName(), tmp);
        }
        topicConfigWrapper.setTopicConfigTable(topicConfigTable);
    }
    //向所有NameServer节点注册Topic信息
    RegisterBrokerResult registerBrokerResult = this.brokerOuterAPI.registerBrokerAll(
        this.brokerConfig.getBrokerClusterName(),
        this.getBrokerAddr(),
        this.brokerConfig.getBrokerName(),
        this.brokerConfig.getBrokerId(),
        this.getHAServerAddr(),
        topicConfigWrapper,
        this.filterServerManager.buildNewFilterServerList(),
        oneway,
        this.brokerConfig.getRegisterBrokerTimeoutMills());

    if (registerBrokerResult != null) {
        if (this.updateMasterHAServerAddrPeriodically && registerBrokerResult.getHaServerAddr() != null) {
            this.messageStore.updateHaMasterAddress(registerBrokerResult.getHaServerAddr());
        }
        //Slave同步Master信息 每分钟定时同步
        this.slaveSynchronize.setMasterAddr(registerBrokerResult.getMasterAddr());

        if (checkOrderConfig) {
            this.getTopicConfigManager().updateOrderTopicConfig(registerBrokerResult.getKvTable());
        }
    }
}