2、RocketMQ 路由中心 NameServer

发表于 | 分类于 ，

NameServer 是 整个 RocketMQ 的 ”大脑“；

路由管理、服务注册、服务发现 机制；

本章重点内容：

  • NameServer 整体架构设计；
  • NameServer 动态路由 发现与剔除机制；

1、NameServer 架构设计

RocketMQ 逻辑部署图：

  • Broker 启动时，向 所有NameServer 注册；

  • Producer在 发送消息 之前，先从 NameServer 获取 Broker 服务器地址列表，然后 根据负载均衡 算法 从 列表中选择一台服务器进行消息发送；

  • NameServer 与 每台 Broker 保持 长链接，并且间隔 30s 检测 Broker是否存活。如果Broker宕机，则从 路由注册表中将其移除；

  • NameServer 本身的 高可用 通过 部署多台NameServer 服务器来实现；但彼此之间不通信；


2、NameServer 启动流程

NamaServer 启动类：org.apache.rocketmq.namesrv.NamesrvStartup

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

2.1、初始化 NamesrvController

NamesrvController controller = createNamesrvController(args);

创建：

org.apache.rocketmq.namesrv.NamesrvStartup.createNamesrvController

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
public static NamesrvController createNamesrvController(String[] args) throws IOException, JoranException {
    System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
    //PackageConflictDetect.detectFastjson();
    // 将 args 转换成 CommandLine 对象
    Options options = ServerUtil.buildCommandlineOptions(new Options());
    commandLine = ServerUtil.parseCmdLine("mqnamesrv", args, buildCommandlineOptions(options), new PosixParser());
    if (null == commandLine) {
        System.exit(-1);
        return null;
    }
    // 读取 CommandLine 中 -c namesrv.conf 中的配置, 加载 到 NamesrvConfig 和 NettyServerConfig 中
    final NamesrvConfig namesrvConfig = new NamesrvConfig();
    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();
        }
    }
    // 如果有 -p 参数，则 向console打印 namesrvConfig 和 nettyServerConfig 所有属性，退出
    if (commandLine.hasOption('p')) {
        InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_CONSOLE_NAME);
        MixAll.printObjectProperties(console, namesrvConfig);
        MixAll.printObjectProperties(console, nettyServerConfig);
        System.exit(0);
    }
    // 将 commandLine 中的其他参数 加载到 namesrvConfig 中
    MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);
    // 如果没有 设置 rocketmqHome，则 退出
    if (null == namesrvConfig.getRocketmqHome()) {
        System.out.printf("Please set the %s variable in your environment to match the location of the RocketMQ installation%n", MixAll.ROCKETMQ_HOME_ENV);
        System.exit(-2);
    }
    // 从 logback_namesrv.xml 中 加载 日志配置信息
    LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
    JoranConfigurator configurator = new JoranConfigurator();
    configurator.setContext(lc);
    lc.reset();
    configurator.doConfigure(namesrvConfig.getRocketmqHome() + "/conf/logback_namesrv.xml");

    log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
    // 向 日志文件.log 打印 namesrvConfig 和 nettyServerConfig
    MixAll.printObjectProperties(log, namesrvConfig);
    MixAll.printObjectProperties(log, nettyServerConfig);
    // 使用 namesrvConfig 和 nettyServerConfig 初始化 NamesrvController
    // 同时初始化 Configuration
    final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);

    // remember all configs to prevent discard
    // 将 properties 全部记录到 Configuration 中
    controller.getConfiguration().registerConfig(properties);

    return controller;
}

代码功能：

  1. 需要先加载 命令行中-c configFile 指定的配置文件 和 命令行中的其他参数 初始化 NamesrvConfig 和 NettyServerConfig；
  2. 使用 NamesrvConfig 和 NettyServerConfig 初始化 NamesrvController；
  3. 然后将 所有参数 持久化在 configFile 中；

org.apache.rocketmq.namesrv.NamesrvController

1
2
3
4
5
6
7
8
9
10
11
12
13
public NamesrvController(
    NamesrvConfig namesrvConfig, NettyServerConfig nettyServerConfig) {
    this.namesrvConfig = namesrvConfig;
    this.nettyServerConfig = nettyServerConfig;
    this.kvConfigManager = new KVConfigManager(this);
    this.routeInfoManager = new RouteInfoManager();
    this.brokerHousekeepingService = new BrokerHousekeepingService(this);
    this.configuration = new Configuration(
        log,
        this.namesrvConfig, this.nettyServerConfig
    );
    this.configuration.setStorePathFromConfig(this.namesrvConfig, "configStorePath");
}

初始化：

  • kvConfigManager：使用 configTable（HashMap）存储 所有配置；
  • routeInfoManager：存储路由信息；
  • brokerHousekeepingService：

2.2、启动 NamesrvController

start(controller);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
org.apache.rocketmq.namesrv.NamesrvController;

public static NamesrvController start(final NamesrvController controller) throws Exception {
    if (null == controller) {
        throw new IllegalArgumentException("NamesrvController is null");
    }
    // 初始化
    boolean initResult = controller.initialize();
    if (!initResult) {
        controller.shutdown();
        System.exit(-3);
    }
    // 注册钩子，JVM 关闭时，关闭controller
    Runtime.getRuntime().addShutdownHook(
        new ShutdownHookThread(log, new Callable<Void>() {
        @Override
        public Void call() throws Exception {
            controller.shutdown();
            return null;
        }
    }));
    // 启动 controller
    controller.start();
    return controller;
}

初始化 controller

boolean initResult = controller.initialize();

  1. 从 configFile 中 加在配置 放在 kvConfigManager 的 configTable（HashMap） 中；
  2. 初始化 NettyRemotingServer；
  3. 创建 业务线程池 remotingExecutor；
  4. 把 remotingExecutor 作为 remotingServer 的 默认处理线程池；
  5. 每 10s 扫描一次不活跃的 Broker，从 brokerLiveTable 中移除；
  6. 每 10s 打印一次 KV配置；
  7. 创建一个文件监控服务，当 sslContext 变更时，重新加载 sslContext；
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
org.apache.rocketmq.namesrv.NamesrvController;

public boolean initialize() {
    // 从 configFile 中 加在配置 放在 kvConfigManager 的 configTable（HashMap） 中
    this.kvConfigManager.load();
    // 初始化 NettyRemotingServer
    this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
    // 创建 业务线程池
    this.remotingExecutor =
        Executors.newFixedThreadPool(
        nettyServerConfig.getServerWorkerThreads(), 
        new ThreadFactoryImpl("RemotingExecutorThread_"));
    // 把 remotingExecutor 作为 remotingServer 的 默认处理线程池
    this.registerProcessor();
    // 每 10s 扫描一次不活跃的 Broker，从 brokerLiveTable 中移除
    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
            NamesrvController.this.routeInfoManager.scanNotActiveBroker();
        }
    }, 5, 10, TimeUnit.SECONDS);
    // 每 10s 打印一次 KV配置
    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

        @Override
        public void run() {
            NamesrvController.this.kvConfigManager.printAllPeriodically();
        }
    }, 1, 10, TimeUnit.MINUTES);
    // 创建一个文件监控服务，当 sslContext 变更时，重新加载 sslContext
    if (TlsSystemConfig.tlsMode != TlsMode.DISABLED) {
        // Register a listener to reload SslContext
        try {
            fileWatchService = new FileWatchService(
                new String[] {
                    TlsSystemConfig.tlsServerCertPath,
                    TlsSystemConfig.tlsServerKeyPath,
                    TlsSystemConfig.tlsServerTrustCertPath
                },
                new FileWatchService.Listener() {
                    boolean certChanged, keyChanged = false;
                    @Override
                    public void onChanged(String path) {
                        if (path.equals(TlsSystemConfig.tlsServerTrustCertPath)) {
                            log.info("The trust certificate changed, reload the ssl context");
                            reloadServerSslContext();
                        }
                        if (path.equals(TlsSystemConfig.tlsServerCertPath)) {
                            certChanged = true;
                        }
                        if (path.equals(TlsSystemConfig.tlsServerKeyPath)) {
                            keyChanged = true;
                        }
                        if (certChanged && keyChanged) {
                            log.info("The certificate and private key changed, reload the ssl context");
                            certChanged = keyChanged = false;
                            reloadServerSslContext();
                        }
                    }
                    private void reloadServerSslContext() {
                        ((NettyRemotingServer) remotingServer).loadSslContext();
                    }
                });
        } catch (Exception e) {
            log.warn("FileWatchService created error, can't load the certificate dynamically");
        }
    }

    return true;
}

controller 的 关闭和启动

controller.shutdown();
controller.start();

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public void start() throws Exception {
    this.remotingServer.start(); // 启动 Netty服务器

    if (this.fileWatchService != null) {
        this.fileWatchService.start();  // 启动 sslContext 文件监控服务
    }
}

public void shutdown() {
    this.remotingServer.shutdown();  // 关闭 Netty服务器
    this.remotingExecutor.shutdown();  // 关闭 线程池
    this.scheduledExecutorService.shutdown();  // 关闭 sslContext 文件监控服务

    if (this.fileWatchService != null) {
        this.fileWatchService.shutdown();
    }
}

2.3、NameServer 路由注册、故障剔除

NameServer的主要作用： 为Producer 和 Consumer 提供 Topic的 路由信息；

NameServer功能：存储路由信息 、管理Broker节点；

1）路由元信息

1
2
3
4
5
HashMap<String/* topic */, List<QueueData>> topicQueueTable;
HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;
  • topicQueueTable：Topic消息队列 路由信息，Producer 根据 路由表 进行负载均衡；
  • brokerAddrTable：Broker基础信息，包含 brokerName、所属集群名称、主备 broker地址；
  • clusterAddrTable：Broker集群信息，存储集群中 所有 Broker 名称；
  • brokerLiveTable：Broker状态信息。NameServer每次收到心跳包时会替换该信息；
  • filterServerTable：Broker上的 FilterServer 列表，用于类模式消息过滤；
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
package org.apache.rocketmq.common.protocol.route;
public class QueueData implements Comparable<QueueData> {
    private String brokerName;
    private int readQueueNums;
    private int writeQueueNums;
    private int perm; // 读写权限，具体含义参考 PermName
    private int topicSynFlag;  // topic同步标记，具体含义参考 TopicSynFlag
}

package org.apache.rocketmq.common.protocol.route;
public class BrokerData implements Comparable<BrokerData> {
    private String cluster;
    private String brokerName;
    // brokerId = 0, 表示Master, brokerId > 0 表示Slave
    private HashMap<Long/* brokerId */, String/* broker address */> brokerAddrs;
	
    public String selectBrokerAddr() {
        String addr = this.brokerAddrs.get(MixAll.MASTER_ID);
        // 先取 Master，取不到Master则随机返回一个Slave
        if (addr == null) {
            List<String> addrs = new ArrayList<String>(brokerAddrs.values());
            return addrs.get(random.nextInt(addrs.size()));
        }
        return addr;
    }
}

package org.apache.rocketmq.namesrv.routeinfo;
class BrokerLiveInfo {
    private long lastUpdateTimestamp;
    private DataVersion dataVersion;
    private Channel channel;
    private String haServerAddr;
}

2）路由注册

机制：通过Broker 与 NameServer 心跳功能实现的。

  1. Broker 启动时 向集群中 所有NameServer 发送心跳语句;
  2. Broker 每隔 30s 向集群中 所有NameServer 发送心跳语句;
  3. NameServer 收到 Broker 心跳包时，会更新 brokerLiveTable 缓存中 BrokerLiveInfo 的 lastUpdateTimestamp；
  4. NameServer 每隔 10s 扫描 brokerLiveTable，如果连续 120s 没有收到 心跳包，则 NameServer将移除 该Broker 的路由信息，同时关闭 socket 连接；

①、Broker发送心跳包核心代码：

org.apache.rocketmq.broker.BrokerController.start()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
public void start() throws Exception {
    ......
    // Broker 启动时 向集群中 所有NameServer 发送心跳语句;
    this.registerBrokerAll(true, false, true);
    // Broker 每隔 30s 向集群中 所有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);
    ......
}

遍历 NameServer 列表，Broker 消息服务器 依次向 NameServer 发送心跳包：

org.apache.rocketmq.broker.out.BrokerOuterAPI

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
/**
 * 向所有 NameServer 注册 Broker
 * @param clusterName String 集群名称
 * @param brokerAddr String broker地址
 * @param brokerName String broker名称
 * @param brokerId long brokerId，0-Master，大于0-Salve
 * @param haServerAddr String Master地址，初次请求时，该值为空，salve 向 NameServer注册后返回
 * @param topicConfigWrapper TopicConfigSerializeWrapper 主题（topic）配置，存储在${}rocket_home}/store/config/topic.json中
 * @param filterServerList List<String> 消息过滤服务器列表
 * @param oneway boolean 是否 单向请求，不关心返回结果
 * @param timeoutMills int
 * @param compressed boolean
 * @return
 */
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) {
        // 构造 requestHeader
        final RegisterBrokerRequestHeader requestHeader = 
            new RegisterBrokerRequestHeader();
        requestHeader.setBrokerAddr(brokerAddr);
        requestHeader.setBrokerId(brokerId);
        requestHeader.setBrokerName(brokerName);
        requestHeader.setClusterName(clusterName);
        requestHeader.setHaServerAddr(haServerAddr);
        requestHeader.setCompressed(compressed);
		// 构造 requestBody
        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);  // 使用 crc32 作为 校验码
        final CountDownLatch countDownLatch = 
            new CountDownLatch(nameServerAddressList.size());
        // 遍历 NameServer 列表，Broker 消息服务器 依次向 NameServer 发送心跳包
        for (final String namesrvAddr : nameServerAddressList) {
            brokerOuterExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        // 向单个 NameServer 注册, 
                        // 从 responseHeader 里 拿到Master地址 和 HaServer地址
                        // 从 response.getBody() 中解析，放入 KVTable
                        RegisterBrokerResult result = registerBroker(
                            namesrvAddr,oneway, timeoutMills,requestHeader,body);
                        if (result != null) {
                            registerBrokerResultList.add(result);
                        }

                        log.info("register broker to name server {} OK", namesrvAddr);
                    } catch (Exception e) {
                        log.warn("registerBroker Exception, {}", namesrvAddr, e);
                    } finally {
                        countDownLatch.countDown();
                    }
                }
            });
        }
		// 使用 countDownLatch 等待 固定时间timeoutMills
        try {
            countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
        }
    }
    // 返回 注册结果
    return registerBrokerResultList;
}

②、NameServer处理心跳包

处理机制：org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor.processRequest() 网络处理 解析请求类型，如果请求类型为 RequestCode.REGISTER_BROKER，则请求最终转发到 RouteInfoManager.registerBroker()；

org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
public RemotingCommand processRequest(ChannelHandlerContext ctx,
    RemotingCommand request) throws RemotingCommandException {
    switch (request.getCode()) {
        .......
        case RequestCode.REGISTER_BROKER:
            // 如果 请求类型 为 Broker 心跳
            Version brokerVersion = MQVersion.value2Version(request.getVersion());
            if (brokerVersion.ordinal() >= MQVersion.Version.V3_0_11.ordinal()) {
                return this.registerBrokerWithFilterServer(ctx, request);
            } else {
                return this.registerBroker(ctx, request);
            }
        ......
        default:
            break;
    }
    return null;
}

org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager.registerBroker()

  1. 路由注册需要加写锁（ReentrantReadWriteLock.writeLock()）, 防止 并发 修改 RouteInfoManager 的路由表；首先判断 Broker 集群 是否存在，如果不存在，则 创建。然后将 broker名 加入到对应集群中；
  2. 维护 BrokerData信息；
  3. 如果Broker是Master，并且 broker中的topic 配置信息发生变化 或 首次注册，则创建或更新 Topic的路由元数据，填充 topicQueueTable；
  4. 更新 BrokerLiveInfo，存活的 Broker信息表；
  5. 注册Broker的过滤器 Server地址列表，一个Broker 上会关联 多个FilterSever消息过滤器；
  6. 如果 brokerName 不是Master，则 获取对应的 masterAddr 和 masterAddr 对应的 haServerAddr 作为返回值；
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
/**
 * 处理 broker来的 心跳包，更新NameServer中的元数据信息；
 * 更新 clusterAddrTable，brokerAddrTable，topicConfigWrapper
 * @param clusterName String
 * @param brokerAddr String
 * @param brokerName String
 * @param brokerId long
 * @param haServerAddr String
 * @param topicConfigWrapper TopicConfigSerializeWrapper
 * @param filterServerList List<String>
 * @param channel Channel
 * @return
 */
public RegisterBrokerResult registerBroker(
    final String clusterName,
    final String brokerAddr,
    final String brokerName,
    final long brokerId,
    final String haServerAddr,
    final TopicConfigSerializeWrapper topicConfigWrapper,
    final List<String> filterServerList,
    final Channel channel) {
    RegisterBrokerResult result = new RegisterBrokerResult();
    try {
        try {
            this.lock.writeLock().lockInterruptibly();
            // 将 brokerName 放入所属的 集群地址表clusterAddrTable（HashMap<String, Set<String>>）
            Set<String> brokerNames = this.clusterAddrTable.get(clusterName);
            if (null == brokerNames) {
                brokerNames = new HashSet<String>();
                this.clusterAddrTable.put(clusterName, brokerNames);
            }
            brokerNames.add(brokerName);
            // 判断是否第一次注册，写入broker地址表 brokerAddrTable（HashMap<String, 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);
            }
            String oldAddr = brokerData.getBrokerAddrs().put(brokerId, brokerAddr);
            registerFirst = registerFirst || (null == oldAddr);

            // 如果 brokerName 是 Master节点，并且topicConfig有变更，则 更新 topicConfigWrapper 中的元数据信息
            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()) {
                            // 添加或更新 brokerName 的 topic 配置
                            this.createAndUpdateQueueData(brokerName, entry.getValue());
                        }
                    }
                }
            }
            // 更新 brokerLiveTable 中的 对应 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);
            }
            // 更新 brokerAddr 对应的 过滤服务器列表
            if (filterServerList != null) {
                if (filterServerList.isEmpty()) {
                    this.filterServerTable.remove(brokerAddr);
                } else {
                    this.filterServerTable.put(brokerAddr, filterServerList);
                }
            }
            // 如果 brokerName 不是Master，则 获取对应的 masterAddr 和 masterAddr 对应的 haServerAddr
            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);
                    }
                }
            }
        } finally {
            this.lock.writeLock().unlock();
        }
    } catch (Exception e) {
        log.error("registerBroker Exception", e);
    }

    return result;
}

3） 路由删除

心跳包：{BrokerId，Broker地址，Broker名称，Broker所属集群名称}；

剔除失效Broker机制：NameServer 每隔 10s 扫描 brokerLiveTable 状态表，如果 BrokerLive的 lastUpdateTimestamp 的时间戳 距 当前时间超过 120s，则 认为 Broker失效；移除 该 Broker，关闭与Broker连接，并同时更新 topicQueueTable、brokerAddrTable、brokerLiveTable、filterServerTable；

RocketMQ 有 两个触发点来 触发 路由删除：

  1. Namaserver 定时扫描brokerLiveTable，检测上次心跳包与当前系统时间差，如果时间间隔 > 120s，则移除该Broker信息；
  2. Broker 在正常被关闭的情况下，会执行 unregisterBroker 指令；

两种方法的触发方式的公共代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
org.apache.rocketmq.namesrv.NamesrvController;

public boolean initialize() {
    ......
    // 每 10s 扫描一次不活跃的 Broker，从 brokerLiveTable 中移除
    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
            NamesrvController.this.routeInfoManager.scanNotActiveBroker();
        }
    }, 5, 10, TimeUnit.SECONDS);
    ......
}
1
org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager;

4）路由发现