RocketMQ?Namesrv架構(gòu)工作原理詳解

更新時(shí)間：2022年08月10日 15:40:19 作者：奔跑的毛球

這篇文章主要為大家介紹了RocketMQ?Namesrv架構(gòu)工作原理詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

1 概念

Namesrv的作用是保存元數(shù)據(jù)，提高Broker的可用性。

Namesrv的主要功能是臨時(shí)存儲(chǔ)，管理Topic路由信息，各個(gè)Namesrv節(jié)點(diǎn)之間是不通信，無(wú)狀態(tài)的，互相不知道對(duì)方的存在。

當(dāng)Broker，生產(chǎn)者，消費(fèi)者啟動(dòng)的時(shí)候，會(huì)輪詢?nèi)康腘amesrv節(jié)點(diǎn)，獲取路由信息。

2 核心數(shù)據(jù)結(jié)構(gòu)和API

2.1 Namesrv的核心數(shù)據(jù)結(jié)構(gòu)

Namesrv中保存的信息是Topic的路由信息，Topic的路由決定了Topic的信息發(fā)送給哪些Broker，或者從哪些Broker獲取消息。

路由數(shù)據(jù)結(jié)構(gòu)的實(shí)現(xiàn)代碼都在org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager中

public class RouteInfoManager {
    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
    //Broker存活的時(shí)間周期，默認(rèn)120秒
    private final static long DEFAULT_BROKER_CHANNEL_EXPIRED_TIME = 1000 * 60 * 2;
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    //保存Topic和隊(duì)列的路由信息
    private final Map<String/* topic */, Map<String, QueueData>> topicQueueTable;
    //Broker名字和Broker信息的對(duì)應(yīng)信息
    private final Map<String/* brokerName */, BrokerData> brokerAddrTable;
    //集群和Broker的對(duì)應(yīng)關(guān)系
    private final Map<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
    //在線的Broker地址和Broker信息的對(duì)應(yīng)關(guān)系
    private final Map<BrokerAddrInfo/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
    //過(guò)濾服務(wù)器消息
    private final Map<BrokerAddrInfo/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;
    private final Map<String/* topic */, Map<String/*brokerName*/, TopicQueueMappingInfo>> topicQueueMappingInfoTable;
    private final BatchUnRegisterService unRegisterService;
    private final NamesrvController namesrvController;
    private final NamesrvConfig namesrvConfig;

2.2 Namesrv的API

Namesrv的的API在org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor中，根據(jù)方法名很容易判斷出來(lái)方法的作用。

switch (request.getCode()) {
    case RequestCode.PUT_KV_CONFIG:
        return this.putKVConfig(ctx, request);
    case RequestCode.GET_KV_CONFIG:
        return this.getKVConfig(ctx, request);
    case RequestCode.DELETE_KV_CONFIG:
        return this.deleteKVConfig(ctx, request);
    case RequestCode.QUERY_DATA_VERSION:
        return this.queryBrokerTopicConfig(ctx, request);
    case RequestCode.REGISTER_BROKER:
        //Broker注冊(cè)自身信息到Namesrv
        return this.registerBroker(ctx, request);
    case RequestCode.UNREGISTER_BROKER:
        //Broker取消注冊(cè)自身信息到Namesrv
        return this.unregisterBroker(ctx, request);
    case RequestCode.BROKER_HEARTBEAT:
        return this.brokerHeartbeat(ctx, request);
    case RequestCode.GET_BROKER_MEMBER_GROUP:
        return this.getBrokerMemberGroup(ctx, request);
    case RequestCode.GET_BROKER_CLUSTER_INFO:
        return this.getBrokerClusterInfo(ctx, request);
    case RequestCode.WIPE_WRITE_PERM_OF_BROKER:
        return this.wipeWritePermOfBroker(ctx, request);
    case RequestCode.ADD_WRITE_PERM_OF_BROKER:
        return this.addWritePermOfBroker(ctx, request);
    case RequestCode.GET_ALL_TOPIC_LIST_FROM_NAMESERVER:
        return this.getAllTopicListFromNameserver(ctx, request);
    case RequestCode.DELETE_TOPIC_IN_NAMESRV:
        return this.deleteTopicInNamesrv(ctx, request);
    case RequestCode.REGISTER_TOPIC_IN_NAMESRV:
        return this.registerTopicToNamesrv(ctx, request);
    case RequestCode.GET_KVLIST_BY_NAMESPACE:
        return this.getKVListByNamespace(ctx, request);
    case RequestCode.GET_TOPICS_BY_CLUSTER:
        return this.getTopicsByCluster(ctx, request);
    case RequestCode.GET_SYSTEM_TOPIC_LIST_FROM_NS:
        return this.getSystemTopicListFromNs(ctx, request);
    case RequestCode.GET_UNIT_TOPIC_LIST:
        return this.getUnitTopicList(ctx, request);
    case RequestCode.GET_HAS_UNIT_SUB_TOPIC_LIST:
        return this.getHasUnitSubTopicList(ctx, request);
    case RequestCode.GET_HAS_UNIT_SUB_UNUNIT_TOPIC_LIST:
        return this.getHasUnitSubUnUnitTopicList(ctx, request);
    case RequestCode.UPDATE_NAMESRV_CONFIG:
        return this.updateConfig(ctx, request);
    case RequestCode.GET_NAMESRV_CONFIG:
        return this.getConfig(ctx, request);
    case RequestCode.GET_CLIENT_CONFIG:
        return this.getClientConfigs(ctx, request);
    default:
        String error = " request type " + request.getCode() + " not supported";
        return RemotingCommand.createResponseCommand(RemotingSysResponseCode.REQUEST_CODE_NOT_SUPPORTED, error);
}

3 Namesrv架構(gòu)

下圖是一個(gè)消息的常規(guī)流轉(zhuǎn)過(guò)程，生產(chǎn)者，消費(fèi)者，Broker通過(guò)與Namesrv交換信息來(lái)實(shí)現(xiàn)自己的功能。

3.1組件

Broker

Broker在啟動(dòng)的時(shí)候，將自己的元數(shù)據(jù)信息，上報(bào)給Namesrv,這部分信息也就是Topic路由。

這里的元數(shù)據(jù)包含Broker本身的元數(shù)據(jù)和該Broker中Topic的信息。

生產(chǎn)者

生產(chǎn)者只關(guān)注Topic路由，從namesrv獲取到Topic路由后就可以知道這個(gè)Topic的消息存放到了哪些Broker中。

消費(fèi)者

消費(fèi)者也只關(guān)注Topic路由，從namesrv獲取到獲取到Topic路由之后，才能知道自己訂閱的Topic的Broker地址，從而獲取消息。

3.2 Namesrv四個(gè)功能模塊

Topic功能管理模塊

這是Namesrv最核心的模塊，Topic路由決定，Topic的數(shù)據(jù)會(huì)保存在哪些Broker上。Broker啟動(dòng)的時(shí)候，會(huì)將自身的信息注冊(cè)到Namesrv中，以供消費(fèi)者和生產(chǎn)者獲取。生產(chǎn)者和消費(fèi)者與Namesrv之間會(huì)有心跳通信，從而獲取最新的Broker信息。