一文详解RocketMQ的NameServer路由管理

冷不防 2024-03-22 08:56 23阅读 0赞

![format_png][]

## 一、NameServer的路由元信息 ##

NameServer的主要作用是为消息生产者和消息消费者提供Topic的路由信息。所以NameServer需要保存和管理路由的基础信息。

NameServer的路由管理的实现类是：`org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager`：

public class RouteInfoManager {
        private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
        // 120秒，broker 上一次心跳时间超过这个数便会被剔除
        private final static long BROKER_CHANNEL_EXPIRED_TIME = 1000 * 60 * 2;
        private final ReadWriteLock lock = new ReentrantReadWriteLock();
        private final HashMap<String/* topic */, Map<String /* brokerName */ , QueueData>> topicQueueTable;
        private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
        private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
        private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
        private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;
    
        public RouteInfoManager() {
            this.topicQueueTable = new HashMap<>(1024);
            this.brokerAddrTable = new HashMap<>(128);
            this.clusterAddrTable = new HashMap<>(32);
            this.brokerLiveTable = new HashMap<>(256);
            this.filterServerTable = new HashMap<>(256);
        }
        //...省略
    }
    复制代码

`org.apache.rocketmq.common.protocol.route.QueueData`：

public class QueueData implements Comparable<QueueData> {
        private String brokerName;
        private int readQueueNums;
        private int writeQueueNums;
        private int perm;
        private int topicSysFlag;
        // ... setter getter
    }
    复制代码

`org.apache.rocketmq.common.protocol.route.BrokerData`：

public class BrokerData implements Comparable<BrokerData> {
        private String cluster;
        private String brokerName;
        private HashMap<Long/* brokerId */, String/* broker address */> brokerAddrs;
    
        private final Random random = new Random();
        // ... setter getter
    }
    复制代码

`org.apache.rocketmq.namesrv.routeinfo.BrokerLiveInfo`：

class BrokerLiveInfo {
        private long lastUpdateTimestamp;
        private DataVersion dataVersion;
        private Channel channel;
        private String haServerAddr;
    
        public BrokerLiveInfo(long lastUpdateTimestamp, DataVersion dataVersion, Channel channel,
                              String haServerAddr) {
            this.lastUpdateTimestamp = lastUpdateTimestamp;
            this.dataVersion = dataVersion;
            this.channel = channel;
            this.haServerAddr = haServerAddr;
        }
        // ... setter getter
    }
    复制代码

*  topicQueueTable：topic消息队列的路由信息，消息发送的时候会根据路由表进行负载均衡。Key为topic名称，value也是一个Map：以brokerName为key，value是队列数据如上代码所示，包含读/写队列数量、权重等。
 *  brokerAddrTable：broker的基础信息，Key为brokerName，value包含brokerName，broker所在的集群信息，主备broker的地址。
 *  clusterAddrTable：broker集群信息，Key为集群名称(clusterName)，value存储的是集群中所有broker的名称(brokerName)。
 *  brokerLiveTable：Broker状态信息，NameServer每次收到心跳包时会替换该信息。这也是NameServer每10秒要扫描的信息。
 *  filterServerTable：Broker上的FilterServer列表，用于类模式消息过滤。**类模式过滤机制在4.4及以后版本被废弃**。

类图如下：

![format_png 1][]

### 1.1 RocketMQ运行时的路由元信息 ###

RocketMQ的一个Topic是可以有多个消息队列，一个Broker默认会为每一个Topic创建4个读队列和4个写队列。多个Broker组成一个集群，多个BrokerName一样的Broker组成主从架构。brokerId大于0表示从节点，brokerId等于0表示是主节点。假如配置如下的broker集群，集群名c1：

在启动Broker的时候，指定配置文件，修改broker配置文件的：brokerClusterName、brokerName和brokerId。

#### 1.1.1 本地运行IDEA Debug查看运行时的路由元信息 ####

首先创建创建四个Broker的配置文件主要修改：brokerClusterName、brokerName和brokerId，**注意store目录也需要修改，每个broker使用不同的store目录，或者是使用不同的rocketmq目录。listenPort也需要指定不同的端口，因为在本地调试使用的同一台电脑，而且listenProt的值不能相隔太近不然会报错：`Address already in use: bind`**

brokerClusterName = c1
    brokerName = broker-a
    brokerId = 0
    namesrvAddr=127.0.0.1:9876
    deleteWhen = 04
    fileReservedTime = 48
    brokerRole = ASYNC_MASTER
    flushDiskType = ASYNC_FLUSH
    listenPort=10911
    # 存储路径E:\java\source\rocketmq\ROCKETMQ
    storePathRootDir=E:\\java\\source\\rocketmq\\ROCKETMQ\\store-a0
    # CommitLog存储路径
    storePathCommitLog=E:\\java\\source\\rocketmq\\ROCKETMQ\\store-a0\\commitlog
    # 消费队列存储路径
    storePathConsumeQueue=E:\\java\\source\\rocketmq\\ROCKETMQ\\store-a0\\consumequeue
    # 消息索引存储路径
    storePathIndex=E:\\java\\source\\rocketmq\\ROCKETMQ\\store-a0\\index
    # checkpoint文件存储路径
    storeCheckpoint=E:\\java\\source\\rocketmq\\ROCKETMQ\\store-a0\\checkpoint
    # abort文件存储路径
    abortFile=E:\\java\\source\\rocketmq\\ROCKETMQ\\store-a0\\abort
    复制代码

IDEA运行四个Broker实例并指定配置文件，如下图所示：

![format_png 2][]

![format_png 3][]

![format_png 4][]

![format_png 5][]

配置完成之后，先以Debug模式运行NamesrvStartup(启动NameServer)，再启动四个Broker。都启动完成之后，在`org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#scanNotActiveBroker`方法里面打个断点。该方法每隔10秒扫描一次brokerLiveTable，移除处于未激活状态的Broker，路由元信息如下：

![format_png 6][]

## 二、路由注册流程分析 ##

Broker通过心跳机制向NameServer发送心跳包，每个隔30秒就会向NameServer集群发送心跳包，NameServer收到心跳包之后会先更新brokerLiveTable的lastUpdateTimestamp。NameServer每隔10秒就会扫描brokerLiveTable中各个Broker上报来的lastUpdateTimestamp，如果连续超过120秒没收到Broker的心跳包，NameServer会把该Broker的路由信息移除。

### 2.1 Broker发送心跳包 ###

我们在本地是通过运行`org.apache.rocketmq.broker.BrokerStartup#start`方法启动，从这里开始看，进到`BrokerController`类的start方法：

public void start() throws Exception {
        // ...省略一些代码
        
        // 定时任务每个30秒向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);
                }
            } // brokerConfig.getRegisterNameServerPeriod() ---> 30 * 1000
        }, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), TimeUnit.MILLISECONDS);
        if (this.brokerStatsManager != null) {
            this.brokerStatsManager.start();
        }
        if (this.brokerFastFailure != null) {
            this.brokerFastFailure.start();
        }
    
    }
    复制代码

点击`org.apache.rocketmq.broker.BrokerController#registerBrokerAll`方法查看：

public synchronized void registerBrokerAll(final boolean checkOrderConfig, boolean oneway, boolean forceRegister) {
        // topicConfig的包装
        TopicConfigSerializeWrapper topicConfigWrapper = this.getTopicConfigManager().buildTopicConfigSerializeWrapper();

[format_png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/3864c163ff4044908424c414edac0e1c.png
[format_png 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/2a9837a5f7f245349f2e02dd75eed445.png
[format_png 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/5ee47e3126ec435ea042f7897e8b9720.png
[format_png 3]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/c5bd726f34a04ababc6d84de23f6b2b0.png
[format_png 4]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/66a698d1f4194fdcbc84425290a7007d.png
[format_png 5]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/3693c3b129cc4ce9ad4514e223a9192d.png
[format_png 6]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/32ca1111e11e4aafb0fb686f709377a6.png