禅城建网站,榆林网站建设公司,二级域名网站好不好,湛江的网站1.前言
此文章是在儒猿课程中的学习笔记#xff0c;感兴趣的想看原来的课程可以去咨询儒猿课堂《从0开始带你成为RocketMQ高手》#xff0c;我本人觉得这个作者还是不错#xff0c;都是从场景来进行分析#xff0c;感觉还是挺适合我这种小白的。这块主要都是我自己的学习笔…1.前言
此文章是在儒猿课程中的学习笔记感兴趣的想看原来的课程可以去咨询儒猿课堂《从0开始带你成为RocketMQ高手》我本人觉得这个作者还是不错都是从场景来进行分析感觉还是挺适合我这种小白的。这块主要都是我自己的学习笔记。想看原版的还是去儒猿课堂。
这篇文章紧挨着上一篇文章NameServer接收Broker的注册请求处理-CSDN博客
前面我们已经看到了NameServer处理Broker的注册请求的源码流程NameServer核心就是基于Netty服务器来接收Broker注册请求然后交给DefaultRequestProcessor这个请求处理组件来注册Broker注册请求。Broker的注册逻辑是放在RouteInfoManager这个路由数据管理组件实现的。最终这些路由数据都会存放到RouteInfoManager内部的一些Map数据结构组成的路由数据表中。
2.Broker 如何进行定时发送心跳
那么Broker是如何定时发送心跳到NameServer让NameServer感知到Broker一直存活这个Broker中的发送注册请求给NameServer这个代码的入口还是在BrokerController中
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);
这块代码就是启动了一个定时调度的任务默认是每30s就会执行Broker的注册请求。
我们紧接着看下RouteInfoManager的注册方法的代码逻辑 public RegisterBrokerResult registerBroker(final String clusterName,final String brokerAddr,final String brokerName,final long brokerId,final String haServerAddr,final TopicConfigSerializeWrapper topicConfigWrapper,final ListString filterServerList,final Channel channel) {RegisterBrokerResult result new RegisterBrokerResult();try {try {//这里有一个加锁的操作this.lock.writeLock().lockInterruptibly();//从clusterAddrTable中获取clusterName对应的BrokeName的set集合//如果对应的clusterName中没有set集合那么就新建一个set集合SetString brokerNames this.clusterAddrTable.computeIfAbsent(clusterName, k - new HashSet());//直接将brokerName添加到set集合中//每次心跳过来了 都要进行添加 用set集合保证brokerName的唯一性brokerNames.add(brokerName);boolean registerFirst false;//这里是根据BrokerName进行获取Broker的数据BrokerData brokerData this.brokerAddrTable.get(brokerName);if (null brokerData) {//如果为空的话 设置registerFirst为true 这个代表是第一次注册的registerFirst true;brokerData new BrokerData(clusterName, brokerName, new HashMap());this.brokerAddrTable.put(brokerName, brokerData);}//下面的这些代码应该是设置如果brokerAddrMap中没有这个brokerId对应的地址那么就添加这个地址//有的话 就要进行判断地址是否正确不正确的就要修改MapLong, 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 brokerAddrTableIteratorEntryLong, String it brokerAddrsMap.entrySet().iterator();while (it.hasNext()) {EntryLong, 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);if (null ! topicConfigWrapper MixAll.MASTER_ID brokerId) {if (this.isBrokerTopicConfigChanged(brokerAddr, topicConfigWrapper.getDataVersion())|| registerFirst) {ConcurrentMapString, TopicConfig tcTable topicConfigWrapper.getTopicConfigTable();if (tcTable ! null) {for (Map.EntryString, TopicConfig entry : tcTable.entrySet()) {this.createAndUpdateQueueData(brokerName, entry.getValue());}}}}//这里就是放入Broker中心跳信息//每次心跳请求过来之后都要进行重新创建一个BrokerLiveInfo对象 对上一次心跳的信息进行覆盖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);}//下面的代码感觉跟心跳的关联关系不大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);}}}} finally {this.lock.writeLock().unlock();}} catch (Exception e) {log.error(registerBroker Exception, e);}return result;}
3.NameSever如何进行故障感知
经过前面的代码我们可以看出来每隔30s进行发送注册请求的时候都会针对RouteInfoManager中的brokerLiveTable对上一次已经发送的注册请求进行覆盖。
我们紧接着看下NameServer是如何进行故障感知的我们直接看NameServer中的initialize方法可以直接找到这个代码
his.scheduledExecutorService.scheduleAtFixedRate(NamesrvController.this.routeInfoManager::scanNotActiveBroker, 5, 10, TimeUnit.SECONDS);这行代码的主要作用就就是启动一个定时调度线程每隔10s进行扫描一次RouteInfoManager的scanNotActiveBroker方法我们接下来看下这个方法 public int scanNotActiveBroker() {// 这块的方法主要是brokerLiveTable的集合中的所有元素//拿到broker最新一次的心跳时间 //broker的最新一次心跳时间120s 小于 当前时间戳 //就把这个broker进行移除掉int removeCount 0;IteratorEntryString, BrokerLiveInfo it this.brokerLiveTable.entrySet().iterator();while (it.hasNext()) {EntryString, 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;}
4.总结
我们还是用一张图来进行总结这篇博客的内容
