RocketMQ路由中心【NameServer】

ゝ一纸荒年。 2021-12-11 04:53 235阅读 0赞

**NameServer主要提供路由管理、服务注册、及服务发现的机制**。

一、**NameServer架构设计**

消息中间件的设计思路一般基于主题的订阅发布机制。消息生产者发送某一主题的消息到消息服务器，消息服务器负责该消息的持久化存储，消息消费者订阅感兴趣的主题，消息服务器根据订阅信息【路由信息】将消息推送到消费者【PUSH模式】或者消息消费者主动向消息服务器拉取消息【PULL消息】从而实现消息生产者与消费者的解耦。

此时会出现以下问题：

1.  如何避免消息服务器的单点故障？
2.  消息生产者如何如何知道消息要发往哪台消息服务器？
3.  如果某一台消息服务器宕机了，生产者如何在不重启服务器的情况下感知呢？

对于问题 1  我们通常会通过部署多台消息服务器来共同承担消息服务器的存储。NameServer本身的高可用可通过部署多台NameServer服务器来实现，但彼此之间互不通信，也就是某一刻NameServer服务器之间的数据并不完全相同，但这对消息发送不会造成任何影响。

对于问题 2  Broker消息服务器在启动时向所有NameServer注册，消息生产者在发送消息之前先从NameServer获取Broker服务器地址列表，然后根据负载算法从列表中选择一台消息服务器进行消息发送。

对于问题 3 NameServer与每台Broker服务器保持长连接，并间隔10s检测Broker是否存活，如果检测到Broker宕机，则从路由注册表中将其移除。但是路由变化不会马上通知消息生产者【为了降低NameServer实现复杂度，在消息发送端提供容错机制保证消息发送高可用】

二、**NameServer启动流程**

NamesrvStartup中的main方法调用如下main0方法：

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;
        }

我们可以发现是需要创建NameServerController ，而创建NameServerController需要初始化

final NamesrvConfig namesrvConfig = new NamesrvConfig();
    final NettyServerConfig nettyServerConfig = new NettyServerConfig();

创建NameServerController后需要调用ininialize方法

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);
            }
    
            Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<Void>() {
                @Override
                public Void call() throws Exception {
                    controller.shutdown();
                    return null;
                }
            }));
    
            controller.start();
    
            return controller;
        }

然后注册JVM钩子函数启动服务器，以便监听Broker，消息生产者的网络请求。

从这里我们学到了如果我们使用了线程池，一种优雅停机的方式就是注册一个JVM钩子函数，在JVM进程关闭之前，先将线程池关闭，及时释放资源。

三、**NameServer路由注册、故障剔除**

NameServer主要作用是为消息生产者消费者提供关于主题Topic的路由信息，所以NameServer需要存储路由的基本信息，还要能够管理Broker节点，包括路由注册、路由删除功能。

1、路由元信息：

private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
    private final static long BROKER_CHANNEL_EXPIRED_TIME = 1000 * 60 * 2;//两分钟
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    //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 状态信息NameServer每次收到心跳包时会替换 该信息
    private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
    //Broker中的FilterServer列表，用于类模式消息过滤，
    private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1lhbmd6YWlMZUhlSGU_size_16_color_FFFFFF_t_70][]

QueueData源码：

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;
        private int topicSynFlag;
        //省略getter和setter
        @Override
        public int hashCode() {
            final int prime = 31;
            int result = 1;
            result = prime * result + ((brokerName == null) ? 0 : brokerName.hashCode());
            result = prime * result + perm;
            result = prime * result + readQueueNums;
            result = prime * result + writeQueueNums;
            result = prime * result + topicSynFlag;
            return result;
        }
    
        @Override
        public boolean equals(Object obj) {
            if (this == obj)
                return true;
            if (obj == null)
                return false;
            if (getClass() != obj.getClass())
                return false;
            QueueData other = (QueueData) obj;
            if (brokerName == null) {
                if (other.brokerName != null)
                    return false;
            } else if (!brokerName.equals(other.brokerName))
                return false;
            if (perm != other.perm)
                return false;
            if (readQueueNums != other.readQueueNums)
                return false;
            if (writeQueueNums != other.writeQueueNums)
                return false;
            if (topicSynFlag != other.topicSynFlag)
                return false;
            return true;
        }
    
        @Override
        public String toString() {
          
        }
    
        
    }

BrokerData源码：

package org.apache.rocketmq.common.protocol.route;
    
    import java.util.ArrayList;
    import java.util.HashMap;
    import java.util.List;
    import java.util.Random;
    import org.apache.rocketmq.common.MixAll;
    
    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();
        public BrokerData() {
        }
        public BrokerData(String cluster, String brokerName, HashMap<Long, String> brokerAddrs) {
            this.cluster = cluster;
            this.brokerName = brokerName;
            this.brokerAddrs = brokerAddrs;
        }
    
        public String selectBrokerAddr() {
            String addr = this.brokerAddrs.get(MixAll.MASTER_ID);
    
            if (addr == null) {
                List<String> addrs = new ArrayList<String>(brokerAddrs.values());
                return addrs.get(random.nextInt(addrs.size()));
            }
    
            return addr;
        }
        @Override
        public int hashCode() {
            final int prime = 31;
            int result = 1;
            result = prime * result + ((brokerAddrs == null) ? 0 : brokerAddrs.hashCode());
            result = prime * result + ((brokerName == null) ? 0 : brokerName.hashCode());
            return result;
        }
    
        @Override
        public boolean equals(Object obj) {
            if (this == obj)
                return true;
            if (obj == null)
                return false;
            if (getClass() != obj.getClass())
                return false;
            BrokerData other = (BrokerData) obj;
            if (brokerAddrs == null) {
                if (other.brokerAddrs != null)
                    return false;
            } else if (!brokerAddrs.equals(other.brokerAddrs))
                return false;
            if (brokerName == null) {
                if (other.brokerName != null)
                    return false;
            } else if (!brokerName.equals(other.brokerName))
                return false;
            return true;
        }
    
        @Override
        public String toString() {
        
        }
    
    }

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;
        }
        @Override
        public String toString() {
            return "BrokerLiveInfo [lastUpdateTimestamp=" + lastUpdateTimestamp + ", dataVersion=" + dataVersion
                + ", channel=" + channel + ", haServerAddr=" + haServerAddr + "]";
        }
    }

2、路由注册：

RocketMQ路由注册是通过Broker与NameServer的心跳功能实现的，Broker启动时向集群中所有的NameServer发送心跳语句，每隔30秒向集群中所有NameServer发送心跳包，NameServer收到Broker心跳包时会更新 brokerLiveTable缓存中的BrokerLiveInfo的lastUpdateTimestamp，然后NameServer每隔10秒扫描 brokerLiveTable ，如果连续120秒 【即前文中的两分钟】没有收到心跳包，NameServer将移除该Broker的路由信息同时关闭Socket连接。

brokerId   0 是Master ；大于0 是Slave。

A、Broker发送心跳包：

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

brokerAddr ： broker地址；brokerId ： brokerId，0：Master大于0：Slave。

brokerName： broker名称；clusterName：集群名称。hasServerAddr ： master地址，初次请求时值为空，slave向NameServer注册后返回，

requestBody：filterServerList  消息过滤服务器列表， topicConfigWrapper： 主题配置。

B、NameServer处理心跳包：

DefaultRequestProcessor处理请求类型为： RequestCode.REGISTER\_BROKER。

路由注册需要加写锁，防止并发修改RouteInfoMapper中的路由表，首先判断Broker所属集群是否存在，如果不存在则创建，然后将Broker名加入到集群Broker集合中；

维护BrokerData信息首先从brokerAddrTable根据BrokerName尝试获取Broker信息，如果不存在，则新建BrokerData并放入到brokerAddrTable，registerFirst设置为true，如果存在直接替换原先的，registerFirst设置我false，表示非第一次注册。

如果Broker为Master，并且Broker Topic 配置信息发生变化或者是初次注册，则需要创建或者是更新Topic路由元数据，填充TopicQueueTable，其实是为默认主题注册路由信息，其中包含MixAll.DEFAULT\_TOPIC的路由信息。

根据TopicConfig创建QueueData数据结构，然后更新topicQueueTable，

更新BrokerLiveInfo，存活Broker信息表，BrokerLiveInfo是执行路由删除的重要依据。

注册Broker的过滤器Server地址列表，一个Broker会关联多个FilterServer消息过滤器，如果此Broker为从节点，则需要查找该Broker的Master的节点信息，并更新对应的masterAddr属性。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1lhbmd6YWlMZUhlSGU_size_16_color_FFFFFF_t_70 1][]

3、路由删除：

Broker每隔30秒向NameServer发送一个心跳包，心跳包中包含BrokerID、Broker地址、broker名称、Broker所属集群名称、Broker关联的FilterServer列表，但是如果Broker宕机，NameServer无法收到心跳包，此时NameServer需要去掉这些无效的Broker，NameServer会每隔10秒，扫描brokerLiveTable状态表，如果brokerLive的lastUpdateTimestamp的时间戳间隔超过120秒，则认为Broker失效，移除该Broker，关闭与Broker的连接，并同时更新，topicQueueTable、brokerAddrTable、brokerLiveTable、filterServerTable。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1lhbmd6YWlMZUhlSGU_size_16_color_FFFFFF_t_70 2][]

NameServer会每隔10秒执行一次，逻辑简单就是：遍历brokerLiveInfo路由表【HashMap】，检测BrokerLiveInfo的lastUpdateTimesTamp上次收到心跳包的时间间隔如果超过120秒，NameServer则认为该Broker已不可用，故需要将其移除，关闭Channel，然后删除与该Broker相关的路由信息，路由表维护过程，需要申请写锁。

维护brokerAddrTable遍历brokerAddrTable，从BrokerData的brokerAddrs中找到具体的Broker，从BrokerData中删除，如果移除后在BrokerData中不在包含其他的Broker，则在brokerAddrTable中移除该brokerName对应的条目。

根据BrokerName，从clusterAddrTable中找到Broker并从集群中移除，如果移除后集群中不包含任何Broker，则将该集群从clusterAddrTable中移除。

根据BrokerName遍历所有主题的队列，如果队列中包含了当前的Broker的队列，则移除，如果Topic只包含，待移除的Broker的队列的话，从路由表中删除该Topic。

释放锁，完成路由删除。

4、路由发现：

RocketMQ的路由发现不是实时的【非实时】，当Topic路由出现变化后，NameServer不主动推送给客户端，而是由客户端定时拉取主题最新的的路由。

根据主题名称拉取路由信息的命令编码为： GET\_ROUTEINFO\_BY\_TOPIC。

//TopicRouteData
    //顺序消息配置内容，来自于kvConfig
    private String orderTopicConf;
    //topic队列元数据
    private List<QueueData> queueDatas;
    //topic分布的broker元数据
    private List<BrokerData> brokerDatas;
    //broker上过滤服务器地址列表
    private HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;

**NameServer路由发现实现类：DefaultRequestProcessor\#getRouteInfoByTopic**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1lhbmd6YWlMZUhlSGU_size_16_color_FFFFFF_t_70 3][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1lhbmd6YWlMZUhlSGU_size_16_color_FFFFFF_t_70 4][]

NameServer路由发现与删除机制会存在这样一种情况：NameServer需要等待120秒才会将失效的Broker从路由表中删除，那如果在Broker故障期间，消息生产者根据主题获取到的路由信息包含已经宕机的Broker，会导致消息发送失败，RocketMQ是怎样解决的呢看-----[RocketMQ解决][RocketMQ]。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1lhbmd6YWlMZUhlSGU_size_16_color_FFFFFF_t_70]: /images/20211211/597a84daa4f94437a6cc5c6ae136a5fd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1lhbmd6YWlMZUhlSGU_size_16_color_FFFFFF_t_70 1]: /images/20211211/0c1ab8fea7cb4d328f2849aa9be2bb9a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1lhbmd6YWlMZUhlSGU_size_16_color_FFFFFF_t_70 2]: /images/20211211/a93f162d106f47f9a2b06123c1ed5674.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1lhbmd6YWlMZUhlSGU_size_16_color_FFFFFF_t_70 3]: /images/20211211/f3287523e95646f6a8fefb14f51ffceb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1lhbmd6YWlMZUhlSGU_size_16_color_FFFFFF_t_70 4]: /images/20211211/09e23ceaae6f42b1a287c206e21c8ed5.png
[RocketMQ]: https://blog.csdn.net/YangzaiLeHeHe/article/details/95456290