天天看點
傳回

RocketMQ筆記：Namesrv

Namesrv簡介

Broker

啟動時，會向所有的

Nameserver

注冊。

消息生産者發送消息前，先去

Nameserver

拉取

broker

伺服器位址清單。然後根據負載均衡算法選擇一台

Broker

進行消息發送。

NameServer

與

Broker

保持長連接配接，并每隔30s檢測

Broker

是否存活。

如果檢測到

Broker

當機，則從路由清單中将其移除。但不會馬上通知消息生産者。而是在消息發送端提供容錯機制來保證消息發送的高可用性。

NameServer

叢集之間并不互相通信，路由資訊并非是強一緻性。

NameServer啟動流程

  1. 加載配置檔案 
    // NamesrvConfig
final NamesrvConfig namesrvConfig = new NamesrvConfig();
// NettyServerConfig
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();
    }
}

    NameServerConfig

     的配置來源有兩個：
    • 通過

      -c

       指定配置檔案路徑
    • 使用

      -- 屬性名 屬性值

       來設定，如

      --listenPort 9876
    NamesrvConfig 
    public class NamesrvConfig {
    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;
}

    • rocketmqHome

       ：rocketmq主目錄，配置環境變量

    • kvConfigPath

       ：存儲KV配置的預設路徑

    • configStorePath

       ：

      nameserver

       預設配置檔案路徑

    • orderMessageEnable

       ：是否支援順序消息
    NettyServerConfig 
    public class NettyServerConfig implements Cloneable {
    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;
}

    • listenPort

       ：監聽端口

    • serverWorkerThread

       ：

      Netty

       業務線程池線程個數

    • serverCallbackExecutorThreads

       ：

      Netty

       公共的線程池線程個數。如果業務類型沒有指定線程池，則由

      public

       線程池執行

    • serverSelectorThreads

       ：IO線程池線程個數。用于解析請求，然後将任務轉發給對應的業務類型，再将結果傳回給調用方。

    • serverOnewaySemaphoreValue

       ：

      send oneway

       消息請求并發度（

      Broker

       ）

    • serverAsyncSemaphoreValue

       ：異步消息發送的最大并發度（

      Broker

       ）

    • serverChannelMaxIdleTimeSeconds

       ：網絡連接配接的最大空閑時間，如果連接配接空閑時間操作此數，連接配接将被關閉

    • serverSocketSndBufSize

       ：網絡

      socket

       發送緩存區大小，預設

      64k

    • serverSocketRcvBufSize

       ：網絡

      socket

       接收緩存區大小，預設

      64k

    • serverPooledByteBufAllocatorEnable

       ：

      ByteBuffer

       是否開啟緩存

    • useEpollNativeSelector

       ：是否啟用

      Epoll IO

       模型，Linux建議開啟
  2. 根據啟動屬性建立

    NamesrvController

     ，并初始化改執行個體。 
    public boolean initialize() {
	// 加載KV配置
    this.kvConfigManager.load();
	// 建立NameServer網絡處理對象
    this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);

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

    this.registerProcessor();
	// 每隔10s掃描一次Broker，移除不激活的Broker
    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

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

    // 每10分鐘列印一次KV配置
    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

        @Override
        public void run() {
            NamesrvController.this.kvConfigManager.printAllPeriodically();
        }
    }, 1, 10, TimeUnit.MINUTES);
	// 省略部分代碼
    return true;
}
  3. 注冊鈎子函數 
    // 注冊鈎子函數
Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<Void>() {
    @Override
    public Void call() throws Exception {
        // 關閉線程池，釋放資源
        controller.shutdown();
        return null;
    }
}));

路由注冊和故障剔除

NameServer

的主要作用是為消息生産者和消息消費者提供

Topic

的路由資訊。那麼

NameServer

需要存儲路由的基本資訊。包括路由注冊和剔除故障的

broker

路由元資訊

org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager

 
public class RouteInfoManager {
   	// Topic消息隊列路由資訊，消息發送時根據路由資訊進行負債均衡
    private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
    
    // Broker基礎資訊，包含brokerName、所屬叢集名稱、主備Broker位址
    private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
    
    // Broker 叢集資訊，存儲叢集中所有Broker的名稱
    private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
    
    // Broker狀态資訊，Namesrv收到資訊包後替換該資訊
    private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
    
    // Broker上的FilterServer清單，用于類模式消息過濾
    private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;
}

路由注冊

路由注冊是通過

NameServer

和

Broker

的心跳包功能實作的。

Broker

啟動時向叢集中所有的

NameServer

發送心跳語句。

每隔 30s 向叢集中所有

NameServer

發送心跳包，

NameServer

收到心跳包後，更新

BrokerLiveInfo

的

lastUpdatestamp

。

NameServer

每隔 10s 掃描一次

brokerLiveTable

，如果一個

Broker

連續120s沒有發送心跳包，将會被

NameServer

移除路由資訊，同時關閉

Socket

。

  • Broker發送心跳包

org.apache.rocketmq.broker.BrokerController#start

中的代碼片段：

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);
           
public List<RegisterBrokerResult> registerBrokerAll(
    final String clusterName,
    final String brokerAddr,
    final String brokerName,
    final long brokerId,
    final String haServerAddr,
    final TopicConfigSerializeWrapper topicConfigWrapper,
    final List<String> filterServerList,
    final boolean oneway,
    final int timeoutMills,
    final boolean compressed) {

    final List<RegisterBrokerResult> registerBrokerResultList = Lists.newArrayList();
    List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
    if (nameServerAddressList != null && nameServerAddressList.size() > 0) {

        final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();
        // 省略設定requestHeader代碼

        RegisterBrokerBody requestBody = new RegisterBrokerBody();
        // 省略處理body代碼
        requestHeader.setBodyCrc32(bodyCrc32);
        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) {
        }
    }

    return registerBrokerResultList;
}

周遊

NameServerList

，依次發送心跳包。

  • NameServer處理心跳包

請求到達

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

，通過請求的

RequestCode

類型分發。

心跳包的請求類型為

RequestCode.REGISTER_BROKER

，根據版本調用

registerBrokerWithFilterServer

或者

registerBroker

，這兩個方法都會調用

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

。

RegisterBrokerResult result = this.namesrvController.getRouteInfoManager().registerBroker(
    // 叢集名稱
    requestHeader.getClusterName(),
    // Broker位址
    requestHeader.getBrokerAddr(),
    requestHeader.getBrokerName(),
    requestHeader.getBrokerId(),
    // master位址
    requestHeader.getHaServerAddr(),
    // 主題配置
    topicConfigWrapper,
    null,
    ctx.channel()
);
  1. 如果broker所屬叢集不存在，則建立并加入到

    clusterAddrTable

     
// 加鎖，防止并發修改路由表
this.lock.writeLock().lockInterruptibly();

Set<String> brokerNames = this.clusterAddrTable.get(clusterName);
if (null == brokerNames) {
    // 如果broker所屬叢集不存在，則建立
    brokerNames = new HashSet<String>();
    // 并加入到集合中
    this.clusterAddrTable.put(clusterName, brokerNames);
}
brokerNames.add(brokerName);
  1. 如果

    brokerAddrTable

     不包含

    brokerName

     對應的

    brokerData

     ，則建立

    brokerData

     并加入到集合中 
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();
//Switch slave to master: first remove <1, IP:PORT> in namesrv, then add <0, IP:PORT>
//The same IP:PORT must only have one record in brokerAddrTable
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);
  1. 如果broker是Master，并且是初次注冊，則需要建立或更新Topic路由中繼資料 
if (null != topicConfigWrapper
    && MixAll.MASTER_ID == brokerId) {
    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());
            }
        }
    }
}

根據

topicConfig

建立

QueueData

，然後更新

topicUpdateTable

。

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: {}", topicConfig.getTopicName(), qd,
                        queueData);
                    it.remove();
                }
            }
        }

        if (addNewOne) {
            queueDataList.add(queueData);
        }
    }
}
  1. 更新

    BrokerInfo

     ，

    BrokerInfo

     中包含上次更新時間，是剔除無效

    Broker

     的重要依據。 
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);
}
  1. 注冊

    Broker

     的過濾器

    Server

     位址清單，一個

    Broker

     上會關聯多個

    FilterServer

     消息過濾伺服器。 
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);
        }
    }
}

TopicQueueTable：Topic消息隊列路由資訊

BrokerAddrTable：Broker基礎資訊，包括brokerName、叢集名稱、主備Broker位址

ClusterAddrTable：Broker叢集資訊，存儲叢集中所有Broker名稱

BrokerLiveTable：存儲brokerLiveInfo，是剔除無效broker的關鍵資料

FilterServerTable：Broker上的FilterServer清單

路由剔除

路由剔除有兩個觸發點：

  • NameServer

     掃描發現上次更新的時間與目前時間差大于120s

  • Broker

     在正常關閉的情況下，會執行

    unRegisterBroker

     
public void scanNotActiveBroker() {
    Iterator<Entry<String, BrokerLiveInfo>> it = this.brokerLiveTable.entrySet().iterator();
    while (it.hasNext()) {
        Entry<String, BrokerLiveInfo> next = it.next();
        long last = next.getValue().getLastUpdateTimestamp();
        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);
            // 關閉通道
            this.onChannelDestroy(next.getKey(), next.getValue().getChannel());
        }
    }
}
  1. 找到要删除的

    broker

     。上寫鎖，從

    brokerLiveTable

     和

    filterServerTable

     删除

    broker

     。 
Iterator<Entry<String, BrokerLiveInfo>> itBrokerLiveTable =
    this.brokerLiveTable.entrySet().iterator();
while (itBrokerLiveTable.hasNext()) {
    Entry<String, BrokerLiveInfo> entry = itBrokerLiveTable.next();
    if (entry.getValue().getChannel() == channel) {
        brokerAddrFound = entry.getKey();
        break;
    }
}

this.lock.writeLock().lockInterruptibly();
this.brokerLiveTable.remove(brokerAddrFound);
this.filterServerTable.remove(brokerAddrFound);
  1. 維護

    brokerAddrTable

     ，周遊

    BrokerAddrTable

     ，從

    brokerAddrs

     找到對應的

    broker

     ，找到具體的

    Broker

     ，并在

    brokerData

     中删除。如果

    brokerData

     不包含其他的

    brokerAddr

     ，則從

    brokerAddrTable

     中删除。 
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",
                brokerId, brokerAddr);
            break;
        }
    }

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

    brokerName

     ，從

    clusterAddrTable

     中找到

    Broker

     并移除。如果移除後，

    brokerNames

     為空，那麼從将該叢集從

    clusterAddrTable

     中删除。 
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, 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);
                it.remove();
            }

            break;
        }
    }
}
  1. 删除

    topicQueueTable

     中的對應的

    queueData

     資料，如果該路由表為空，則删除。 
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, because channel destroyed",
                    topic, queueData);
            }
        }

        if (queueDataList.isEmpty()) {
            itTopicQueueTable.remove();
            log.info("remove topic[{}] all queue, from topicQueueTable, because channel destroyed",
                topic);
        }
    }
}
  1. 釋放鎖 
finally {
    this.lock.writeLock().unlock();
}

路由發現

RocketMQ的路由資訊不是實時的，是有消息生産者從

NameServer

拉取路由資訊。拉取路由資訊的編碼為

RequestCode.GET_ROUTEINTO_BY_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 = 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;
}

通過

RouteInfoManager

擷取

topicRouteData

，如果

topicRouteData

存在，并且是

OrderMessageEnable

的，則從

KvConfigManager

擷取

KVConfig

中的

NAMESPACE_ORDER_TOPIC_CONFIG

值，并填充到

topicRouteData

的

OrderTopicConf

中。

mq java RocketMQ nameSrv
上一篇: 遊戲服務端開發-排行榜開頭結構排序時間間隔實作性能提升
下一篇: RocketMq Namesrv 啟動源碼2_NamesrvController.start()

繼續閱讀