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結束。

 

說在最後

本次解析僅表明我的觀點，僅供參考。

 

加入技術微信羣

釘釘技術羣