RocketMQ深度解析（二）：NameServer

阿新 • • 發佈：2019-12-31

NamerServer

NameServer是一個非常簡單的Topic路由註冊中心，其角色類似Dubbo中的zookeeper，支援Broker的動態註冊與發現。主要包括兩個功能：Broker管理，NameServer接受Broker叢集的註冊資訊並且儲存下來作為路由資訊的基本資料。然後提供心跳檢測機制，檢查Broker是否還存活；路由資訊管理，每個NameServer將儲存關於Broker叢集的整個路由資訊和用於客戶端查詢的佇列資訊。然後Producer和Conumser通過NameServer就可以知道整個Broker叢集的路由資訊，從而進行訊息的投遞和消費。NameServer通常也是叢集的方式部署，各例項間相互不進行資訊通訊。Broker是向每一臺NameServer註冊自己的路由資訊，所以每一個NameServer例項上面都儲存一份完整的路由資訊。當某個NameServer因某種原因下線了，Broker仍然可以向其它NameServer同步其路由資訊，Producer,Consumer仍然可以動態感知Broker的路由的資訊。

原始碼解析

NameServer啟動流程

填充啟動引數

首先需要建立NameServerConfig（NameServer業務引數）、NettyServerConfig（NameServer網路引數）。

org.apache.rocketmq.namesrv.NamesrvStartup#createNamesrvController

// 首先建立NameServer業務引數
final NamesrvConfig namesrvConfig = new NamesrvConfig();
// 然後建立NameServer網路引數
final NettyServerConfig nettyServerConfig = new NettyServerConfig();
// 預設監聽埠
nettyServerConfig.setListenPort(9876);
// 通過檔案來配置引數
if 
 (commandLine.hasOption('c')) {
    String file = commandLine.getOptionValue('c');
    if (file != null) {
        InputStream in = new BufferedInputStream(new FileInputStream(file));
        properties = new Properties();
        properties.load(in);
        MixAll.properties2Object(properties,namesrvConfig);
        MixAll.properties2Object(properties,nettyServerConfig);

        namesrvConfig.setConfigStorePath(file);

        System.out.printf("load config properties file OK,%s%n" 
,file);
        in.close();
    }
}
// 通過命令列引數
if (commandLine.hasOption('p')) {
    InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_CONSOLE_NAME);
    MixAll.printObjectProperties(console,namesrvConfig);
    MixAll.printObjectProperties(console,nettyServerConfig);
    System.exit(0);
}

MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine),namesrvConfig);
複製程式碼

NamesrvConfig屬性

//rocketMQ主目錄
private String rocketmqHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY,System.getenv(MixAll.ROCKETMQ_HOME_ENV));
//儲存KV配置屬性的持久化路徑
private String kvConfigPath = System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "kvConfig.json";
//預設配置檔案路徑，不生效。
private String configStorePath = System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "namesrv.properties";
private String productEnvName = "center";
private boolean clusterTest = false;
// 是否支援順序訊息，預設是不支援。
private boolean orderMessageEnable = false;
複製程式碼

NettyServerConfig屬性

// 監聽埠
private int listenPort = 8888;
//Netty業務執行緒池執行緒個數
private int serverWorkerThreads = 8;
//Netty public任務執行緒數 Netty網路設計，根據不同的業務型別會建立不同的執行緒池，比如處理訊息傳送、訊息消費、心跳檢測等。
//如果該業務型別未註冊執行緒池，則有Public執行緒池執行。
private int serverCallbackExecutorThreads = 0;
//IO執行緒池執行緒個數
private int serverSelectorThreads = 3;
// send oneway訊息請求併發度
private int serverOnewaySemaphoreValue = 256;
// 非同步訊息傳送併發度
private int serverAsyncSemaphoreValue = 64;
// 網路連線最大空閒時間
private int serverChannelMaxIdleTimeSeconds = 120;

//Socket傳送緩衝區大小
private int serverSocketSndBufSize = NettySystemConfig.socketSndbufSize;
//Socket接受緩衝區大小
private int serverSocketRcvBufSize = NettySystemConfig.socketRcvbufSize;
//ByteBuffer是否開啟快取
private boolean serverPooledByteBufAllocatorEnable = true;
//是否啟用Epoll IO模型 
private boolean useEpollNativeSelector = false;
複製程式碼

根據啟動屬性建立NamesrvController例項，並初始化該例項。

org.apache.rocketmq.namesrv.NamesrvStartup#start

public static NamesrvController start(final NamesrvController controller) throws Exception {

    if (null == controller) {
        throw new IllegalArgumentException("NamesrvController is null");
    }

    //初始化controller例項
    boolean initResult = controller.initialize();
    if (!initResult) {
        controller.shutdown();
        System.exit(-3);
    }

    //註冊JVM鉤子函式，在JVM關閉之前先關閉執行緒池。
    Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log,new Callable<Void>() {
        @Override
        public Void call() throws Exception {
            controller.shutdown();
            return null;
        }
    }));

    //啟動controller
    controller.start();

    return controller;
}
複製程式碼

org.apache.rocketmq.namesrv.NamesrvController#initialize

public boolean initialize() {

    this.kvConfigManager.load();

    this.remotingServer = new NettyRemotingServer(this.nettyServerConfig,this.brokerHousekeepingService);

    this.remotingExecutor =
        Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(),new ThreadFactoryImpl("RemotingExecutorThread_"));

    // 用於處理網路請求
    this.registerProcessor();

    // 10秒掃描一次BrokerLiveTable，移除處於不啟用狀態的Broker
    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

        @Override
        public void run() {
            NamesrvController.this.routeInfoManager.scanNotActiveBroker();
        }
    },5,10,TimeUnit.SECONDS);

    // nameServer每隔10分鐘列印一次KV配置
    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
            NamesrvController.this.kvConfigManager.printAllPeriodically();
        }
    },1,TimeUnit.MINUTES);
    return true;
}
複製程式碼

可以看到，NameServer啟動主要就是載入了兩個配置，然後開啟了兩個心跳檢測執行緒，用於心跳檢測掃描Broker以及列印KV配置。

路由元資訊

NameServer主要作用是為生產者和消費者提供關於Topic的路由資訊，還要能夠管理Broker節點。我們先看一下路由元資料的資料結構。

org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager

// Topic訊息佇列路由資訊，訊息傳送時根據路由表進行負載均衡。
private final HashMap<String/* topic */,List<QueueData>> topicQueueTable;
// Broker基礎資訊
private final HashMap<String/* brokerName */,BrokerData> brokerAddrTable;
// Broker叢集基礎資訊
private final HashMap<String/* clusterName */,Set<String/* brokerName */>> clusterAddrTable;
// Broker狀態資訊
private final HashMap<String/* brokerAddr */,BrokerLiveInfo> brokerLiveTable;
// 類訊息過濾
private final HashMap<String/* brokerAddr */,List<String>/* Filter Server */> filterServerTable;
複製程式碼

路由註冊

路由註冊時通過Broker與NameServer的心跳功能實現的。Broker啟動時向叢集中所有的NameServer保持長連線，每隔30秒向所有的NameServer傳送心跳包，NameServer收到心跳包時會更新brokerLiveTable快取中BrokerLiveInfo的lastUpdateTimestamp，然後NameServer每隔10s掃描brokerLiveTable，如果連續120s沒有收到心跳包，NameServer將移除該Broker的路由資訊同時關閉Socket連線。

Broker心跳傳送

Broker端心跳包傳送（org.apache.rocketmq.broker.BrokerController#start）

// 每隔10s向NameServer傳送心跳包
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

    @Override
    public void run() {
        try {
            BrokerController.this.registerBrokerAll(true,false,brokerConfig.isForceRegister());
        }   catch (Throwable e) {
            log.error("registerBrokerAll Exception",e);
        }
}
},1000 * 10,Math.max(10000,Math.min(brokerConfig.getRegisterNameServerPeriod(),60000)),TimeUnit.MILLISECONDS);

複製程式碼

org.apache.rocketmq.broker.BrokerController#registerBrokerAll

// 獲取nameServerAddress列表
List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
if (nameServerAddressList != null && nameServerAddressList.size() > 0) {

    final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();
    requestHeader.setBrokerAddr(brokerAddr);
    requestHeader.setBrokerId(brokerId);
    requestHeader.setBrokerName(brokerName);
    requestHeader.setClusterName(clusterName);
    requestHeader.setHaServerAddr(haServerAddr);
    requestHeader.setCompressed(compressed);

    RegisterBrokerBody requestBody = new RegisterBrokerBody();
    requestBody.setTopicConfigSerializeWrapper(topicConfigWrapper);
    requestBody.setFilterServerList(filterServerList);
    final byte[] body = requestBody.encode(compressed);
    final int bodyCrc32 = UtilAll.crc32(body);
    requestHeader.setBodyCrc32(bodyCrc32);
    // 使用countDownLatch阻塞當前執行緒，等待心跳同步執行緒
    final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());
    for (final String namesrvAddr : nameServerAddressList) {
        brokerOuterExecutor.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    // 傳送心跳同步資訊
                    RegisterBrokerResult result = registerBroker(namesrvAddr,oneway,timeoutMills,requestHeader,body);
                    if (result != null) {
                        registerBrokerResultList.add(result);
                    }

                    log.info("register broker[{}]to name server {} OK",brokerId,namesrvAddr);
                } catch (Exception e) {
                    log.warn("registerBroker Exception,{}",namesrvAddr,e);
                } finally {
                    countDownLatch.countDown();
                }
            }
            });
    }
    try {
        //等待心跳同步執行緒全部執行完畢才往下執行
        countDownLatch.await(timeoutMills,TimeUnit.MILLISECONDS);
    } catch (InterruptedException e) {
    }
}
複製程式碼

org.apache.rocketmq.broker.out.BrokerOuterAPI#registerBroker

    // 封裝請求頭
    RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.REGISTER_BROKER,requestHeader);
    request.setBody(body);
    //傳送訊息
    if (oneway) {
        try {
            this.remotingClient.invokeOneway(namesrvAddr,request,timeoutMills);
        } catch (RemotingTooMuchRequestException e) {
            // Ignore
        }
        return null;
    }
    傳送訊息
    RemotingCommand response = this.remotingClient.invokeSync(namesrvAddr,timeoutMills);
複製程式碼

向NameServer伺服器傳送型別為RequestCode.REGISTER_BROKER的請求。

NameServer處理心跳

NameServer處理網路請求在org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor#processRequest，如果請求型別為RequestCode.REGISTER_BROKER，則請求最終轉發到org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#registerBroker。

org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#registerBroker clusterAddrTable維護

                // 寫鎖
                this.lock.writeLock().lockInterruptibly();
                // 判斷叢集是否存在
                Set<String> brokerNames = this.clusterAddrTable.get(clusterName);
                if (null == brokerNames) {
                    brokerNames = new HashSet<String>();
                    this.clusterAddrTable.put(clusterName,brokerNames);
                }
                brokerNames.add(brokerName);
複製程式碼

clusterAddrTable維護

                boolean registerFirst = false;
            
                BrokerData brokerData = this.brokerAddrTable.get(brokerName);
                // 如果是新的
                if (null == brokerData) {
                    registerFirst = true;
                    brokerData = new BrokerData(clusterName,brokerName,new HashMap<Long,String>());
                    this.brokerAddrTable.put(brokerName,brokerData);
                }
                
                Map<Long,String> brokerAddrsMap = brokerData.getBrokerAddrs();
                Iterator<Entry<Long,String>> it = brokerAddrsMap.entrySet().iterator();
                // 刪除過時的
                while (it.hasNext()) {
                    Entry<Long,String> item = it.next();
                    if (null != brokerAddr && brokerAddr.equals(item.getValue()) && brokerId != item.getKey()) {
                        it.remove();
                    }
                }
                String oldAddr = brokerData.getBrokerAddrs().put(brokerId,brokerAddr);
                registerFirst = registerFirst || (null == oldAddr);
複製程式碼

topicQueueTable維護

//如果是master
if (null != topicConfigWrapper && MixAll.MASTER_ID == brokerId) {
    // 如果config發生了改變或者是第一次註冊
    if (this.isBrokerTopicConfigChanged(brokerAddr,topicConfigWrapper.getDataVersion())
            || registerFirst) {
        ConcurrentMap<String,TopicConfig> tcTable =
        topicConfigWrapper.getTopicConfigTable();
        if (tcTable != null) {
            for (Map.Entry<String,TopicConfig> entry : tcTable.entrySet()) {
                 // 更新路由元資料
                 this.createAndUpdateQueueData(brokerName,entry.getValue());
            }
        }
    }
}
複製程式碼

如果Broker為Master，並且BrokerTopic配置資訊發生變化或者是初次註冊，則需要建立或更新Topic路由元資料，填充TopicQueueTable，其實就是為預設主題自動註冊路由資訊，其中包含MixAll.DEFAULT_TOPIC的路由資訊。如果該主題為建立並且BrokerConfig的autoCreateTopicEnable為true時，將返回MixAll.DEFAULT_TOPIC的路由資訊。

org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#createAndUpdateQueueData 根據TopicConfig建立QueueData資料結構，然後更新topicQueueTable

private void createAndUpdateQueueData(final String brokerName,final TopicConfig topicConfig) {
        QueueData queueData = new QueueData();
        queueData.setBrokerName(brokerName);
        queueData.setWriteQueueNums(topicConfig.getWriteQueueNums());
        queueData.setReadQueueNums(topicConfig.getReadQueueNums());
        queueData.setPerm(topicConfig.getPerm());
        queueData.setTopicSynFlag(topicConfig.getTopicSysFlag());

        List<QueueData> queueDataList = this.topicQueueTable.get(topicConfig.getTopicName());
        if (null == queueDataList) {
            queueDataList = new LinkedList<QueueData>();
            queueDataList.add(queueData);
            this.topicQueueTable.put(topicConfig.getTopicName(),queueDataList);
            log.info("new topic registered,{} {}",topicConfig.getTopicName(),queueData);
        } else {
            boolean addNewOne = true;

            Iterator<QueueData> it = queueDataList.iterator();
            while (it.hasNext()) {
                QueueData qd = it.next();
                if (qd.getBrokerName().equals(brokerName)) {
                    if (qd.equals(queueData)) {
                        addNewOne = false;
                    } else {
                        log.info("topic changed,{} OLD: {} NEW: {}",qd,queueData);
                        it.remove();
                    }
                }
            }

            if (addNewOne) {
                queueDataList.add(queueData);
            }
        }
    }
複製程式碼

根據TopicConfig建立QueueData資料結構，然後更新topicQueueTable。

brokerLiveTable維護

                BrokerLiveInfo prevBrokerLiveInfo = this.brokerLiveTable.put(brokerAddr,new BrokerLiveInfo(
                        System.currentTimeMillis(),topicConfigWrapper.getDataVersion(),channel,haServerAddr));
                if (null == prevBrokerLiveInfo) {
                    log.info("new broker registered,{} HAServer: {}",brokerAddr,haServerAddr);
                }

複製程式碼

brokerLiveTable是執行路由刪除的重要依據。

filterServerTable維護

if (filterServerList != null) {
    if (filterServerList.isEmpty()) {
        this.filterServerTable.remove(brokerAddr);
    } else {
        this.filterServerTable.put(brokerAddr,filterServerList);
    }
}

if (MixAll.MASTER_ID != brokerId) {
    String masterAddr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
    if (masterAddr != null) {
        BrokerLiveInfo brokerLiveInfo = this.brokerLiveTable.get(masterAddr);
        if (brokerLiveInfo != null) {
            result.setHaServerAddr(brokerLiveInfo.getHaServerAddr());
            result.setMasterAddr(masterAddr);
        }
    }
}
複製程式碼

路由註冊總結

路由註冊就是Broker每30s向所有NameServer傳送一次心跳並帶上自己的資訊，然後NameServer處理這些資訊，更新路由元資料的過程。可以看到，這裡路由表使用了HashMap資料結構實現，同時使用了鎖粒度較小的讀寫鎖來控制併發，允許多個Client併發讀，但是同一時刻只允許處理一個心跳包。

路由刪除

上面已經介紹過，NameServer和Broker保持長連線，Broker狀態儲存在BrokerLiveTable裡，NameServer會每10s掃描一次NameServer，一旦發現已經有120s沒有收到Broker傳送過來的心跳資訊，就移除該Broker並關閉與Broker的連線，同時更新路由元資訊。還有一種情況是Broker正常關閉，會執行unRegisterBroker指令。

掃描brokerLiveTable

上面說過，在NameServer啟動時會開啟一個執行緒，每10秒呼叫一次scanNotActiveBroker方法。

org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#scanNotActiveBroker

    public void scanNotActiveBroker() {
        Iterator<Entry<String,BrokerLiveInfo>> it = this.brokerLiveTable.entrySet().iterator();
        // 遍歷brokerLiveTable的lastUpdateTimestamp
        while (it.hasNext()) {
            Entry<String,BrokerLiveInfo> next = it.next();
            long last = next.getValue().getLastUpdateTimestamp();
            //超過120秒就刪除
            if ((last + BROKER_CHANNEL_EXPIRED_TIME) < System.currentTimeMillis()) {
                RemotingUtil.closeChannel(next.getValue().getChannel());
                it.remove();
                log.warn("The broker channel expired,{} {}ms",next.getKey(),BROKER_CHANNEL_EXPIRED_TIME);
                //然後關閉Channel
                this.onChannelDestroy(next.getKey(),next.getValue().getChannel());
            }
        }
    }
複製程式碼

維護路由表

org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#onChannelDestroy

                try {
                    // 加鎖
                    this.lock.writeLock().lockInterruptibly();
                    this.brokerLiveTable.remove(brokerAddrFound);
                    this.filterServerTable.remove(brokerAddrFound);
                    String brokerNameFound = null;
                    boolean removeBrokerName = false;
                    Iterator<Entry<String,BrokerData>> itBrokerAddrTable =
                        this.brokerAddrTable.entrySet().iterator();
                    while (itBrokerAddrTable.hasNext() && (null == brokerNameFound)) {
                        BrokerData brokerData = itBrokerAddrTable.next().getValue();

                        Iterator<Entry<Long,String>> it = brokerData.getBrokerAddrs().entrySet().iterator();
                        while (it.hasNext()) {
                            Entry<Long,String> entry = it.next();
                            Long brokerId = entry.getKey();
                            String brokerAddr = entry.getValue();
                            if (brokerAddr.equals(brokerAddrFound)) {
                                brokerNameFound = brokerData.getBrokerName();
                                it.remove();
                                log.info("remove brokerAddr[{},{}] from brokerAddrTable,because channel destroyed",brokerAddr);
                                break;
                            }
                        }

                        if (brokerData.getBrokerAddrs().isEmpty()) {
                            removeBrokerName = true;
                            itBrokerAddrTable.remove();
                            log.info("remove brokerName[{}] from brokerAddrTable,brokerData.getBrokerName());
                        }
                    }

                    if (brokerNameFound != null && removeBrokerName) {
                        Iterator<Entry<String,Set<String>>> it = this.clusterAddrTable.entrySet().iterator();
                        while (it.hasNext()) {
                            Entry<String,Set<String>> entry = it.next();
                            String clusterName = entry.getKey();
                            Set<String> brokerNames = entry.getValue();
                            boolean removed = brokerNames.remove(brokerNameFound);
                            if (removed) {
                                log.info("remove brokerName[{}],clusterName[{}] from clusterAddrTable,brokerNameFound,clusterName);

                                if (brokerNames.isEmpty()) {
                                    log.info("remove the clusterName[{}] from clusterAddrTable,because channel destroyed and no broker in this cluster",clusterName);
                                    it.remove();
                                }

                                break;
                            }
                        }
                    }

                    if (removeBrokerName) {
                        Iterator<Entry<String,List<QueueData>>> itTopicQueueTable =
                            this.topicQueueTable.entrySet().iterator();
                        while (itTopicQueueTable.hasNext()) {
                            Entry<String,List<QueueData>> entry = itTopicQueueTable.next();
                            String topic = entry.getKey();
                            List<QueueData> queueDataList = entry.getValue();

                            Iterator<QueueData> itQueueData = queueDataList.iterator();
                            while (itQueueData.hasNext()) {
                                QueueData queueData = itQueueData.next();
                                if (queueData.getBrokerName().equals(brokerNameFound)) {
                                    itQueueData.remove();
                                    log.info("remove topic[{} {}],from topicQueueTable,topic,queueData);
                                }
                            }

                            if (queueDataList.isEmpty()) {
                                itTopicQueueTable.remove();
                                log.info("remove topic[{}] all queue,topic);
                            }
                        }
                    }
                } finally {
                    this.lock.writeLock().unlock();
                }
複製程式碼

路由刪除總結

主要就是NameServer定時執行緒每10s掃描一次BrokerLiveTable，發現120s沒動靜的Broker就移除該Broker相關資訊，維護路由表就不再贅述。

路由發現

路由發現並不是實時的，當Topic路由出現變化後，NameServer不主動推送給客戶端。由客戶端定時拉取Topic最新的路由。

org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor#getRouteInfoByTopic

public RemotingCommand getRouteInfoByTopic(ChannelHandlerContext ctx,RemotingCommand request) throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);
    final GetRouteInfoRequestHeader requestHeader =
        (GetRouteInfoRequestHeader) request.decodeCommandCustomHeader(GetRouteInfoRequestHeader.class);
    // 從路由表中找到主題對應的路由資訊填充topicRouteData
    TopicRouteData topicRouteData = this.namesrvController.getRouteInfoManager().pickupTopicRouteData(requestHeader.getTopic());

    if (topicRouteData != null) {
        if (this.namesrvController.getNamesrvConfig().isOrderMessageEnable()) {
            String orderTopicConf =
                this.namesrvController.getKvConfigManager().getKVConfig(NamesrvUtil.NAMESPACE_ORDER_TOPIC_CONFIG,requestHeader.getTopic());
            topicRouteData.setOrderTopicConf(orderTopicConf);
        }

        byte[] content = topicRouteData.encode();
        response.setBody(content);
        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }

    response.setCode(ResponseCode.TOPIC_NOT_EXIST);
    response.setRemark("No topic route info in name server for the topic: " + requestHeader.getTopic()
        + FAQUrl.suggestTodo(FAQUrl.APPLY_TOPIC_URL));
    return response;
}
複製程式碼

#總結

如上圖所示，Broker每30s向NameServer叢集傳送心跳資訊，NameServer收到心跳資訊後更新brokerLiveTable中的資訊，記錄lastUpdateTime，NameServer每隔10s掃描brokerLiveTable，比較brokerLiveTable中每一個brokerLiveInfo裡的lastUpdateTime，如果發現超過120s，則刪除broker相關的所有資訊。當Topic資訊發生變化時，NameServer並不通知客戶端，而是客戶端定時拉去Topic路由資訊。

參考文獻

RocketMQ官方Doc
RocketMQ技術內幕

RocketMQ深度解析（二）：NameServer

NamerServer

原始碼解析

NameServer啟動流程

填充啟動引數

根據啟動屬性建立NamesrvController例項，並初始化該例項。

路由元資訊

路由註冊

Broker心跳傳送

NameServer處理心跳

路由註冊總結

路由刪除

掃描brokerLiveTable

維護路由表

路由刪除總結

路由發現

參考文獻

RocketMQ深度解析（二）：NameServer

RocketMQ深度解析（四）：Consumer

RocketMQ深度解析（一）：RocketMQ總體設計

RocketMQ深度解析（三）：Producer

JavaScript ECMA-262-3 深入解析（二）：變數物件例項詳解

C#中的深度學習（二）：預處理識別硬幣的資料集

【雲原生小課堂】Envoy請求流程原始碼解析（二）：請求解析

Druid原始碼解析（二）：連線池初始化（init 方法）

深度學習論文翻譯解析（八）：Rich feature hierarchies for accurate object detection and semantic segmentation

深度學習論文翻譯解析（九）：Spatial Pyramid Pooling in Deep Convolutional Networks for Visual Recognition

深度學習論文翻譯解析（十）：Visualizing and Understanding Convolutional Networks

Mybatis原始碼閱讀（二）：動態節點解析2.2 —— SqlSourceBuilder與三種SqlSource

深度學習程式碼入門（二）：tf下的Keras介面，非自定義網路初步實現

HotSpot學習（二）：虛擬機器的啟動過程原始碼解析

【異常檢測】DAGMM：結合深度自編碼器器和GMM的端到端無監督網路（二）：程式碼實戰（PyTorch）

RocketMQ架構原理解析（三）：訊息索引

Dubbo原始碼解析（二）Dubbo擴充套件機制SPI

伯陽的網路筆記（二）：HTTP基礎

Java SE基礎鞏固（二）：String類

Spring Boot （二）：模版引擎 Thymeleaf 渲染 Web 頁面

RocketMQ深度解析（二）：NameServer

NamerServer

原始碼解析

NameServer啟動流程

填充啟動引數

根據啟動屬性建立NamesrvController例項，並初始化該例項。

路由元資訊

路由註冊

Broker心跳傳送

NameServer處理心跳

路由註冊總結

路由刪除

掃描brokerLiveTable

維護路由表

路由刪除總結

路由發現

參考文獻

相關推薦