网站建设文化传播有限公司/搜索关键词

RocketMQ 路由管理 服务注册及服务发现由NameServer提供。

服务发现：
分布式服务 SOA（全称：Service Oriented Architecture 面向服务的架构）构体系中会有服务注册中心，分布式服务 SOA 的注册中心主要提供服务调用的解析服务，即指引服务调用方（消费者）找到“远方”的服务提供者，完成网络通信。

NameServer 介绍

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

NameServer 与每台 Broker 服务器保持长连接，并间隔 30s 检测 Broker 是否存活，如果检测到 Broker 从路由注册表中将其移除，但是路由变化不会马上通知消息生产者。

为什么要这样设计呢？
为了降低 NameServer 实现的复杂，在消息发送端提供容错机制来保证消息发送的高可用性。

NameServer 本身的高可用可通过部署多台 NameServer 服务器来实现，但彼此之间互不通信，也就是NameServer 务器之间在某一时刻的数据并不会完全相同，但这对消息发送不会造成任何影响，这也是RocketMQ NameServer 设计的 一个亮点， RocketMQ NameServer 计追求简单高效。

NameServer核心类解析

1. RouteInfoManager：用于管理心跳信息以及路由管理
2. KVConfigManager：用于管理以及加载KV配置。加载NamesrvController指定的kvConfig配置文件(常为xxx/kvConfig.json)到内存
3. NameSrvController：NameSever控制器，有点像三层架构的Controller层，用于请求转发，包装核心类，并且负责NameServer服务器的初始化和关闭操作。
4. NamesrvStartUp：NameServer服务启动类，帮助读取配置，创建Controller并启动服务。
5. NameSrvConfig：NameServer业务参数配置类
6. NettyServerConfig：NameServer网络参数配置类，因为涉及到网络交互
7. BrokerHousekeepingServer：事件监听相关的，用于处理与Broker的Channel事件，例如处理连接事件、关闭事件、异常事件等
8. NettyRemotingServer：用于实现网络交互的，并且还存在一些内部类。本章节不会详细介绍，会放到通信章节详细介绍，简单理解就是通过此实例与Broker、Producer、Consumer进行交互与数据传输。
9. DefaultRequestProcessor：默认请求处理器，用于处理请求。

NameServer 启动

NameServer启动入口：org.apache.rocketmq.namesrv.NamesrvStartup。实际就是一个main方法，启动后如下表示启动成功：

NamesrvStartup#main0

    public static NamesrvController main0(String[] args) {try {// 解析配置并且创建NameSrv控制器NamesrvController controller = createNamesrvController(args);// 执行NamesrvController initialize()逻辑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;}

NamesrvStartup#createNamesrvController

public static NamesrvController createNamesrvController(String[] args) throws IOException, JoranException {// 设置rocketmq.remoting.version属性// 其实是设置rocketMq的版本System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));//PackageConflictDetect.detectFastjson();//  构建命令行选项Options options = ServerUtil.buildCommandlineOptions(new Options());// 解析命令行相关 生成命令行对象 用于处于命令行指令commandLine = ServerUtil.parseCmdLine("mqnamesrv", args, buildCommandlineOptions(options), new PosixParser());if (null == commandLine) {System.exit(-1);return null;}// 创建NameSrv 业务参数配置类final NamesrvConfig namesrvConfig = new NamesrvConfig();// 创建NameServer 网络参数配置类final NettyServerConfig nettyServerConfig = new NettyServerConfig();// 修改启动端口为9876nettyServerConfig.setListenPort(9876);// 条件成立：说明命令行输入了-c参数 java CLITester -c configFile地址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);// 尝试将properties解析的属性赋值给namesrvConfig的属性中MixAll.properties2Object(properties, namesrvConfig);// 尝试将properties解析的属性赋值给nettyServerConfig的属性中MixAll.properties2Object(properties, nettyServerConfig);// 设置文件路径namesrvConfig.setConfigStorePath(file);System.out.printf("load config properties file OK, %s%n", file);in.close();}}// 条件成立：说明命令行输入了-p参数 java CLITester -p// 会进行打印配置项相关if (commandLine.hasOption('p')) {InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_CONSOLE_NAME);MixAll.printObjectProperties(console, namesrvConfig);MixAll.printObjectProperties(console, nettyServerConfig);System.exit(0);}// 尝试将命令行配置的参数都设置到namesrvConfig的属性中MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);// 条件成立 程序退出 因为需要配置rocketMq Homeif (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);}// 日志相关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);MixAll.printObjectProperties(log, namesrvConfig);MixAll.printObjectProperties(log, nettyServerConfig);// 生成NamesrvController对象// 参数一：namesrv配置对象// 参数二：netty服务启动配置对象final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);// remember all configs to prevent discard// 记住所有的配置以防止丢弃controller.getConfiguration().registerConfig(properties);return controller;}

NamesrvStartup#start

public static NamesrvController start(final NamesrvController controller) throws Exception {if (null == controller) {throw new IllegalArgumentException("NamesrvController is null");}// 执行controller.initialize() 和  controller.start()逻辑// 返回值：初始化结果 true 代表初始化成功 false代表失败boolean initResult = controller.initialize();// 条件成立：说明初始化失败if (!initResult) {controller.shutdown();System.exit(-3);}// 向JVM中增加一个关闭的钩子 当JVM关闭的时候，会执行ShutdownHookThread.run()方法// ShutdownHookThread.run()方法内部又会执行Callable.call()方法Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<Void>() {@Overridepublic Void call() throws Exception {controller.shutdown();return null;}}));// 启动namesrcController对象// 主要是启动内部的remotingServer对象controller.start();return controller;}

NamesrvController#initialize

/*** 1.加载kv配置* 2.创建NettyRemotingServer对象（后续通信章节会详细讲解）* 3.注册默认请求处理器 （后续通信章节会详细讲解）* 4.注册两个定时任务* 4.1 每隔10秒扫描是否有满足失效条件的Broker* 4.2 每隔10分钟打印所有的配置* @return*/public boolean initialize() {// 加载kv配置this.kvConfigManager.load();// 创建网络层server对象// 参数一：netty服务器启动配置类// 参数二：broker内部处理服务 用于监听不同的事件 处理不同的对象this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);// 生成网络层线程池this.remotingExecutor =Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));// 注册处理器this.registerProcessor();// 注册定时任务 延时5秒执行，10秒为一个周期// 作用：扫描下线的brokerthis.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {NamesrvController.this.routeInfoManager.scanNotActiveBroker();}}, 5, 10, TimeUnit.SECONDS);// 注册定时任务 延时1秒后执行 10分钟为一个周期// 作用：打印所有配置相关this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {NamesrvController.this.kvConfigManager.printAllPeriodically();}}, 1, 10, TimeUnit.MINUTES);// ssl相关 注册一个监听器去加载SslContext 不关注此逻辑if (TlsSystemConfig.tlsMode != TlsMode.DISABLED) {// Register a listener to reload SslContexttry {fileWatchService = new FileWatchService(new String[] {TlsSystemConfig.tlsServerCertPath,TlsSystemConfig.tlsServerKeyPath,TlsSystemConfig.tlsServerTrustCertPath},new FileWatchService.Listener() {boolean certChanged, keyChanged = false;@Overridepublic 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;}

private void registerProcessor() {// 测试用的，默认不成立if (namesrvConfig.isClusterTest()) {this.remotingServer.registerDefaultProcessor(new ClusterTestRequestProcessor(this, namesrvConfig.getProductEnvName()),this.remotingExecutor);} else {// 默认走这里// 向网络层服务对象注册一个默认的处理器// 参数一：默认请求处理器// 参数二：网络层线程池 用于执行默认请求处理器里面的逻辑this.remotingServer.registerDefaultProcessor(new DefaultRequestProcessor(this), this.remotingExecutor);}}

NameServer 配置

NamesrvConfig

NamesrvConfig是NameServer 业务参数配置。

public class NamesrvConfig {//rocketmq home目录//先取 rocketmq.home.dir 不存在后 取 ROCKETMQ_HOME//可以通过 -Drocketmq.home.dir=path 或通过设置环境变量 ROCKETMQ_HOME 来配置 RocketMQ 的主目录private String rocketmqHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY, System.getenv(MixAll.ROCKETMQ_HOME_ENV));//NameServer 存储 KV配置属性的持久化路径private String kvConfigPath = System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "kvConfig.json";//NameServer  默认配置文件路径，不生效。NameServer 启动时如果要通过配置文件配置 NameServer 启动属性的话，请使用-c选项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;
}    

NettyServerConfig

NettyServerConfig是NameServer 网络参数配置。

public class NettyServerConfig implements Cloneable {//NameServer 监昕端口，该值默认会被初始化为 9876private int listenPort = 8888;//Netty 业务线程池线程个数private int serverWorkerThreads = 8;//Netty public 务线程池线程个数， //Netty 网络设计，根据业务类型会创建不同的线程池，比如处理消息发送、消息消费、心跳检测等//如果该业务类型（RequestCode）未注册线程池，则由public 线程池执行private int serverCallbackExecutorThreads = 0;//IO 线程池线程个数，主要是 NameServer、Broker 端解析请求、返回相应的线程个数，这类线程主要是处理网络请求的，解析请求包， 然后转发到各个业务线程池完成具体的业务操作，然后将结果再返回调用方private int serverSelectorThreads = 3;// send oneway 消息请求并发度（ Broker 端参数）private int serverOnewaySemaphoreValue = 256;//异步消息发送最大并发度（ Broker 端参数）private int serverAsyncSemaphoreValue = 64;//网络连接最大空闲时间，默认 120s 如果连接空闲时间超过该参数设置的值，连接将被关闭private int serverChannelMaxIdleTimeSeconds = 120;//网络socket 发送缓存区大小， 默认 64kprivate int serverSocketSndBufSize = NettySystemConfig.socketSndbufSize;//网络 socket 接收缓存区大 ，默认 64kprivate int serverSocketRcvBufSize = NettySystemConfig.socketRcvbufSize;private int writeBufferHighWaterMark = NettySystemConfig.writeBufferHighWaterMark;private int writeBufferLowWaterMark = NettySystemConfig.writeBufferLowWaterMark;private int serverSocketBacklog = NettySystemConfig.socketBacklog;//ByteBuffer 是否开启缓存，建议开启private boolean serverPooledByteBufAllocatorEnable = true;//是否启用 Epoll IO 模型， Linux 环境建议开启private boolean useEpollNativeSelector = false;}   

定时任务

NameServer 开启两个定时任务，在 RocketMQ中此类定时任务统称为心跳检测。

Broker状态检测

NameServer 每隔 10s 扫描一次 Broker 移除处于不激活状态的 Broker。

由org.apache.rocketmq.namesrv.NamesrvController#initialize调用：

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

打印KV配置

NameServer 每隔 10 分钟打印一次 KV配置。
由org.apache.rocketmq.namesrv.NamesrvController#initialize调用：

public void printAllPeriodically() {try {this.lock.readLock().lockInterruptibly();try {log.info("--------------------------------------------------------");{log.info("configTable SIZE: {}", this.configTable.size());Iterator<Entry<String, HashMap<String, String>>> it =this.configTable.entrySet().iterator();while (it.hasNext()) {Entry<String, HashMap<String, String>> next = it.next();Iterator<Entry<String, String>> itSub = next.getValue().entrySet().iterator();while (itSub.hasNext()) {Entry<String, String> nextSub = itSub.next();log.info("configTable NS: {} Key: {} Value: {}", next.getKey(), nextSub.getKey(),nextSub.getValue());}}}} finally {this.lock.readLock().unlock();}} catch (InterruptedException e) {log.error("printAllPeriodically InterruptedException", e);}}

NameServer 动态路由发现与剔除机制

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

路由元信息

org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager为路由信息管理类，包含以下信息：

public class RouteInfoManager {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;
}

QueueData队列信息：

public class QueueData implements Comparable<QueueData> {//broker名称private String brokerName;//读队列数量private int readQueueNums;//写队列数量private int writeQueueNums;//读写权限//取值：org.apache.rocketmq.common.constant.PermName//6：同时支持读写//4：禁写//2：禁读private int perm;// topic 同步标记//取值：org.apache.rocketmq.common.sysflag.TopicSysFlagprivate int topicSysFlag;
}

一个Broker 为每一topic默认创建4个读队列和4个写队列 。

BrokerData Broker信息：

public class BrokerData implements Comparable<BrokerData> {//broker.conf brokerClusterName 配置private String cluster;//broker.conf brokerName 配置private String brokerName;//brokerld =O 表示Master,大于0表示从Slaveprivate HashMap<Long/* brokerId */, String/* broker address */> brokerAddrs;
}

BrokerLiveInfo Broker存活信息：

class BrokerLiveInfo {//最后一次更新时间， namesrv 收到 broker 心跳后，会更新该信息。private long lastUpdateTimestamp;private DataVersion dataVersion;// netty 中的 channel, 即 socketprivate Channel channel;// BrokerController#getHAServerAddr// 格式如下：ip:portprivate String haServerAddr;
}

路由信息注册

RocketMQ 路由注册是通过 Broker与NameServer 的心跳功能实现的 。

1.Broker想NameServer发送心跳包：

this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {@Overridepublic 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);

Broker 启动时向集群中 所有的 NameServer 发送心跳包，每隔 30s 向集群 中所有 NameServer 发送心跳包， NameServer 收到 Broker 跳包时会更新 brokerLiveTable 缓存中 BrokerLivelnfo的lastUpdateTimestamp 。

心跳包中包含Broker Id、 Broker 地址、Broker 名称、 Broker 所属集群名称、Broker 关联 FilterServer列表。

2.NameServer处理心跳包
org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor#processRequest 接收到code为RequestCode#REGISTER_BROKER请求后分发到RouteInfoManager#registerBroker处理。

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();//更新集群中 BrokerNameSet<String> brokerNames = this.clusterAddrTable.get(clusterName);if (null == brokerNames) {brokerNames = new HashSet<String>();this.clusterAddrTable.put(clusterName, brokerNames);}brokerNames.add(brokerName);boolean registerFirst = false;//更新Broker 地址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 brokerAddrTableIterator<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);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());}}}}//更新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);}//注册Broker 的过滤器 Server 地址列表。一个Broker 会关联多 FilterServer消息过滤服务器if (filterServerList != null) {if (filterServerList.isEmpty()) {this.filterServerTable.remove(brokerAddr);} else {this.filterServerTable.put(brokerAddr, filterServerList);}}//非 master Broker时，返回master Broker 地址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.NameServer 定时检查Broker心跳是否超时
NameServer 每隔 10s 扫描 brokerLiveTable ，如果连续 120s 没有收到心跳包， NameServer 将移除该 Broker 的路由信息同时关闭Socket 连接。

路由信息删除

RocktMQ 两个触发点来触发路由删除：

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

由于不管是何种方式触发的路由删除，路由删除的方法都是一样的，就是从 topicQueueTable、brokerAddrTable 、brokerLiveTable、filterServerTable 删除与 Broker 关的信息。

路由发现

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

client向NameServer拉取Topic路由信息时，请求code：RequestCode#GET_ROUTEINFO_BY_TOPIC。

NameServer返回TopicRouteData：

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