rocketmq源码解析默认请求处理器③

说在前面

默认请求处理器，取消broker注册

 

源码解析

进入这个方法，org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor#unregisterBroker

public RemotingCommand unregisterBroker(ChannelHandlerContext ctx,RemotingCommand request) throws RemotingCommandException {final RemotingCommand response = RemotingCommand.createResponseCommand(null);final UnRegisterBrokerRequestHeader requestHeader =(UnRegisterBrokerRequestHeader) request.decodeCommandCustomHeader(UnRegisterBrokerRequestHeader.class);//        =》this.namesrvController.getRouteInfoManager().unregisterBroker(requestHeader.getClusterName(),requestHeader.getBrokerAddr(),requestHeader.getBrokerName(),requestHeader.getBrokerId());response.setCode(ResponseCode.SUCCESS);response.setRemark(null);return response;    }

进入这个方法，org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#unregisterBroker

public void unregisterBroker(final String clusterName,final String brokerAddr,final String brokerName,final long brokerId) {try {try {this.lock.writeLock().lockInterruptibly();//                删除存活的broker信息BrokerLiveInfo brokerLiveInfo = this.brokerLiveTable.remove(brokerAddr);log.info("unregisterBroker, remove from brokerLiveTable {}, {}",brokerLiveInfo != null ? "OK" : "Failed",brokerAddr);//                删除过滤的broker列表this.filterServerTable.remove(brokerAddr);boolean removeBrokerName = false;//                获取broker数据BrokerData brokerData = this.brokerAddrTable.get(brokerName);if (null != brokerData) {//                    从broker数据中删除brokerString addr = brokerData.getBrokerAddrs().remove(brokerId);log.info("unregisterBroker, remove addr from brokerAddrTable {}, {}",addr != null ? "OK" : "Failed",brokerAddr);if (brokerData.getBrokerAddrs().isEmpty()) {//                        删除broker数据this.brokerAddrTable.remove(brokerName);log.info("unregisterBroker, remove name from brokerAddrTable OK, {}",brokerName);removeBrokerName = true;}}if (removeBrokerName) {//                    获取broker集群Set<String> nameSet = this.clusterAddrTable.get(clusterName);if (nameSet != null) {boolean removed = nameSet.remove(brokerName);log.info("unregisterBroker, remove name from clusterAddrTable {}, {}",removed ? "OK" : "Failed",brokerName);if (nameSet.isEmpty()) {this.clusterAddrTable.remove(clusterName);log.info("unregisterBroker, remove cluster from clusterAddrTable {}",clusterName);}}//                    删除broker的topic=》this.removeTopicByBrokerName(brokerName);}} finally {this.lock.writeLock().unlock();}} catch (Exception e) {log.error("unregisterBroker Exception", e);}    }

进入这个方法，删除broker的topic，org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#removeTopicByBrokerName

private void removeTopicByBrokerName(final String brokerName) {Iterator<Entry<String, List<QueueData>>> itMap = this.topicQueueTable.entrySet().iterator();while (itMap.hasNext()) {Entry<String, List<QueueData>> entry = itMap.next();String topic = entry.getKey();List<QueueData> queueDataList = entry.getValue();Iterator<QueueData> it = queueDataList.iterator();while (it.hasNext()) {QueueData qd = it.next();//                获取broker的消息队列并删除if (qd.getBrokerName().equals(brokerName)) {log.info("removeTopicByBrokerName, remove one broker's topic {} {}", topic, qd);it.remove();}}//            若是topic的消息队列为空，删除topic的消息队列if (queueDataList.isEmpty()) {log.info("removeTopicByBrokerName, remove the topic all queue {}", topic);itMap.remove();}}    }

返回到这个方法，org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor#unregisterBroker结束。

 

说在最后

本次解析仅表明我的观点，仅供参考。

 

加入技术微信群

钉钉技术群