1. NameServer 解决什么问题

为避免消息服务器因单点故障导致整个系统瘫痪，通常会部署多台消息服务器共同承担消息的存储，那么就会存在几个问题，生产者如何知道将消息发送给哪台消息服务器呢？某台消息服务器重启宕机了，生产者如何在不重启服务器的情况下感知呢？所以RocketMQ 引入NameServer 来解决这些问题

2. NameServer 架构设计

NameServer 路由中心 - 图1

Broker 消息服务器在启动时会向所有NameServer 注册，生产者在发送消息之前会先从NameServer获取Broker 服务器的地址列表，然后根据负载算法从列表中选择一台消息服务器发送消息。NameServer与每台Broker服务器保持长连接，并间隔10s检测Broker是否存活，如果检测道Broker宕机会从NameServer中的注册表中删除该节点，但是路由变化不会立刻通知生产者，这是为了降低NameServer 实现的复杂性，因此需要在消息发送端提供容错机制来保证消息发送的高可用性。

NameServer 本身高可用性可通过部署多台NameServer服务器来实现，但彼此之间互不通信。这样虽然在某一个时刻数据并不会完全相同，但对消息发送不会造成重大影响，无非就是短暂造成消息发送不均衡，这也是RocketMQ NameServer 设计的一个亮点。

消息客户端和NameServer、Broker 的交互设计特点如下：

  1. Broker 每个30s向NameServer集群的每一台机器发送心跳包，包含自身创建的topic路由等信息。
  2. 消息客户端每个30s向NameServer更新对应的topic的路由信息
  3. NameServer 收到Broker 发送的心跳包，会记录时间戳放到brokerLiveTable中，然后每隔10s扫描一次brokerLiveTable，如果在120s以内没有收到心跳包就认为Broker失效，更新topic路由信息，将失效的Broker信息移除

3. NameServer 启动

3.1. 时序图

NameServer 路由中心 - 图2

3.2. 源码分析

NameServer的启动入口类是org.apache.rocketmq.namesrv.NamesrvStartup

  1. public class NamesrvStartup {
  2.     public static NamesrvController main0(String[] args) {
  4.         try {
  5.             // 创建NamesrvController
  6.             NamesrvController controller = createNamesrvController(args);
  7.             // 启动NamesrvController
  8.             start(controller);
  9.             String tip = "The Name Server boot success. serializeType=" + RemotingCommand.getSerializeTypeConfigInThisServer();
  10.             log.info(tip);
  11.             System.out.printf("%s%n", tip);
  12.             return controller;
  13.         } catch (Throwable e) {
  14.             e.printStackTrace();
  15.             System.exit(-1);
  16.         }
  18.         return null;
  19.     }
  20. }

3.2.1. 创建NamesrvController

org.apache.rocketmq.namesrv.NamesrvStartup.createNamesrvController() 方法创建NamesrvController

  1. public static NamesrvController createNamesrvController(String[] args) throws IOException, JoranException {
  2.         // 设置版本号为当前版本号
  3.         System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
  4.         //PackageConflictDetect.detectFastjson();
  5.         //构造org.apache.commons.cli.Options,并添加-h -n参数，-h参数是打印帮助信息，-n参数是指定namesrvAddr
  6.         Options options = ServerUtil.buildCommandlineOptions(new Options());
  7.         //初始化commandLine，并在options中添加-c -p参数，-c指定nameserver的配置文件路径，-p标识打印配置信息
  8.         commandLine = ServerUtil.parseCmdLine("mqnamesrv", args, buildCommandlineOptions(options), new PosixParser());
  9.         if (null == commandLine) {
  10.             System.exit(-1);
  11.             return null;
  12.         }
  14.         // 创建NamesrvConfig 对象，包含业务参数
  15.         final NamesrvConfig namesrvConfig = new NamesrvConfig();
  16.         // 创建NettyServerConfig 对象，包含网络参数
  17.         final NettyServerConfig nettyServerConfig = new NettyServerConfig();
  18.         // 指定NamesrvServer 端口，默认9876
  19.         nettyServerConfig.setListenPort(9876);
  20.         if (commandLine.hasOption('c')) {
  21.             String file = commandLine.getOptionValue('c');
  22.             if (file != null) {
  23.                 InputStream in = new BufferedInputStream(new FileInputStream(file));
  24.                 properties = new Properties();
  25.                 properties.load(in);
  26.                 MixAll.properties2Object(properties, namesrvConfig);
  27.                 MixAll.properties2Object(properties, nettyServerConfig);
  28.                 // 配置文件路径
  29.                 namesrvConfig.setConfigStorePath(file);
  31.                 System.out.printf("load config properties file OK, %s%n", file);
  32.                 in.close();
  33.             }
  34.         }
  35.         //命令行带有-p，说明是打印参数的命令，那么就打印出NamesrvConfig和NettyServerConfig的属性
  36.         // 在启动NameServer时可以先使用./mqnameserver -c configFile -p打印当前加载的配置属性
  37.         if (commandLine.hasOption('p')) {
  38.             InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_CONSOLE_NAME);
  39.             MixAll.printObjectProperties(console, namesrvConfig);
  40.             MixAll.printObjectProperties(console, nettyServerConfig);
  41.             //打印参数命令不需要启动nameserver服务，只需要打印参数即可
  42.             System.exit(0);
  43.         }
  44.         //解析命令行参数，并加载到namesrvConfig中
  45.         MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);
  46.         //检查ROCKETMQ_HOME，不能为空
  47.         if (null == namesrvConfig.getRocketmqHome()) {
  48.             System.out.printf("Please set the %s variable in your environment to match the location of the RocketMQ installation%n", MixAll.ROCKETMQ_HOME_ENV);
  49.             System.exit(-2);
  50.         }
  51.         //初始化logback日志工厂，rocketmq默认使用logback作为日志输出
  52.         LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
  53.         JoranConfigurator configurator = new JoranConfigurator();
  54.         configurator.setContext(lc);
  55.         lc.reset();
  56.         configurator.doConfigure(namesrvConfig.getRocketmqHome() + "/conf/logback_namesrv.xml");
  58.         log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
  60.         MixAll.printObjectProperties(log, namesrvConfig);
  61.         MixAll.printObjectProperties(log, nettyServerConfig);
  63.         // 根据namesrvConfig和nettyServerConfig 创建NamesrvController 对象
  64.         final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);
  66.         // remember all configs to prevent discard
  67.         // 合并配置文件
  68.         controller.getConfiguration().registerConfig(properties);
  70.         return controller;
  71.     }

创建NamesrvController主要分为两步

第一步：通过命令行中获取配置。赋值给NamesrvConfig和NettyServerConfig类 第二步：根据配置类NamesrvConfig和NettyServerConfig构造一个NamesrvController实例

NamesrvConfig 是NameServer的配置信息，NeetyServerConfig是NameServer的网络配置信息

NamesrvConfig的配置信息

  1. public class NamesrvConfig {
  2.     private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
  3.     // RocketMQ 主目录，通过-Drocketmq.home.dir=path 或设置环境变量ROCKETMQ_HOME 可以配置RocketMQ的主目录
  4.     private String rocketmqHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY, System.getenv(MixAll.ROCKETMQ_HOME_ENV));
  5.     // NameServer存储KV配置属性的持久化路径
  6.     private String kvConfigPath = System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "kvConfig.json";
  7.     // NameServer 默认配置文件路径，NameServer 启动时如果要通过配置文件配置NameServer启动属性，使用-c 选项
  8.     private String configStorePath = System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "namesrv.properties";
  9.     private String productEnvName = "center";
  10.     private boolean clusterTest = false;
  11.     // 支持顺序消息，默认不支持
  12.     private boolean orderMessageEnable = false;

NettyServerConfig的配置信息

  1. public class NettyServerConfig implements Cloneable {
  2.     // 默认监听端口，在启动时会被初始化为9876
  3.     private int listenPort = 8888;
  4.     // Netty 业务线程池线程个数
  5.     private int serverWorkerThreads = 8;
  6.     // Netty public 任务线程池线程个数，Netty 网络会根据业务类型创建不同的线程池，比如处理消息发送、消息消费、心跳检测等。如果该业务类型（RequestCode）未注册线程池，则由public 线程池执行
  7.     private int serverCallbackExecutorThreads = 0;
  8.     // I/O线程池个数，主要是NameServer、Broker端解析请求、返回相应线程个数。这类线程主要用于处理网络请求，先解析请求包，然后转发到各业务线程池完成具体的业务操作，最后将结果返回给调用方
  9.     private int serverSelectorThreads = 3;
  10.     // send oneway 消息请求的并发度（Broker 端参数）
  11.     private int serverOnewaySemaphoreValue = 256;
  12.     private int serverAsyncSemaphoreValue = 64;
  13.     // 网络链接最大空闲时间，默认为120s，如果链接空闲时间超过该参数设置的值，链接将被关闭
  14.     private int serverChannelMaxIdleTimeSeconds = 120;
  15.     // 网络socket 发送缓存区大小，默认为64KB
  16.     private int serverSocketSndBufSize = NettySystemConfig.socketSndbufSize;
  17.     // 网络socket 接收缓存区大小，默认为64KB
  18.     private int serverSocketRcvBufSize = NettySystemConfig.socketRcvbufSize;
  19.     private int writeBufferHighWaterMark = NettySystemConfig.writeBufferHighWaterMark;
  20.     private int writeBufferLowWaterMark = NettySystemConfig.writeBufferLowWaterMark;
  21.     private int serverSocketBacklog = NettySystemConfig.socketBacklog;
  22.     // Buffer 是否开启缓存，默认开启
  23.     private boolean serverPooledByteBufAllocatorEnable = true;
  25.     /**
  26.      * make make install
  27.      *
  28.      *
  29.      * ../glibc-2.10.1/configure \ --prefix=/usr \ --with-headers=/usr/include \
  30.      * --host=x86_64-linux-gnu \ --build=x86_64-pc-linux-gnu \ --without-gd
  31.      */
  32.     // 是否启用Epool I/O模型，Linux 环境下建议开启
  33.     private boolean useEpollNativeSelector = false;
  35.     // 省略Set/Get
  36. }

3.2.2. 初始化NamesrvController

创建好NamesrvController对象后，接下来就是对它进行初始化。org.apache.rocketmq.namesrv.initialize()初始化NamesrvController

  1.  public boolean initialize() {
  2.         // 加载KV配置，创建NettyServer 网络处理对象
  3.         this.kvConfigManager.load();
  4.         // 根据nettyServerConfig初始化一个netty服务器
  5.         //brokerHousekeepingService是在NamesrvController实例化时构造函数里实例化的，该类负责Broker连接事件的处理，实现了ChannelEventListener，主要用来管理RouteInfoManager的brokerLiveTable
  6.         this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
  7.         //初始化负责处理Netty网络交互数据的线程池，默认线程数是8个
  8.         this.remotingExecutor =
  9.             Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));
  10.         //注册Netty服务端业务处理逻辑，如果开启了clusterTest，那么注册的请求处理类是ClusterTestRequestProcessor，否则请求处理类是DefaultRequestProcessor
  11.         this.registerProcessor();
  13.         // 定时任务（心跳检测）每隔10s扫描一次Broker，移除处于未激活状态的Broker
  14.         this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
  16.             @Override
  17.             public void run() {
  18.                 NamesrvController.this.routeInfoManager.scanNotActiveBroker();
  19.             }
  20.         }, 5, 10, TimeUnit.SECONDS);
  22.         // 定时任务（心跳检测）每隔10min 打印一次KV 配置
  23.         this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
  25.             @Override
  26.             public void run() {
  27.                 NamesrvController.this.kvConfigManager.printAllPeriodically();
  28.             }
  29.         }, 1, 10, TimeUnit.MINUTES);
  31.         if (TlsSystemConfig.tlsMode != TlsMode.DISABLED) {
  32.             // Register a listener to reload SslContext
  33.             try {
  34.                 fileWatchService = new FileWatchService(
  35.                     new String[] {
  36.                         TlsSystemConfig.tlsServerCertPath,
  37.                         TlsSystemConfig.tlsServerKeyPath,
  38.                         TlsSystemConfig.tlsServerTrustCertPath
  39.                     },
  40.                     new FileWatchService.Listener() {
  41.                         boolean certChanged, keyChanged = false;
  42.                         @Override
  43.                         public void onChanged(String path) {
  44.                             if (path.equals(TlsSystemConfig.tlsServerTrustCertPath)) {
  45.                                 log.info("The trust certificate changed, reload the ssl context");
  46.                                 reloadServerSslContext();
  47.                             }
  48.                             if (path.equals(TlsSystemConfig.tlsServerCertPath)) {
  49.                                 certChanged = true;
  50.                             }
  51.                             if (path.equals(TlsSystemConfig.tlsServerKeyPath)) {
  52.                                 keyChanged = true;
  53.                             }
  54.                             if (certChanged && keyChanged) {
  55.                                 log.info("The certificate and private key changed, reload the ssl context");
  56.                                 certChanged = keyChanged = false;
  57.                                 reloadServerSslContext();
  58.                             }
  59.                         }
  60.                         private void reloadServerSslContext() {
  61.                             ((NettyRemotingServer) remotingServer).loadSslContext();
  62.                         }
  63.                     });
  64.             } catch (Exception e) {
  65.                 log.warn("FileWatchService created error, can't load the certificate dynamically");
  66.             }
  67.         }
  69.         return true;
  70.     }

4. 心跳检测（路由删除）

4.1. 时序图

NameServer 路由中心 - 图3

4.2. 源码分析

心跳检测的源码入口org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#scanNotActiveBroker()；
每隔10s执行一次，遍历brokerLiveInfo路由表，HashMap结构。检测BrokerLiveInfo的LastUpdateTimestamp上次收到的心跳包时间，如果超过120s，就认为Broker不可用，然后从brokerLiveInfo路由表中移除该Broker。

  1. public void scanNotActiveBroker() {
  2.         // 遍历brokerLiveTable，这是Broker 状态信息
  3.         Iterator<Entry<String, BrokerLiveInfo>> it = this.brokerLiveTable.entrySet().iterator();
  4.         // 遍历Broker 信息
  5.         while (it.hasNext()) {
  6.             Entry<String, BrokerLiveInfo> next = it.next();
  7.             // Broker 最后一次注册的心跳时间
  8.             long last = next.getValue().getLastUpdateTimestamp();
  9.             // 最后一次注册时间大于120s，从brokerLiveTable 移除Broker
  10.             if ((last + BROKER_CHANNEL_EXPIRED_TIME) < System.currentTimeMillis()) {
  11.                 RemotingUtil.closeChannel(next.getValue().getChannel());
  12.                 it.remove();
  13.                 log.warn("The broker channel expired, {} {}ms", next.getKey(), BROKER_CHANNEL_EXPIRED_TIME);
  14.                 this.onChannelDestroy(next.getKey(), next.getValue().getChannel());
  15.             }
  16.         }
  17.     }

从brokerLiveTable 移除Broker后，调用onChannelDestroy()删除与Broker相关的路由信息

  1. public void onChannelDestroy(String remoteAddr, Channel channel) {
  2.         String brokerAddrFound = null;
  3.         if (channel != null) {
  4.             try {
  5.                 try {
  6.                     // 加锁
  7.                     this.lock.readLock().lockInterruptibly();
  8.                     Iterator<Entry<String, BrokerLiveInfo>> itBrokerLiveTable =
  9.                         this.brokerLiveTable.entrySet().iterator();
  10.                     // 对比channel 是否一样
  11.                     while (itBrokerLiveTable.hasNext()) {
  12.                         Entry<String, BrokerLiveInfo> entry = itBrokerLiveTable.next();
  13.                         if (entry.getValue().getChannel() == channel) {
  14.                             brokerAddrFound = entry.getKey();
  15.                             break;
  16.                         }
  17.                     }
  18.                 } finally {
  19.                     this.lock.readLock().unlock();
  20.                 }
  21.             } catch (Exception e) {
  22.                 log.error("onChannelDestroy Exception", e);
  23.             }
  24.         }
  26.         if (null == brokerAddrFound) {
  27.             brokerAddrFound = remoteAddr;
  28.         } else {
  29.             log.info("the broker's channel destroyed, {}, clean it's data structure at once", brokerAddrFound);
  30.         }
  32.         if (brokerAddrFound != null && brokerAddrFound.length() > 0) {
  34.             try {
  35.                 try {
  36.                     // Step1:申请写锁，根据brokerLiveTable、filterServerTable中移除Broker相关信息
  37.                     this.lock.writeLock().lockInterruptibly();
  38.                     // 从brokerLiveTable删除Broker
  39.                     this.brokerLiveTable.remove(brokerAddrFound);
  40.                     // 从filterServerTable 删除Broker
  41.                     this.filterServerTable.remove(brokerAddrFound);
  42.                     // Broker名称
  43.                     String brokerNameFound = null;
  44.                     // Broker 删除标志位，默认没删除
  45.                     boolean removeBrokerName = false;
  47.                     // Step2：维护brokerAddrTable
  48.                     // 遍历brokerAddTable，从BrokerData的HashMap中找到具体的Broker，从BrokerData中将其移除。如果移除后BrokerData中不在包含其他Broker
  49.                     // 就在brokerAddrTable中移除该brokerName对应的条目
  50.                     Iterator<Entry<String, BrokerData>> itBrokerAddrTable =
  51.                         this.brokerAddrTable.entrySet().iterator();
  52.                     while (itBrokerAddrTable.hasNext() && (null == brokerNameFound)) {
  53.                         BrokerData brokerData = itBrokerAddrTable.next().getValue();
  55.                         Iterator<Entry<Long, String>> it = brokerData.getBrokerAddrs().entrySet().iterator();
  56.                         while (it.hasNext()) {
  57.                             Entry<Long, String> entry = it.next();
  58.                             Long brokerId = entry.getKey();
  59.                             String brokerAddr = entry.getValue();
  60.                             if (brokerAddr.equals(brokerAddrFound)) {
  61.                                 brokerNameFound = brokerData.getBrokerName();
  62.                                 it.remove();
  63.                                 log.info("remove brokerAddr[{}, {}] from brokerAddrTable, because channel destroyed",
  64.                                     brokerId, brokerAddr);
  65.                                 break;
  66.                             }
  67.                         }
  69.                         if (brokerData.getBrokerAddrs().isEmpty()) {
  70.                             removeBrokerName = true;
  71.                             itBrokerAddrTable.remove();
  72.                             log.info("remove brokerName[{}] from brokerAddrTable, because channel destroyed",
  73.                                 brokerData.getBrokerName());
  74.                         }
  75.                     }
  76.                     // Step3：根据BrokerName从clusterAddrTable中找到Broker并将其从集群中移除
  77.                     // 如果移除后，集群中不包含任何Broker，就将该集群中clusterAddrTable中移除
  78.                     if (brokerNameFound != null && removeBrokerName) {
  79.                         // 遍历Broker集群
  80.                         Iterator<Entry<String, Set<String>>> it = this.clusterAddrTable.entrySet().iterator();
  81.                         while (it.hasNext()) {
  82.                             Entry<String, Set<String>> entry = it.next();
  83.                             String clusterName = entry.getKey();
  84.                             Set<String> brokerNames = entry.getValue();
  85.                             boolean removed = brokerNames.remove(brokerNameFound);
  86.                             if (removed) {
  87.                                 log.info("remove brokerName[{}], clusterName[{}] from clusterAddrTable, because channel destroyed",
  88.                                     brokerNameFound, clusterName);
  89.                                 // 如果集群中已经不包含任何Broker，移除该集群
  90.                                 if (brokerNames.isEmpty()) {
  91.                                     log.info("remove the clusterName[{}] from clusterAddrTable, because channel destroyed and no broker in this cluster",
  92.                                         clusterName);
  93.                                     it.remove();
  94.                                 }
  96.                                 break;
  97.                             }
  98.                         }
  99.                     }
  100.                     // Step4:根据BrokerName，遍历所有主题队列，如果队列中包含当前Broker的队列，就移除
  101.                     // 如果topic中只包含待移除的Broker队列，就从路由表中删除该topic
  102.                     if (removeBrokerName) {
  103.                         // 遍历所有主题队列
  104.                         Iterator<Entry<String, List<QueueData>>> itTopicQueueTable =
  105.                             this.topicQueueTable.entrySet().iterator();
  106.                         while (itTopicQueueTable.hasNext()) {
  107.                             Entry<String, List<QueueData>> entry = itTopicQueueTable.next();
  108.                             String topic = entry.getKey();
  109.                             List<QueueData> queueDataList = entry.getValue();
  110.                             // 主题的所有队列信息
  111.                             Iterator<QueueData> itQueueData = queueDataList.iterator();
  112.                             while (itQueueData.hasNext()) {
  113.                                 // 主题的某个队列信息
  114.                                 QueueData queueData = itQueueData.next();
  115.                                 // 队列所属Broker名称和被移除的Broker名称一样，就移除该队列
  116.                                 if (queueData.getBrokerName().equals(brokerNameFound)) {
  117.                                     itQueueData.remove();
  118.                                     log.info("remove topic[{} {}], from topicQueueTable, because channel destroyed",
  119.                                         topic, queueData);
  120.                                 }
  121.                             }
  122.                             // 如果队列已经为空，就移除主题
  123.                             if (queueDataList.isEmpty()) {
  124.                                 itTopicQueueTable.remove();
  125.                                 log.info("remove topic[{}] all queue, from topicQueueTable, because channel destroyed",
  126.                                     topic);
  127.                             }
  128.                         }
  129.                     }
  130.                 } finally {
  131.                     // Step5:释放锁
  132.                     this.lock.writeLock().unlock();
  133.                 }
  134.             } catch (Exception e) {
  135.                 log.error("onChannelDestroy Exception", e);
  136.             }
  137.         }
  138.     }

5. Broker 心跳注册（路由注册）

5.1. 时序图

NameServer 路由中心 - 图4

5.2. 路由元数据信息

Broker 会把自己的信息注册到Nameserver中，注册的信息就叫路由元数据信息，具体注册信息如下：

  1. // 1. Topic 消息队列的路由信息，消息发送时根据路由表进行负载均衡
  2. private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
  3. // 2. Broker 基础信息，包含brokerName、所属集群名称、主备Broker地址
  4. private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
  5. // 3. Broker集群信息
  6. private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
  7. // 4. Broker 状态信息，NameServer 每次收到心跳包时会替换该信息
  8. private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
  9. // 5. Broker 上的FilterServer列表，用于类模式消息过滤
  10. private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;
  1. topicQueueTable：topic 消息队列的路由信息，消息发送时根据路由表进行负债均衡
  2. brokerAddrTable：Broker基本信息，包含brokerName、所属集群名称、主备Broker地址
  3. clusterAddrTable：Broker集群信息，存储集群中所有Broker 的名称
  4. brokerLiveTable：Broker 状态信息，NameServer 每次收到心跳包（注册Broker信息）会替换该值

  5. filterServerTable：Broker上的FilterServer列表，用于类模式消息过滤。

    5.3. 源码分析

    5.3.1. Broker 发送心跳包（注册）

    Broker 会每隔10s向Nameserver 发送一次心跳，发送心跳的入口代码是：org.apache.rocketmq.broker#start()
    Broker启动后会创建一个线程，在程序启动10秒后执行，每隔30秒（默认30s，时间间隔在10秒到60秒之间，BrokerConfig.getRegisterNameServerPeriod()的默认值是30秒）执行一次

    1.      // 向NameServer 发送心跳包，每隔10s发送一次
    2.      this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    4.          @Override
    5.          public void run() {
    6.              try {
    7.                  BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister());
    8.              } catch (Throwable e) {
    9.                  log.error("registerBrokerAll Exception", e);
    10.              }
    11.          }
    12.      }, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), TimeUnit.MILLISECONDS);

    这里主要是组装一些心跳包检测的必要参数

    1. public synchronized void registerBrokerAll(final boolean checkOrderConfig, boolean oneway, boolean forceRegister) {
    2.      // 构建Topic信息
    3.      TopicConfigSerializeWrapper topicConfigWrapper = this.getTopicConfigManager().buildTopicConfigSerializeWrapper();
    5.      if (!PermName.isWriteable(this.getBrokerConfig().getBrokerPermission())
    6.          || !PermName.isReadable(this.getBrokerConfig().getBrokerPermission())) {
    7.          ConcurrentHashMap<String, TopicConfig> topicConfigTable = new ConcurrentHashMap<String, TopicConfig>();
    8.          for (TopicConfig topicConfig : topicConfigWrapper.getTopicConfigTable().values()) {
    9.              TopicConfig tmp =
    10.                  new TopicConfig(topicConfig.getTopicName(), topicConfig.getReadQueueNums(), topicConfig.getWriteQueueNums(),
    11.                      this.brokerConfig.getBrokerPermission());
    12.              topicConfigTable.put(topicConfig.getTopicName(), tmp);
    13.          }
    14.          topicConfigWrapper.setTopicConfigTable(topicConfigTable);
    15.      }
    17.      // 强制注册或者需要注册时，向NameServer 注册Broker信息
    18.      if (forceRegister || needRegister(this.brokerConfig.getBrokerClusterName(),
    19.          this.getBrokerAddr(),
    20.          this.brokerConfig.getBrokerName(),
    21.          this.brokerConfig.getBrokerId(),
    22.          this.brokerConfig.getRegisterBrokerTimeoutMills())) {
    23.          // 发送心跳包
    24.          doRegisterBrokerAll(checkOrderConfig, oneway, topicConfigWrapper);
    25.      }
    26.  }

    心跳包的发送

    1. /**
    2.   * 向NameServer 注册Broker
    3.   * @param checkOrderConfig
    4.   * @param oneway
    5.   * @param topicConfigWrapper
    6.   */
    7.  private void doRegisterBrokerAll(boolean checkOrderConfig, boolean oneway,
    8.      TopicConfigSerializeWrapper topicConfigWrapper) {
    9.      // 发送心跳包
    10.      List<RegisterBrokerResult> registerBrokerResultList = this.brokerOuterAPI.registerBrokerAll(
    11.          this.brokerConfig.getBrokerClusterName(),
    12.          this.getBrokerAddr(),
    13.          this.brokerConfig.getBrokerName(),
    14.          this.brokerConfig.getBrokerId(),
    15.          this.getHAServerAddr(),
    16.          topicConfigWrapper,
    17.          this.filterServerManager.buildNewFilterServerList(),
    18.          oneway,
    19.          this.brokerConfig.getRegisterBrokerTimeoutMills(),
    20.          this.brokerConfig.isCompressedRegister());
    22.      if (registerBrokerResultList.size() > 0) {
    23.          RegisterBrokerResult registerBrokerResult = registerBrokerResultList.get(0);
    24.          if (registerBrokerResult != null) {
    25.              if (this.updateMasterHAServerAddrPeriodically && registerBrokerResult.getHaServerAddr() != null) {
    26.                  this.messageStore.updateHaMasterAddress(registerBrokerResult.getHaServerAddr());
    27.              }
    29.              this.slaveSynchronize.setMasterAddr(registerBrokerResult.getMasterAddr());
    31.              if (checkOrderConfig) {
    32.                  this.getTopicConfigManager().updateOrderTopicConfig(registerBrokerResult.getKvTable());
    33.              }
    34.          }
    35.      }
    36.  }

    Broker 心跳包的发送最终由org.apache.rocketmq.broker.out.BrokerOuterAPI#registerBrokerAll()方法实现发送，遍历所有Nameserver，依次向每台Nameserver发送心跳
    发送心跳包的具体逻辑：

  6. brokerAddr： broker地址

  7. brokerId：brokerId=0表示主节点，brokerId>0表示从节点
  8. brokerName：broker 名称
  9. clusterName ：集群名称
  10. haServerAddr：主节点地址，初次请求时该值为空，从节点向NameServer注册后返回
  11. requestBody 心跳包参数对象，包含topicConfigSerializeWrapper属性，这个属性内部封装的是TopicConfigManager中的topicConfigTable属性，是一个HashMap，topicConfigTable属性存储的是Broker启动时默认的一些topic

  12. filterServerList：消息过滤服务器列表

    1. /**
    2.   * 发送心跳包
    3.   * @param clusterName           集群名称
    4.   * @param brokerAddr            broker地址
    5.   * @param brokerName            broker名称
    6.   * @param brokerId              brokerId=0表示主节点，brokerId>0表示从节点
    7.   * @param haServerAddr          主节点地址，初次请求时该值为空，从节点向NameServer注册后返回
    8.   * @param topicConfigWrapper    主题配置，内部封装的是TopicConfigManager中的topicConfigTable，内部存储的是Broker启动时默认的一些topic
    9.   *                              如：MinA11.SELF_TEST_TOPIC、MixA11.DEFAULT_TOPIC（AutoCreateTopicEnable=true
    10.   * @param filterServerList      消息过滤服务列表
    11.   * @param oneway
    12.   * @param timeoutMills
    13.   * @param compressed
    14.   * @return
    15.   */
    16.  public List<RegisterBrokerResult> registerBrokerAll(
    17.      final String clusterName,
    18.      final String brokerAddr,
    19.      final String brokerName,
    20.      final long brokerId,
    21.      final String haServerAddr,
    22.      final TopicConfigSerializeWrapper topicConfigWrapper,
    23.      final List<String> filterServerList,
    24.      final boolean oneway,
    25.      final int timeoutMills,
    26.      final boolean compressed) {
    27.      // 向NameServer发送心跳包的结果结合
    28.      final List<RegisterBrokerResult> registerBrokerResultList = new CopyOnWriteArrayList<>();
    29.      // 获取所有NameServer列表
    30.      List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
    31.      if (nameServerAddressList != null && nameServerAddressList.size() > 0) {
    33.          final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();
    34.          requestHeader.setBrokerAddr(brokerAddr);
    35.          requestHeader.setBrokerId(brokerId);
    36.          requestHeader.setBrokerName(brokerName);
    37.          requestHeader.setClusterName(clusterName);
    38.          requestHeader.setHaServerAddr(haServerAddr);
    39.          requestHeader.setCompressed(compressed);
    40.          // 向NameServer 发送心跳包的参数
    41.          RegisterBrokerBody requestBody = new RegisterBrokerBody();
    42.          //主题配置，内部封装的是TopicConfigManager中的topicConfigTable，内部存储的是Broker启动时默认的一些topic
    43.          // 如：MinA11.SELF_TEST_TOPIC、MixA11.DEFAULT_TOPIC（AutoCreateTopicEnable=true
    44.          requestBody.setTopicConfigSerializeWrapper(topicConfigWrapper);
    45.          // 消息过滤服务器列表
    46.          requestBody.setFilterServerList(filterServerList);
    47.          final byte[] body = requestBody.encode(compressed);
    48.          final int bodyCrc32 = UtilAll.crc32(body);
    49.          requestHeader.setBodyCrc32(bodyCrc32);
    50.          final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());
    51.          // 遍历所有NameServer，依次向每个NameServer发送心跳包
    52.          for (final String namesrvAddr : nameServerAddressList) {
    53.              brokerOuterExecutor.execute(new Runnable() {
    54.                  @Override
    55.                  public void run() {
    56.                      try {
    57.                          // 向NameServer发送心跳，返回心跳发送结果
    58.                          RegisterBrokerResult result = registerBroker(namesrvAddr, oneway, timeoutMills, requestHeader, body);
    59.                          // 将发送结果放到registerBrokerResultList 集合
    60.                          if (result != null) {
    61.                              registerBrokerResultList.add(result);
    62.                          }
    64.                          log.info("register broker[{}]to name server {} OK", brokerId, namesrvAddr);
    65.                      } catch (Exception e) {
    66.                          log.warn("registerBroker Exception, {}", namesrvAddr, e);
    67.                      } finally {
    68.                          countDownLatch.countDown();
    69.                      }
    70.                  }
    71.              });
    72.          }
    74.          try {
    75.              // 等待Broker 注冊，如果注册超时抛异常
    76.              countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS);
    77.          } catch (InterruptedException e) {
    78.          }
    79.      }
    81.      return registerBrokerResultList;
    82.  }

    路由注册最终通过registerBroker()方法实现

    1. private RegisterBrokerResult registerBroker(
    2.      final String namesrvAddr,
    3.      final boolean oneway,
    4.      final int timeoutMills,
    5.      final RegisterBrokerRequestHeader requestHeader,
    6.      final byte[] body
    7.  ) throws RemotingCommandException, MQBrokerException, RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException,
    8.      InterruptedException {
    9.      // 注册Broker 的Netty请求对象，Broker 注册的requestCode=RequestCode.REGISTER_BROKER
    10.      // Nameserver 端的接收请求后，会根据requestCode 转发给不同的业务处理
    11.      RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.REGISTER_BROKER, requestHeader);
    12.      request.setBody(body);
    14.      if (oneway) {
    15.          try {
    16.              // 单向调用，没有返回值，不返回nameserver 返回结果
    17.              this.remotingClient.invokeOneway(namesrvAddr, request, timeoutMills);
    18.          } catch (RemotingTooMuchRequestException e) {
    19.              // Ignore
    20.          }
    21.          return null;
    22.      }
    23.      // 异步调用向nameserver注册Broker，会有返回值
    24.      RemotingCommand response = this.remotingClient.invokeSync(namesrvAddr, request, timeoutMills);
    25.      assert response != null;
    26.      switch (response.getCode()) {
    27.          case ResponseCode.SUCCESS: {
    28.              // 注册成功后，获取返回的responseHeader
    29.              RegisterBrokerResponseHeader responseHeader =
    30.                  (RegisterBrokerResponseHeader) response.decodeCommandCustomHeader(RegisterBrokerResponseHeader.class);
    31.              // 封装返回结果，更新masterAddr和haServerAddr
    32.              RegisterBrokerResult result = new RegisterBrokerResult();
    33.              result.setMasterAddr(responseHeader.getMasterAddr());
    34.              result.setHaServerAddr(responseHeader.getHaServerAddr());
    35.              if (response.getBody() != null) {
    36.                  result.setKvTable(KVTable.decode(response.getBody(), KVTable.class));
    37.              }
    38.              return result;
    39.          }
    40.          default:
    41.              break;
    42.      }
    44.      throw new MQBrokerException(response.getCode(), response.getRemark(), requestHeader == null ? null : requestHeader.getBrokerAddr());
    45.  }

    5.3.2. NameServer 处理心跳包（注册）

    在Broker 发送心跳包后，Nameserver负责接收，并处理心跳。Broker发出路由注册的心跳包之后，NameServer会根据心跳包中的requestCode进行处理。NameServer的默认网络处理器是org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor类，Broker向NameServer通过Netty发送心跳时，会发送requestCode=RequestCode._REGISTER_BROKER，接着会将心跳注册请求转发给_registerBroker()方法处理

    1. /**
    2.   * Netty 接收请求处理
    3.   * @param ctx
    4.   * @param request
    5.   * @return
    6.   * @throws RemotingCommandException
    7.   */
    8.  @Override
    9.  public RemotingCommand processRequest(ChannelHandlerContext ctx,
    10.      // ...省略代码
    11.      switch (request.getCode()) {
    12.          // ...省略代码
    13.          case RequestCode.REGISTER_BROKER:
    14.              Version brokerVersion = MQVersion.value2Version(request.getVersion());
    15.              if (brokerVersion.ordinal() >= MQVersion.Version.V3_0_11.ordinal()) {
    16.                  return this.registerBrokerWithFilterServer(ctx, request);
    17.              } else {
    18.                  // 向NameServer 注册Broker
    19.                  return this.registerBroker(ctx, request);
    20.              }
    21.          //... 省略代码
    22.      }
    23.      return null;
    24.  }

Nameserver 接收Broker心跳包最终由org.apache.rocketmq.namesrv.routeinfo.registerBroker()处理

  1. /**
  2.      * 向NameServer注册Broker 信息
  3.      * @param clusterName               Broker 集群名称
  4.      * @param brokerAddr                Broker 地址
  5.      * @param brokerName                Broker 名称
  6.      * @param brokerId                  BrokerId 大于0为从节点，0表示主节点
  7.      * @param haServerAddr              主节点地址，初次请求时为空，从节点向Nameserver注册后返回
  8.      * @param topicConfigWrapper        Topic 配置信息
  9.      * @param filterServerList          消息过滤服务器列表
  10.      * @param channel                   Netty 通道对象，知道是哪个Netty客户端
  11.      * @return
  12.      */
  13. public RegisterBrokerResult registerBroker(
  14.     final String clusterName,
  15.     final String brokerAddr,
  16.     final String brokerName,
  17.     final long brokerId,
  18.     final String haServerAddr,
  19.     final TopicConfigSerializeWrapper topicConfigWrapper,
  20.     final List<String> filterServerList,
  21.     final Channel channel) {
  22.     RegisterBrokerResult result = new RegisterBrokerResult();
  23.     try {
  24.         try {
  25.             // Step1：路由注册加写锁，防止并发修改RouteInfoManager中路由表
  26.             this.lock.writeLock().lockInterruptibly();
  28.             // 判断Broker 所属集群是否存在，不存在则创建集群，然后将Broker名加入集群Broker集合
  29.             Set<String> brokerNames = this.clusterAddrTable.get(clusterName);
  30.             if (null == brokerNames) {
  31.                 brokerNames = new HashSet<String>();
  32.                 // 将Broker 加入 clusterAddrTable 集群集合
  33.                 this.clusterAddrTable.put(clusterName, brokerNames);
  34.             }
  35.             brokerNames.add(brokerName);
  37.             // 是否第一次注册Broker信息，默认不是第一次
  38.             boolean registerFirst = false;
  40.             // Step2：维护BrokerData信息，尝试从brokerAddrTable 获取Broker信息
  41.             // 首先从brokerAddrTable中根据broker名称获取Broker信息，如果不存在就新建BrokerData并放入brokerAddrTable
  42.             BrokerData brokerData = this.brokerAddrTable.get(brokerName);
  43.             // Broker信息不存在
  44.             if (null == brokerData) {
  45.                 // 设置Broker信息已经被注册，true表示第一次注册
  46.                 registerFirst = true;
  47.                 // 创建BrokerData信息
  48.                 brokerData = new BrokerData(clusterName, brokerName, new HashMap<Long, String>());
  49.                 // BrokerData信息加入brokerAddrTable
  50.                 this.brokerAddrTable.put(brokerName, brokerData);
  51.             }
  52.             Map<Long, String> brokerAddrsMap = brokerData.getBrokerAddrs();
  53.             //Switch slave to master: first remove <1, IP:PORT> in namesrv, then add <0, IP:PORT>
  54.             //The same IP:PORT must only have one record in brokerAddrTable
  55.             Iterator<Entry<Long, String>> it = brokerAddrsMap.entrySet().iterator();
  56.             while (it.hasNext()) {
  57.                 Entry<Long, String> item = it.next();
  58.                 if (null != brokerAddr && brokerAddr.equals(item.getValue()) && brokerId != item.getKey()) {
  59.                     it.remove();
  60.                 }
  61.             }
  63.             String oldAddr = brokerData.getBrokerAddrs().put(brokerId, brokerAddr);
  64.             registerFirst = registerFirst || (null == oldAddr);
  65.             // Step3：如果Broker 为主节点，并且Broker的topic配置信息发生变化或者是初次注册
  66.             // 需要创建或者更新topic路由元数据，填充topicQueueTabe，如果是默认主题自动注册路由信息，其中包含
  67.             // MixAll.DEFAULTOPIC的路由信息。当消息生产者发送主题时，如果该主题没有创建，并且BrokerConfig的autoCreateTopicEnable=true
  68.             // 那么就返回MixAll_DEFAULT_TOPIC的路由信息
  69.             if (null != topicConfigWrapper
  70.                 && MixAll.MASTER_ID == brokerId) {
  71.                 // Broker的topic配置信息发生变化或者初次注册Broker
  72.                 // 创建或者更新topic路由元信息
  73.                 // 校验版本，决定是否需要创建或者更新topic路由元信息
  74.                 if (this.isBrokerTopicConfigChanged(brokerAddr, topicConfigWrapper.getDataVersion())
  75.                     || registerFirst) {
  76.                     ConcurrentMap<String, TopicConfig> tcTable =
  77.                         topicConfigWrapper.getTopicConfigTable();
  78.                     if (tcTable != null) {
  79.                         // 创建或者更新topic 路由元信息
  80.                         for (Map.Entry<String, TopicConfig> entry : tcTable.entrySet()) {
  81.                             // 根据topicConfig创建QueueData数据结构，然后更新topicQueueTable
  82.                             this.createAndUpdateQueueData(brokerName, entry.getValue());
  83.                         }
  84.                     }
  85.                 }
  86.             }
  88.             // Step4：更新BrokerLiveInfo，存储状态正常的Broker信息表，BrokerLiveInfo是执行路由删除操作的重要依据
  89.             BrokerLiveInfo prevBrokerLiveInfo = this.brokerLiveTable.put(brokerAddr,
  90.                                                                          new BrokerLiveInfo(
  91.                                                                              System.currentTimeMillis(),
  92.                                                                              topicConfigWrapper.getDataVersion(),
  93.                                                                              channel,
  94.                                                                              haServerAddr));
  95.             if (null == prevBrokerLiveInfo) {
  96.                 log.info("new broker registered, {} HAServer: {}", brokerAddr, haServerAddr);
  97.             }
  99.             // Step5：注册Broker的过滤器Server地址列表，一个Broker上会关联多个FilterServer消息过滤服务器
  100.             // 如果Broker是从节点，需要查找该Broker的主节点信息，并更新对应的masterAddr属性
  101.             if (filterServerList != null) {
  102.                 if (filterServerList.isEmpty()) {
  103.                     this.filterServerTable.remove(brokerAddr);
  104.                 } else {
  105.                     this.filterServerTable.put(brokerAddr, filterServerList);
  106.                 }
  107.             }
  108.             // Broker 是从节点
  109.             if (MixAll.MASTER_ID != brokerId) {
  110.                 // Broker主节点地址
  111.                 String masterAddr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
  112.                 if (masterAddr != null) {
  113.                     BrokerLiveInfo brokerLiveInfo = this.brokerLiveTable.get(masterAddr);
  114.                     // 更新masterAddr
  115.                     if (brokerLiveInfo != null) {
  116.                         result.setHaServerAddr(brokerLiveInfo.getHaServerAddr());
  117.                         result.setMasterAddr(masterAddr);
  118.                     }
  119.                 }
  120.             }
  121.         } finally {
  122.             this.lock.writeLock().unlock();
  123.         }
  124.     } catch (Exception e) {
  125.         log.error("registerBroker Exception", e);
  126.     }
  128.     return result;
  129. }