这篇文章主要讲解RocketMQ路由管理、服务注册及服务发现机制。
1、NameServer架构设计
消息中间件的设计思路一般是基于主题的订阅发布机制,消息生产者(Producer)发送某一主题的消息到消息服务器,消息服务器负责该消息的持久化存储,消息消费者(Consumer)订阅感兴趣的主题,消息服务器根据订阅信息(路由信息)将消息推送给消费者(push模式)或者消息消费者主动向消息服务器拉取消息(pull模式),从而实现消息生产者与消息消费者的解耦。为了避免因消息服务器的单点故障导致的整个系统瘫痪,通常会部署多台消息服务器共同承担消息的存储。那么消息生产者如何知道消息要发往哪台消息服务器呢?如 果某一台消息服务器宕机了,生产者如何在不重启服务的情况下感知呢?
为了解决上述问题,NameServer设计成支持集群模式,路由管理、服务注册、服务发现架构,如下图:
Broker消息服务器在启动时向所有NameServer注册,消息生产者 在发送消息之前先从NameServer获取Broker服务器的地址列表,然后 根据负载算法从列表中选择一台消息服务器发送消息。NameServer与 每台Broker服务器保持长连接,并间隔10s检测Broker是否存活,如果检测到Broker宕机,则从路由注册表中将其移除,但是路由变化不会马上通知消息生产者。这样设计是为了降低 NameServer实现的复杂性,因此需要在消息发送端提供容错机制来保证消息发送的高可用性。NameServer本身的高可用性可通过部署多台NameServer服务器来实现,但彼此之间互不通信。虽然NameServer服务器之间在某一时刻 的数据并不会完全相同,但对消息发送不会造成重大影响,无非就是短暂造成消息发送不均衡,这也是RocketMQ NameServer设计的一个亮点。
消息客户端与NameServer、Broker的交互:
Broker每隔30s向NameServer集群的每一台机器发送心跳包, 包含自身创建的topic路由等信息。消息客户端每隔30s向NameServer更新对应topic的路由信息。NameServer收到Broker发送的心跳包时会记录时间戳。NameServer每隔10s会扫描一次brokerLiveTable(存放心跳包的时间戳信息),如果在120s内没有收到心跳包,则认为Broker失效,更新topic的路由信息,将失效的Broker信息移除。
2、NameServer启动流程源码分析
namesrv模块下,找到NameServer启动类NamesrvStartup.java,重点关注NameServer相关启动参数。
首先是解析配置文件,需要填充NamesrvConfig、 NettyServerConfig属性值
方法流转:main0
->createNamesrvController
先创建NamesrvConfig(NameServer业务参 数)、NettyServerConfig(NameServer网络参数),然后在解析启动 时把指定的配置文件或启动命令中的选项值填充到NamesrvConfig、 NettyServerConfig对象中。参数来源有如下两种方式:
-c configFile通过-c命令指定配置文件的路径。
使用“–属性名 属性值”命令,例如 --listenPort 9876。
部分代码如下:
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();}}if (commandLine.hasOption('p')) {InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_CONSOLE_NAME);MixAll.printObjectProperties(console, namesrvConfig);MixAll.printObjectProperties(console, nettyServerConfig);System.exit(0);}MixAll.properties2Object(mandLine2Properties(commandLine), namesrvConfig);
NamesrvConfig对象中的默认参数:
/*** RocketMQ主目录,通过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对象中默认的参数:
/*** NameServer监听端口,该值默认会被初始化为9876*/private int listenPort = 8888;/*** Netty业务线程池线程个数*/private int serverWorkerThreads = 8;/*** Netty public任务线程池* 线程个数。Netty网络会根据业务类型创建不同的线程池,比如处理消* 息发送、消息消费、心跳检测等。如果该业务类型(RequestCode)未* 注册线程池,则由public线程池执行*/private int serverCallbackExecutorThreads = 0;/*** I/O线程池线程个数,主要是* NameServer、Broker端解析请求、返回相应的线程个数。这类线程主* 要用于处理网络请求,先解析请求包,然后转发到各个业务线程池完* 成具体的业务操作,最后将结果返回给调用方*/private int serverSelectorThreads = 3;/*** send oneway消息请求的并发度(Broker端参数)*/private int serverOnewaySemaphoreValue = 256;/*** 异步消息发送的最大并发度* (Broker端参数)*/private int serverAsyncSemaphoreValue = 64;/*** 网络连接最大空闲时* 间,默认为120s。如果连接空闲时间超过该参数设置的值,连接将被* 关闭*/private int serverChannelMaxIdleTimeSeconds = 120;/*** 网络socket发送缓存区大小,默认为64KB*/private int serverSocketSndBufSize = NettySystemConfig.socketSndbufSize;/*** 网络socket接收缓存区大小,默认为64KB*/private int serverSocketRcvBufSize = NettySystemConfig.socketRcvbufSize;private int writeBufferHighWaterMark = NettySystemConfig.writeBufferHighWaterMark;private int writeBufferLowWaterMark = NettySystemConfig.writeBufferLowWaterMark;private int serverSocketBacklog = NettySystemConfig.socketBacklog;/*** ByteBuffer是否开启缓存,建议开启*/private boolean serverPooledByteBufAllocatorEnable = true;/*** make make install*** ../glibc-2.10.1/configure \ --prefix=/usr \ --with-headers=/usr/include \* --host=x86_64-linux-gnu \ --build=x86_64-pc-linux-gnu \ --without-gd*//*** 是否启用Epoll I/O模型,Linux环境下建议开启*/private boolean useEpollNativeSelector = false;
注意: 在启动NameServer时,可以先使用./mqnameserver -c configFile -p 命令打印当前加载的配置属性。
根据启动属性创建NamesrvController实例并初始化, NameServerController实例为NameServer核心控制器:
方法流转:main0
->createNamesrvController
->start
->initialize
加载KV配置,先创建NettyServer网络处理对象,然后开启两个定时任务,在RocketMQ中此类定时任务统称为心跳检测。
1)定时任务1:NameServer每隔10s扫描一次Broker,移除处于未 激活状态的Broker。
2)定时任务2:NameServer每隔10min打印一次KV配置。
public boolean initialize() {//加载KV配置this.kvConfigManager.load();/*** 创建NettyServer网络处理对象*/this.remotingServer = new NettyRemotingServer(tyServerConfig, this.brokerHousekeepingService);this.remotingExecutor =Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));this.registerProcessor();/*** 定时任务1:NameServer每隔10s扫描一次Broker,移除处于未* 激活状态的Broker*/this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {NamesrvController.this.routeInfoManager.scanNotActiveBroker();}}, 5, 10, TimeUnit.SECONDS);/*** 定时任务2::NameServer每隔10min打印一次KV配置*/this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {NamesrvController.this.kvConfigManager.printAllPeriodically();}}, 1, 10, TimeUnit.MINUTES);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;}
方法流转:main0
->createNamesrvController
->start
注册JVM钩子函数并启动服务器,以便监听Broker、消息 生产者的网络请求。
这里展示一种常用的编程技巧,如果代码中使用了 线程池,一种优雅停机的方式就是注册一个JVM钩子函数,在JVM进程 关闭之前,先将线程池关闭,及时释放资源。
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钩子函数并启动服务器,以便监听Broker、消息* 生产者的网络请求*/Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, (Callable<Void>) () -> {controller.shutdown();return null;}));controller.start();return controller;}
3、NameServer路由注册、故障剔除
3.1、路由元信息
NameServer的路由实现类是 org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager。在了解路由注册之前,我们先看一下NameServer到底存储了哪些信息。
RocketMQ基于订阅发布机制,一个topic拥有多个消息队列,一个 Broker默认为每一主题创建4个读队列和4个写队列。多个Broker组成 一个集群,BrokerName由相同的多台Broker组成主从架构, brokerId=0代表主节点,brokerId>0表示从节点。BrokerLiveInfo中 的lastUpdateTimestamp存储上次收到Broker心跳包的时间。
/*** topic消息队列的路由信息,消息发送时根* 据路由表进行负载均衡*/private final HashMap<String/* topic */, Map<String /* brokerName */ , 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;
3.2、路由注册
RocketMQ路由注册是通过Broker与NameServer的心跳功能实现的。Broker启动时向集群中所有的NameServer发送心跳,每隔30s 向集群中所有的NameServer发送心跳,NameServer收到Broker心跳时会先更新brokerLiveTable缓存中BrokerLiveInfo的 lastUpdateTimestamp,然后每隔10s扫描一次brokerLiveTable,如果 连续120s没有收到心跳,NameServer将移除该Broker的路由信息, 同时关闭Socket连接。
Broker发送心跳包
在broker模块下,找到BrokerStartup.java启动类。
方法流转:main
->start
->BrokerController#start
Broker发送心跳包的核心代码:
/*** 向nameServer发送心跳,实现注册,默认30秒发送一次*/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);
方法流转:main
->start
->BrokerController#start
->BrokerController#start#registerBrokerAll
->BrokerController#start#registerBrokerAll#doRegisterBrokerAll
->BrokerOuterAPI#registerBrokerAll
该方法遍历NameServer列表,Broker消息服务器依次向 NameServer发送心跳。
/*** 遍历所有NameServer 列表*/for (final String namesrvAddr : nameServerAddressList) {brokerOuterExecutor.execute(new Runnable() {@Overridepublic void run() {try {//向 NameServer 注册RegisterBrokerResult result = registerBroker(namesrvAddr, oneway, timeoutMills, requestHeader, body);if (result != null) {registerBrokerResultList.add(result);}log.info("register broker[{}]to name server {} OK", brokerId, namesrvAddr);} catch (Exception e) {log.warn("registerBroker Exception, {}", namesrvAddr, e);} finally {countDownLatch.countDown();}}});}
发送心跳包的具体逻辑,首先封装请求包头 (Header)。
1)brokerAddr:broker地址。
2)brokerId:brokerId=0表示主节点,brokerId>0表示从节点。
3)brokerName:broker名称。
4)clusterName:集群名称。
5)haServerAddr:主节点地址,初次请求时该值为空,从节点向 NameServer注册后返回。
6)requestBody:
topicConfigWrapper,主题配置,topicConfigWrapper内部封装 的是TopicConfig Manager中的topicConfigTable,内部存储的是 Broker启动时默认的一些topic,如MixAll.SELF_TEST_TOPIC、 MixAll.DEFAULT_TOPIC(AutoCreateTopic-Enable=true)、 MixAll.BENCHMARK_TOPIC、MixAll.OFFSET_MOVED_EVENT、 BrokerConfig#brokerClusterName、BrokerConfig#brokerName。 Broker中topic默认存储在 ${Rocket_Home}/store/confg/topics.json中。filterServerList,消息过滤服务器列表。
private RegisterBrokerResult registerBroker(final String namesrvAddr,final boolean oneway,final int timeoutMills,final RegisterBrokerRequestHeader requestHeader,final byte[] body) throws RemotingCommandException, MQBrokerException, RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException,InterruptedException {RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.REGISTER_BROKER, requestHeader);request.setBody(body);if (oneway) {try {this.remotingClient.invokeOneway(namesrvAddr, request, timeoutMills);} catch (RemotingTooMuchRequestException e) {// Ignore}return null;}RemotingCommand response = this.remotingClient.invokeSync(namesrvAddr, request, timeoutMills);assert response != null;switch (response.getCode()) {case ResponseCode.SUCCESS: {RegisterBrokerResponseHeader responseHeader =(RegisterBrokerResponseHeader) response.decodeCommandCustomHeader(RegisterBrokerResponseHeader.class);RegisterBrokerResult result = new RegisterBrokerResult();result.setMasterAddr(responseHeader.getMasterAddr());result.setHaServerAddr(responseHeader.getHaServerAddr());if (response.getBody() != null) {result.setKvTable(KVTable.decode(response.getBody(), KVTable.class));}return result;}default:break;}throw new MQBrokerException(response.getCode(), response.getRemark(), requestHeader == null ? null : requestHeader.getBrokerAddr());}
NameServer处理心跳包
在namesrv模块下,找到org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor是网络处理器解析请求类型,如果请求类型为 RequestCode.REGISTER_BROKER,则请求最终转发到RouteInfoMan ager#registerBroker。
switch (request.getCode()) {case RequestCode.PUT_KV_CONFIG:return this.putKVConfig(ctx, request);case RequestCode.GET_KV_CONFIG:return this.getKVConfig(ctx, request);case RequestCode.DELETE_KV_CONFIG:return this.deleteKVConfig(ctx, request);case RequestCode.QUERY_DATA_VERSION:return queryBrokerTopicConfig(ctx, request);/*** 如果请求类型为* RequestCode.REGISTER_BROKER,则请求最终转发到RouteInfoMan* ager#registerBroker*/case RequestCode.REGISTER_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);}case RequestCode.UNREGISTER_BROKER:return this.unregisterBroker(ctx, request);case RequestCode.GET_ROUTEINFO_BY_TOPIC:return this.getRouteInfoByTopic(ctx, request);case RequestCode.GET_BROKER_CLUSTER_INFO:return this.getBrokerClusterInfo(ctx, request);case RequestCode.WIPE_WRITE_PERM_OF_BROKER:return this.wipeWritePermOfBroker(ctx, request);case RequestCode.ADD_WRITE_PERM_OF_BROKER:return this.addWritePermOfBroker(ctx, request);case RequestCode.GET_ALL_TOPIC_LIST_FROM_NAMESERVER:return getAllTopicListFromNameserver(ctx, request);case RequestCode.DELETE_TOPIC_IN_NAMESRV:return deleteTopicInNamesrv(ctx, request);case RequestCode.GET_KVLIST_BY_NAMESPACE:return this.getKVListByNamespace(ctx, request);case RequestCode.GET_TOPICS_BY_CLUSTER:return this.getTopicsByCluster(ctx, request);case RequestCode.GET_SYSTEM_TOPIC_LIST_FROM_NS:return this.getSystemTopicListFromNs(ctx, request);case RequestCode.GET_UNIT_TOPIC_LIST:return this.getUnitTopicList(ctx, request);case RequestCode.GET_HAS_UNIT_SUB_TOPIC_LIST:return this.getHasUnitSubTopicList(ctx, request);case RequestCode.GET_HAS_UNIT_SUB_UNUNIT_TOPIC_LIST:return this.getHasUnitSubUnUnitTopicList(ctx, request);case RequestCode.UPDATE_NAMESRV_CONFIG:return this.updateConfig(ctx, request);case RequestCode.GET_NAMESRV_CONFIG:return this.getConfig(ctx, request);default:break;}
路由注册需要加写锁,防止并发修改RouteInfoManager 中的路由表。首先判断Broker所属集群是否存在,如果不存在,则创 建集群,然后将broker名加入集群Broker集合。
this.lock.writeLock().lockInterruptibly();Set<String> brokerNames = this.clusterAddrTable.get(clusterName);if (null == brokerNames) {brokerNames = new HashSet<String>();this.clusterAddrTable.put(clusterName, brokerNames);}brokerNames.add(brokerName);
维护BrokerData信息,首先从brokerAddrTable中根据 broker名尝试获取Broker信息,如果不存在,则新建BrokerData并放 入brokerAddrTable,registerFirst设置为true;如果存在,直接替 换原先的Broker信息,registerFirst设置为false,表示非第一次注册。
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);}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()) {log.debug("remove entry {} from brokerData", item);it.remove();}}String oldAddr = brokerData.getBrokerAddrs().put(brokerId, brokerAddr);if (MixAll.MASTER_ID == brokerId) {log.info("cluster [{}] brokerName [{}] master address change from {} to {}",brokerData.getCluster(), brokerData.getBrokerName(), oldAddr, brokerAddr);}registerFirst = registerFirst || (null == oldAddr);
如果Broker为主节点,并且Broker的topic配置信息发生 变化或者是初次注册,则需要创建或更新topic路由元数据,并填充 topicQueueTable,其实就是为默认主题自动注册路由信息,其中包含 MixAll.DEFAULT_TOPIC的路由信息。当消息生产者发送主题时,如果该主题未创建,并且BrokerConfig的autoCreateTopicEnable为true, 则返回MixAll.DEFAULT_TOPIC的路由信息。
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());}}}}
根据topicConfig创建QueueData数据结构,然后更新 topicQueueTable。
private void createAndUpdateQueueData(final String brokerName, final TopicConfig topicConfig) {QueueData queueData = new QueueData();queueData.setBrokerName(brokerName);queueData.setWriteQueueNums(topicConfig.getWriteQueueNums());queueData.setReadQueueNums(topicConfig.getReadQueueNums());queueData.setPerm(topicConfig.getPerm());queueData.setTopicSysFlag(topicConfig.getTopicSysFlag());Map<String, QueueData> queueDataMap = this.topicQueueTable.get(topicConfig.getTopicName());if (null == queueDataMap) {queueDataMap = new HashMap<>();queueDataMap.put(queueData.getBrokerName(), queueData);this.topicQueueTable.put(topicConfig.getTopicName(), queueDataMap);log.info("new topic registered, {} {}", topicConfig.getTopicName(), queueData);} else {QueueData old = queueDataMap.put(queueData.getBrokerName(), queueData);if (old != null && !old.equals(queueData)) {log.info("topic changed, {} OLD: {} NEW: {}", topicConfig.getTopicName(), old,queueData);}}}
更新BrokerLiveInfo,存储状态正常的Broker信息表, BrokeLiveInfo是执行路由删除操作的重要依据。
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消息过滤服务器。如果此Broker为从节点,则需要查找该Broker的主节点信息,并更新对应的masterAddr属性。
if (filterServerList != null) {if (filterServerList.isEmpty()) {this.filterServerTable.remove(brokerAddr);} else {this.filterServerTable.put(brokerAddr, filterServerList);}}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);}}}
设计亮点:
NameServer与Broker保持长连接,Broker的状态信息存储 在brokerLive-Table中,NameServer每收到一个心跳包,将更新 brokerLiveTable中关于Broker的状态信息以及路由表 (topicQueueTable、brokerAddrTable、brokerLiveTable、 filterServer-Table)。更新上述路由表(HashTable)使用了锁粒度 较少的读写锁,允许多个消息发送者并发读操作,保证消息发送时的 高并发。同一时刻NameServer只处理一个Broker心跳包,多个心跳包 请求串行执行。这也是读写锁经典的使用场景。
3.3、路由删除
Broker每隔30s向NameServer发送一个心跳包,心 跳包中包含BrokerId、Broker地址、Broker名称、Broker所属集群名 称。如果Broker宕机,NameServer无法收到心跳包,NameServer会每隔10s扫描一次brokerLiveTable状态表,如果 BrokerLive的lastUpdate-Timestamp时间戳距当前时间超过120s,则 认为Broker失效,移除该Broker,关闭与Broker的连接,同时更新 topicQueueTable、brokerAddrTable、brokerLiveTable、 filterServerTable。
RocketMQ有两个触发点来触发路由删除操作。
1)NameServer定时扫描brokerLiveTable,检测上次心跳包与当 前系统时间的时间戳,如果时间戳大于120s,则需要移除该Broker信息。
2)Broker在正常关闭的情况下,会执行unregisterBroker指令。
因为不管是何种方式触发的路由删除,删除方法是一样的,都是 从topicQueueTable、brokerAddrTable、brokerLiveTable、 filterServerTable中删除与该Broker相关的信息,所以RocketMQ用这 两种方式维护路由信息时会抽取公共代码,本节将以第一种方式为例 展开分析。
方法流转:RouteInfoManager#scanNotActiveBroker
scanNotActiveBroker在NameServer中每10s执行一次。 逻辑也很简单,先遍历brokerLiveInfo路由表(HashMap),检测 BrokerLiveInfo的LastUpdateTimestamp上次收到心跳包的时间,如果超过120s,则认为该Broker已不可用,然后将它移除并关闭连接,最后删除与该Broker相关的路由信息。
public int scanNotActiveBroker() {int removeCount = 0;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());removeCount++;}}return removeCount;}
申请写锁。根据brokerAddress从brokerLiveTable、 filterServerTable中移除Broker相关的信息
this.lock.writeLock().lockInterruptibly();this.brokerLiveTable.remove(brokerAddrFound);this.filterServerTable.remove(brokerAddrFound);
维护brokerAddrTable。遍历HashMap brokerAddrTable,从BrokerData的 HashMap brokerAddrs中,找到具体的Broker,从BrokerData中将其移除。如果移除后在BrokerData中不再包含其他Broker,则在brokerAddrTable中 移除该brokerName对应的条目。
String brokerNameFound = null;boolean removeBrokerName = false;Iterator<Entry<String, BrokerData>> itBrokerAddrTable =this.brokerAddrTable.entrySet().iterator();while (itBrokerAddrTable.hasNext() && (null == brokerNameFound)) {BrokerData brokerData = itBrokerAddrTable.next().getValue();Iterator<Entry<Long, String>> it = brokerData.getBrokerAddrs().entrySet().iterator();while (it.hasNext()) {Entry<Long, String> entry = it.next();Long brokerId = entry.getKey();String brokerAddr = entry.getValue();if (brokerAddr.equals(brokerAddrFound)) {brokerNameFound = brokerData.getBrokerName();it.remove();log.info("remove brokerAddr[{}, {}] from brokerAddrTable, because channel destroyed",brokerId, brokerAddr);break;}}if (brokerData.getBrokerAddrs().isEmpty()) {removeBrokerName = true;itBrokerAddrTable.remove();log.info("remove brokerName[{}] from brokerAddrTable, because channel destroyed",brokerData.getBrokerName());}
根据BrokerName,从clusterAddrTable中找到Broker并将其从集群中移除。如果移除后,集群中不包含任何Broker,则将该集群从clusterAddrTable中移除。
if (brokerNameFound != null && removeBrokerName) {Iterator<Entry<String, Set<String>>> it = this.clusterAddrTable.entrySet().iterator();while (it.hasNext()) {Entry<String, Set<String>> entry = it.next();String clusterName = entry.getKey();Set<String> brokerNames = entry.getValue();boolean removed = brokerNames.remove(brokerNameFound);if (removed) {log.info("remove brokerName[{}], clusterName[{}] from clusterAddrTable, because channel destroyed",brokerNameFound, clusterName);if (brokerNames.isEmpty()) {log.info("remove the clusterName[{}] from clusterAddrTable, because channel destroyed and no broker in this cluster",clusterName);it.remove();}break;}}}
根据BrokerName,遍历所有主题的队列,如果队列中包含当前Broker的队列,则移除,如果topic只包含待移除Broker的队列,从路由表中删除该topic。
if (removeBrokerName) {String finalBrokerNameFound = brokerNameFound;Set<String> needRemoveTopic = new HashSet<>();topicQueueTable.forEach((topic, queueDataMap) -> {QueueData old = queueDataMap.remove(finalBrokerNameFound);log.info("remove topic[{} {}], from topicQueueTable, because channel destroyed",topic, old);if (queueDataMap.size() == 0) {log.info("remove topic[{}] all queue, from topicQueueTable, because channel destroyed",topic);needRemoveTopic.add(topic);}});needRemoveTopic.forEach(topicQueueTable::remove);}
释放锁,完成路由删除。
finally {this.lock.writeLock().unlock();}
3.4、路由发现
RocketMQ路由发现是非实时的,当topic路由出现变化后, NameServer不主动推送给客户端,而是由客户端定时拉取主题最新的 路由。根据主题名称拉取路由信息的命令编码为 GET_ROUTEINTO_BY_TOPIC。RocketMQ路由结果如图:
public class TopicRouteData extends RemotingSerializable {/*** 顺序消息配置内容,来自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
。
public RemotingCommand getRouteInfoByTopic(ChannelHandlerContext ctx,RemotingCommand request) throws RemotingCommandException {final RemotingCommand response = RemotingCommand.createResponseCommand(null);final GetRouteInfoRequestHeader requestHeader =(GetRouteInfoRequestHeader) request.decodeCommandCustomHeader(GetRouteInfoRequestHeader.class);TopicRouteData topicRouteData = this.namesrvController.getRouteInfoManager().pickupTopicRouteData(requestHeader.getTopic());if (topicRouteData != null) {if (this.namesrvController.getNamesrvConfig().isOrderMessageEnable()) {String orderTopicConf =this.namesrvController.getKvConfigManager().getKVConfig(NamesrvUtil.NAMESPACE_ORDER_TOPIC_CONFIG,requestHeader.getTopic());topicRouteData.setOrderTopicConf(orderTopicConf);}byte[] content = topicRouteData.encode();response.setBody(content);response.setCode(ResponseCode.SUCCESS);response.setRemark(null);return response;}response.setCode(ResponseCode.TOPIC_NOT_EXIST);response.setRemark("No topic route info in name server for the topic: " + requestHeader.getTopic()+ FAQUrl.suggestTodo(FAQUrl.APPLY_TOPIC_URL));return response;}
调用RouterInfoManager的方法,从路由表 topicQueueTable、brokerAddrTable、filterServerTable中分别填充 TopicRouteData中的List、List和 filterServer地址表。
如果找到主题对应的路由信息并且该主题为顺序消息,则从NameServer KVConfig中获取关于顺序消息相关的配置填充路由信息。如果找不到路由信息Code,则使用TOPIC_NOT_EXISTS,表示没有找到对应的路由。