03.05 RocketMQ 源碼分析NameServer（8）

2.2.1 架構設計

消息中間件的設計思路一般是基於主題訂閱發佈的機制，消息生產者（Producer）發送某一個主題到消息服務器，消息服務器負責將消息持久化存儲，消息消費者（Consumer）訂閱該興趣的主題，消息服務器根據訂閱信息（路由信息）將消息推送到消費者（Push模式）或者消費者主動向消息服務器拉去（Pull模式），從而實現消息生產者與消息消費者解耦。為了避免消息服務器的單點故障導致的整個系統癱瘓，通常會部署多臺消息服務器共同承擔消息的存儲。那消息生產者如何知道消息要發送到哪臺消息服務器呢？如果某一臺消息服務器宕機了，那麼消息生產者如何在不重啟服務情況下感知呢？

NameServer就是為了解決以上問題設計的。

Broker消息服務器在啟動的時向所有NameServer註冊，消息生產者（Producer）在發送消息時之前先從NameServer獲取Broker服務器地址列表，然後根據負載均衡算法從列表中選擇一臺服務器進行發送。NameServer與每臺Broker保持長連接，並間隔30S檢測Broker是否存活，如果檢測到Broker宕機，則從路由註冊表中刪除。但是路由變化不會馬上通知消息生產者。這樣設計的目的是為了降低NameServer實現的複雜度，在消息發送端提供容錯機制保證消息發送的可用性。

NameServer本身的高可用是通過部署多臺NameServer來實現，但彼此之間不通訊，也就是NameServer服務器之間在某一個時刻的數據並不完全相同，但這對消息發送並不會造成任何影響，這也是NameServer設計的一個亮點，總之，RocketMQ設計追求簡單高效。

2.2.2 啟動流程

啟動類：org.apache.rocketmq.namesrv.NamesrvStartup

步驟一

解析配置文件，填充NameServerConfig、NettyServerConfig屬性值，並創建NamesrvController

代碼：NamesrvController#createNamesrvController

<code>//創建NamesrvConfig
final NamesrvConfig namesrvConfig = new NamesrvConfig();
//創建NettyServerConfig
final NettyServerConfig nettyServerConfig = new NettyServerConfig();
//設置啟動端口號
nettyServerConfig.setListenPort(9876);
//解析啟動-c參數
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();
    }
}
//解析啟動-p參數
if (commandLine.hasOption('p')) {
    InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_CONSOLE_NAME);
    MixAll.printObjectProperties(console, namesrvConfig);
    MixAll.printObjectProperties(console, nettyServerConfig);
    System.exit(0);
}
//將啟動參數填充到namesrvConfig,nettyServerConfig
MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);

//創建NameServerController 

final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);/<code>

NamesrvConfig屬性

<code>private String rocketmqHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY, System.getenv(MixAll.ROCKETMQ_HOME_ENV));
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;/<code>

rocketmqHome：rocketmq主目錄

kvConfig：NameServer存儲KV配置屬性的持久化路徑

configStorePath：nameServer默認配置文件路徑

orderMessageEnable：是否支持順序消息

NettyServerConfig屬性

<code>private int listenPort = 8888;
private int serverWorkerThreads = 8;
private int serverCallbackExecutorThreads = 0;
private int serverSelectorThreads = 3;
private int serverOnewaySemaphoreValue = 256;
private int serverAsyncSemaphoreValue = 64;
private int serverChannelMaxIdleTimeSeconds = 120;
private int serverSocketSndBufSize = NettySystemConfig.socketSndbufSize;
private int serverSocketRcvBufSize = NettySystemConfig.socketRcvbufSize;
private boolean serverPooledByteBufAllocatorEnable = true; 

private boolean useEpollNativeSelector = false;/<code>

listenPort：NameServer監聽端口，該值默認會被初始化為9876 serverWorkerThreads：Netty業務線程池線程個數 serverCallbackExecutorThreads：Netty public任務線程池線程個數，Netty網絡設計，根據業務類型會創建不同的線程池，比如處理消息發送、消息消費、心跳檢測等。如果該業務類型未註冊線程池，則由public線程池執行。 serverSelectorThreads：IO線程池個數，主要是NameServer、Broker端解析請求、返回相應的線程個數，這類線程主要是處理網路請求的，解析請求包，然後轉發到各個業務線程池完成具體的操作，然後將結果返回給調用方; serverOnewaySemaphoreValue：send oneway消息請求併發讀（Broker端參數）; serverAsyncSemaphoreValue：異步消息發送最大併發度; serverChannelMaxIdleTimeSeconds ：網絡連接最大的空閒時間，默認120s。 serverSocketSndBufSize：網絡socket發送緩衝區大小。 serverSocketRcvBufSize： 網絡接收端緩存區大小。 serverPooledByteBufAllocatorEnable：ByteBuffer是否開啟緩存; useEpollNativeSelector：是否啟用Epoll IO模型。

步驟二

根據啟動屬性創建NamesrvController實例，並初始化該實例。NameServerController實例為NameServer核心控制器

代碼：NamesrvController#initialize

<code>public boolean initialize() {
\t//加載KV配置
    this.kvConfigManager.load();
\t//創建NettyServer網絡處理對象
    this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
\t//開啟定時任務:每隔10s掃描一次Broker,移除不活躍的Broker
    this.remotingExecutor =
        Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));
    this.registerProcessor();
    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
            NamesrvController.this.routeInfoManager.scanNotActiveBroker();
        }
    }, 5, 10, TimeUnit.SECONDS);
\t//開啟定時任務:每隔10min打印一次KV配置
\tthis.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

        @Override
        public void run() {
            NamesrvController.this.kvConfigManager.printAllPeriodically();
        }
    }, 1, 10, TimeUnit.MINUTES);
    return true;
}/<code>

步驟三

在JVM進程關閉之前，先將線程池關閉，及時釋放資源

<code>//註冊JVM鉤子函數代碼
Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<void>() {
    @Override
    public Void call() throws Exception {
        //釋放資源
        controller.shutdown();
        return null;
    }
}));/<void>/<code>

2.2.3 路由管理

NameServer的主要作用是為消息的生產者和消息消費者提供關於主題Topic的路由信息，那麼NameServer需要存儲路由的基礎信息，還要管理Broker節點，包括路由註冊、路由刪除等。

2.2.3.1 路由元信息

代碼：RouteInfoManager

<code>private final HashMap<string>> topicQueueTable;
private final HashMap<string> brokerAddrTable;
private final HashMap<string>> clusterAddrTable;
private final HashMap<string> brokerLiveTable;
private final HashMap<string>/* Filter Server */> filterServerTable;/<string>/<string>/<string>/<string>/<code>

topicQueueTable：Topic消息隊列路由信息，消息發送時根據路由表進行負載均衡

brokerAddrTable：Broker基礎信息，包括brokerName、所屬集群名稱、主備Broker地址

clusterAddrTable：Broker集群信息，存儲集群中所有Broker名稱

brokerLiveTable：Broker狀態信息，NameServer每次收到心跳包是會替換該信息

filterServerTable：Broker上的FilterServer列表，用於類模式消息過濾。

RocketMQ基於定於發佈機制，一個Topic擁有多個消息隊列，一個Broker為每一個主題創建4個讀隊列和4個寫隊列。多個Broker組成一個集群，集群由相同的多臺Broker組成Master-Slave架構，brokerId為0代表Master，大於0為Slave。BrokerLiveInfo中的lastUpdateTimestamp存儲上次收到Broker心跳包的時間。

2.2.3.2 路由註冊

1）發送心跳包

RocketMQ路由註冊是通過Broker與NameServer的心跳功能實現的。Broker啟動時向集群中所有的NameServer發送心跳信息，每隔30s向集群中所有NameServer發送心跳包，NameServer收到心跳包時會更新brokerLiveTable緩存中BrokerLiveInfo的lastUpdataTimeStamp信息，然後NameServer每隔10s掃描brokerLiveTable，如果連續120S沒有收到心跳包，NameServer將移除Broker的路由信息同時關閉Socket連接。

代碼：BrokerController#start

<code>

代碼：BrokerOuterAPI#registerBrokerAll

<code>

代碼：BrokerOutAPI#registerBroker

<code>if (oneway) {
    try {
        this.remotingClient.invokeOneway(namesrvAddr, request, timeoutMills);
    } catch (RemotingTooMuchRequestException e) {
        // Ignore
    }
    return null;
}
RemotingCommand response = this.remotingClient.invokeSync(namesrvAddr, request, timeoutMills);/<code>

2）處理心跳包

org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor網路處理類解析請求類型，如果請求類型是為REGISTER_BROKER，則將請求轉發到RouteInfoManager#regiesterBroker

代碼：DefaultRequestProcessor#processRequest

<code>

代碼：DefaultRequestProcessor#registerBroker

<code>RegisterBrokerResult result = this.namesrvController.getRouteInfoManager().registerBroker(
    requestHeader.getClusterName(),
    requestHeader.getBrokerAddr(),
    requestHeader.getBrokerName(), 

    requestHeader.getBrokerId(),
    requestHeader.getHaServerAddr(),
    topicConfigWrapper,
    null,
    ctx.channel()
);/<code>

代碼：RouteInfoManager#registerBroker

維護路由信息

<code>

<code>//維護brokerAddrTable
BrokerData brokerData = this.brokerAddrTable.get(brokerName);
//第一次註冊,則創建brokerData
if (null == brokerData) {
    registerFirst = true;
    brokerData = new BrokerData(clusterName, brokerName, new HashMap<long>());
    this.brokerAddrTable.put(brokerName, brokerData);
}
//非第一次註冊,更新Broker
Map<long> brokerAddrsMap = brokerData.getBrokerAddrs();
Iterator<entry>> it = brokerAddrsMap.entrySet().iterator();
while (it.hasNext()) {
    Entry<long> 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);/<long>/<entry>/<long>/<long>/<code>

<code>//維護topicQueueTable
if (null != topicConfigWrapper && MixAll.MASTER_ID == brokerId) {
    if (this.isBrokerTopicConfigChanged(brokerAddr, topicConfigWrapper.getDataVersion()) || 
        registerFirst) {
        ConcurrentMap<string> tcTable = topicConfigWrapper.getTopicConfigTable();
        if (tcTable != null) {
            for (Map.Entry<string> entry : tcTable.entrySet()) {
                this.createAndUpdateQueueData(brokerName, entry.getValue());
            }
        }
    }
}/<string>/<string>/<code>

代碼：RouteInfoManager#createAndUpdateQueueData

<code>private void createAndUpdateQueueData(final String brokerName, final TopicConfig topicConfig) {
    //創建QueueData
\tQueueData queueData = new QueueData();
\tqueueData.setBrokerName(brokerName);
\tqueueData.setWriteQueueNums(topicConfig.getWriteQueueNums());
\tqueueData.setReadQueueNums(topicConfig.getReadQueueNums());
\tqueueData.setPerm(topicConfig.getPerm());
\tqueueData.setTopicSynFlag(topicConfig.getTopicSysFlag());
\t//獲得topicQueueTable中隊列集合
\tList<queuedata> queueDataList = this.topicQueueTable.get(topicConfig.getTopicName());
    //topicQueueTable為空,則直接添加queueData到隊列集合
\tif (null == queueDataList) {
\t    queueDataList = new LinkedList<queuedata>();
\t    queueDataList.add(queueData);
\t    this.topicQueueTable.put(topicConfig.getTopicName(), queueDataList);
\t    log.info("new topic registered, {} {}", topicConfig.getTopicName(), queueData);
\t} else {
        //判斷是否是新的隊列
\t    boolean addNewOne = true;
\t    Iterator<queuedata> it = queueDataList.iterator();
\t    while (it.hasNext()) {
\t        QueueData qd = it.next();
            //如果brokerName相同,代表不是新的隊列
\t        if (qd.getBrokerName().equals(brokerName)) {
\t            if (qd.equals(queueData)) {
\t                addNewOne = false;
\t        } else {
\t                    log.info("topic changed, {} OLD: {} NEW: {}", topicConfig.getTopicName(), qd,
\t                        queueData);
\t                    it.remove();
\t                }
\t            }
\t        }
\t\t//如果是新的隊列,則添加隊列到queueDataList
        if (addNewOne) {
            queueDataList.add(queueData);
        }
    }
}/<queuedata>/<queuedata>/<queuedata>/<code>

<code>//維護brokerLiveTable 

BrokerLiveInfo prevBrokerLiveInfo = this.brokerLiveTable.put(brokerAddr,new BrokerLiveInfo(
    System.currentTimeMillis(),
    topicConfigWrapper.getDataVersion(),
    channel,
    haServerAddr));/<code>

<code>//維護filterServerList
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);
        }
    }
}/<code>

2.2.3.3 路由刪除

Broker每隔30s向NameServer發送一個心跳包，心跳包包含BrokerId，Broker地址，Broker名稱，Broker所屬集群名稱、Broker關聯的FilterServer列表。但是如果Broker宕機，NameServer無法收到心跳包，此時NameServer如何來剔除這些失效的Broker呢？NameServer會每隔10s掃描brokerLiveTable狀態表，如果BrokerLive的lastUpdateTimestamp的時間戳距當前時間超過120s，則認為Broker失效，移除該Broker，關閉與Broker連接，同時更新topicQueueTable、brokerAddrTable、brokerLiveTable、filterServerTable。

RocketMQ有兩個觸發點來刪除路由信息：

• NameServer定期掃描brokerLiveTable檢測上次心跳包與當前系統的時間差，如果時間超過120s，則需要移除broker。
• Broker在正常關閉的情況下，會執行unregisterBroker指令

這兩種方式路由刪除的方法都是一樣的，就是從相關路由表中刪除與該broker相關的信息。

代碼：NamesrvController#initialize

<code>//每隔10s掃描一次為活躍Broker
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

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

代碼：RouteInfoManager#scanNotActiveBroker

<code>public void scanNotActiveBroker() {
    //獲得brokerLiveTable
    Iterator<entry>> it = this.brokerLiveTable.entrySet().iterator();
    //遍歷brokerLiveTable
    while (it.hasNext()) {
        Entry<string> next = it.next();
        long last = next.getValue().getLastUpdateTimestamp();
        //如果收到心跳包的時間距當時時間是否超過120s
        if ((last + BROKER_CHANNEL_EXPIRED_TIME) < System.currentTimeMillis()) {
            //關閉連接
            RemotingUtil.closeChannel(next.getValue().getChannel());
            //移除broker
            it.remove();
            //維護路由表
            this.onChannelDestroy(next.getKey(), next.getValue().getChannel());
        }
    }
}/<string>/<entry>/<code>

代碼：RouteInfoManager#onChannelDestroy

<code>//申請寫鎖,根據brokerAddress從brokerLiveTable和filterServerTable移除
this.lock.writeLock().lockInterruptibly();
this.brokerLiveTable.remove(brokerAddrFound);
this.filterServerTable.remove(brokerAddrFound);/<code>

<code>//維護brokerAddrTable 

String brokerNameFound = null;
boolean removeBrokerName = false;
Iterator<entry>> itBrokerAddrTable =this.brokerAddrTable.entrySet().iterator();
//遍歷brokerAddrTable
while (itBrokerAddrTable.hasNext() && (null == brokerNameFound)) {
    BrokerData brokerData = itBrokerAddrTable.next().getValue();
    //遍歷broker地址
    Iterator<entry>> it = brokerData.getBrokerAddrs().entrySet().iterator();
    while (it.hasNext()) {
        Entry<long> entry = it.next();
        Long brokerId = entry.getKey();
        String brokerAddr = entry.getValue();
        //根據broker地址移除brokerAddr
        if (brokerAddr.equals(brokerAddrFound)) {
            brokerNameFound = brokerData.getBrokerName();
            it.remove();
            log.info("remove brokerAddr[{}, {}] from brokerAddrTable, because channel destroyed",
                brokerId, brokerAddr);
            break;
        }
    }
\t//如果當前主題只包含待移除的broker,則移除該topic
    if (brokerData.getBrokerAddrs().isEmpty()) {
        removeBrokerName = true;
        itBrokerAddrTable.remove();
        log.info("remove brokerName[{}] from brokerAddrTable, because channel destroyed",
            brokerData.getBrokerName());
    }
}/<long>/<entry>/<entry>/<code>

<code>//維護clusterAddrTable
if (brokerNameFound != null && removeBrokerName) {
    Iterator<entry>>> it = this.clusterAddrTable.entrySet().iterator();
    //遍歷clusterAddrTable
    while (it.hasNext()) {
        Entry<string>> entry = it.next();
        //獲得集群名稱
        String clusterName = entry.getKey();
        //獲得集群中brokerName集合
        Set<string> brokerNames = entry.getValue();
        //從brokerNames中移除brokerNameFound
        boolean removed = brokerNames.remove(brokerNameFound);
        if (removed) {
            log.info("remove brokerName[{}], clusterName[{}] from clusterAddrTable, because channel destroyed",
                brokerNameFound, clusterName); 


            if (brokerNames.isEmpty()) {
                log.info("remove the clusterName[{}] from clusterAddrTable, because channel destroyed and no broker in this cluster",
                    clusterName);
                //如果集群中不包含任何broker,則移除該集群
                it.remove();
            }

            break;
        }
    }
}/<string>/<string>/<entry>/<code>

<code>//維護topicQueueTable隊列
if (removeBrokerName) {
    //遍歷topicQueueTable
    Iterator<entry>>> itTopicQueueTable =
        this.topicQueueTable.entrySet().iterator();
    while (itTopicQueueTable.hasNext()) {
        Entry<string>> entry = itTopicQueueTable.next();
        //主題名稱
        String topic = entry.getKey();
        //隊列集合
        List<queuedata> queueDataList = entry.getValue();
\t\t//遍歷該主題隊列
        Iterator<queuedata> itQueueData = queueDataList.iterator();
        while (itQueueData.hasNext()) {
            //從隊列中移除為活躍broker信息
            QueueData queueData = itQueueData.next();
            if (queueData.getBrokerName().equals(brokerNameFound)) {
                itQueueData.remove();
                log.info("remove topic[{} {}], from topicQueueTable, because channel destroyed",
                    topic, queueData);
            }
        }
\t\t//如果該topic的隊列為空,則移除該topic
        if (queueDataList.isEmpty()) {
            itTopicQueueTable.remove();
            log.info("remove topic[{}] all queue, from topicQueueTable, because channel destroyed",
                topic);
        }
    }
}/<queuedata>/<queuedata>/<string>/<entry>/<code>

<code>//釋放寫鎖
finally {
    this.lock.writeLock().unlock();
}/<code>

2.2.3.4 路由發現

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

代碼：DefaultRequestProcessor#getRouteInfoByTopic

<code>public RemotingCommand getRouteInfoByTopic(ChannelHandlerContext ctx,
    RemotingCommand request) throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);
    final GetRouteInfoRequestHeader requestHeader =
        (GetRouteInfoRequestHeader) request.decodeCommandCustomHeader(GetRouteInfoRequestHeader.class);
\t//調用RouteInfoManager的方法,從路由表topicQueueTable、brokerAddrTable、filterServerTable中分別填充TopicRouteData的List<queuedata>、List<brokerdata>、filterServer
    TopicRouteData topicRouteData = this.namesrvController.getRouteInfoManager().pickupTopicRouteData(requestHeader.getTopic());
\t//如果找到主題對應你的路由信息並且該主題為順序消息，則從NameServer KVConfig中獲取關於順序消息相關的配置填充路由信息
    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;
}/<brokerdata>/<queuedata>/<code>

2.2.4 小結

閱讀更多 中間件小哥哥 的文章

關鍵字: 設計 源碼 算法