【RocketMQ】Broker服务注册

Posted on

Broker服务注册

BrokerOuterAPI

服务注册

Broker服务注册的详细代码在BrokerOuterAPI的registerBrokerAll方法中：

1.构建Request请求头和请求体，并设置相关的参数；

2.因为Broker会向所有的NameServer注册，所以构建CountDownLatch，用来等待向所有的NameServer注册的任务都执行完毕；

3.使用线程池执行向每一个NameServer注册的任务，底层是通过Netty Client向NameServer发送注册请求的；

4.将注册结果放入结果集中，等待所有NameServer执行完毕返回结果列表；

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
// 注册的详细代码
   public List<RegisterBrokerResult> registerBrokerAll(
       final String clusterName,
       final String brokerAddr,
       final String brokerName,
       final long brokerId,
       final String haServerAddr,
       final TopicConfigSerializeWrapper topicConfigWrapper,
       final List<String> filterServerList,
       final boolean oneway,
       final int timeoutMills,
       final boolean compressed) {
       // 创建一个列表保存注册结果
       final List<RegisterBrokerResult> registerBrokerResultList = Lists.newArrayList();
       // 获取NameServer地址列表
       List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
       if (nameServerAddressList != null && nameServerAddressList.size() > 0) {
           // 构建Request请求头
           final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();
           // 设置broker地址
           requestHeader.setBrokerAddr(brokerAddr);
           // 设置broker id
           requestHeader.setBrokerId(brokerId);
           // 设置broker name
           requestHeader.setBrokerName(brokerName);
           // 设置集群名称
           requestHeader.setClusterName(clusterName);
           requestHeader.setHaServerAddr(haServerAddr);
           requestHeader.setCompressed(compressed);
           // 创建请求体
           RegisterBrokerBody requestBody = new RegisterBrokerBody();
           requestBody.setTopicConfigSerializeWrapper(topicConfigWrapper);
           requestBody.setFilterServerList(filterServerList);
           final byte[] body = requestBody.encode(compressed);
           final int bodyCrc32 = UtilAll.crc32(body);
           requestHeader.setBodyCrc32(bodyCrc32);
           // 构建了一个CountDownLatch
           final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());
           for (final String namesrvAddr : nameServerAddressList) {
               // 使用线程池执行注册任务
               brokerOuterExecutor.execute(new Runnable() {
                   @Override
                   public 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();
                       }
                   }
               });
           }

           try {
               // 等待Broker对所有的NameServer注册完毕
               countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS);
           } catch (InterruptedException e) {
           }
       }

       return registerBrokerResultList;
   }

服务注册请求的发送

进入到registerBroker看一下服务注册请求发送的详细代码，可以看到底层是使用Netty Client进行请求发送的：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
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，封装请求头和请求体
     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;
     }
     // 使用Netty Client发送网络请求
     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());
 }
NettyRemotingClient

进入NettyRemotingClient中的invokeSync方法，看下详细的发送过程：

1.获取一个Channel，也就是获取和NameServer的一个连接，在获取Channel的时候首先会从缓存中获取Channel，如果没有获取到才会创建新的Channel；

2.调用invokeSyncImpl方法，通过Channel向NameServer发送请求；

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
  @Override
    public RemotingCommand invokeSync(String addr, final RemotingCommand request, long timeoutMillis)
        throws InterruptedException, RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException {
        // 获取当前时间
        long beginStartTime = System.currentTimeMillis();
        // 获取Channel,也就是获取了一个和NameServer的一个连接
        final Channel channel = this.getAndCreateChannel(addr);
        if (channel != null && channel.isActive()) {
            try {
                doBeforeRpcHooks(addr, request);
                long costTime = System.currentTimeMillis() - beginStartTime;
                if (timeoutMillis < costTime) {
                    throw new RemotingTimeoutException("invokeSync call timeout");
                }
                // 真正发送请求的地方
                RemotingCommand response = this.invokeSyncImpl(channel, request, timeoutMillis - costTime);
                doAfterRpcHooks(RemotingHelper.parseChannelRemoteAddr(channel), request, response);
                return response;
            } catch (RemotingSendRequestException e) {
                log.warn("invokeSync: send request exception, so close the channel[{}]", addr);
                this.closeChannel(addr, channel);
                throw e;
            } catch (RemotingTimeoutException e) {
                if (nettyClientConfig.isClientCloseSocketIfTimeout()) {
                    this.closeChannel(addr, channel);
                    log.warn("invokeSync: close socket because of timeout, {}ms, {}", timeoutMillis, addr);
                }
                log.warn("invokeSync: wait response timeout exception, the channel[{}]", addr);
                throw e;
            }
        } else {
            this.closeChannel(addr, channel);
            throw new RemotingConnectException(addr);
        }
    }
 
// 获取Channel
    private Channel getAndCreateChannel(final String addr) throws RemotingConnectException, InterruptedException {
        if (null == addr) {
            // 获取Channel
            return getAndCreateNameserverChannel();
        }
        // 从缓存中获取channel
        ChannelWrapper cw = this.channelTables.get(addr);
        if (cw != null && cw.isOK()) {
            return cw.getChannel();
        }
        // 创建channel
        return this.createChannel(addr);
    }

创建Channel的代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
// 创建Channel
    private Channel createChannel(final String addr) throws InterruptedException {
        // 先从缓存中获取连接
        ChannelWrapper cw = this.channelTables.get(addr);
        if (cw != null && cw.isOK()) {
            return cw.getChannel();
        }
        // 加锁
        if (this.lockChannelTables.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
            try {
                // 同样是从缓存中获取连接
                boolean createNewConnection;
                cw = this.channelTables.get(addr);
                if (cw != null) {

                    if (cw.isOK()) {
                        return cw.getChannel();
                    } else if (!cw.getChannelFuture().isDone()) {
                        createNewConnection = false;
                    } else {
                        this.channelTables.remove(addr);
                        createNewConnection = true;
                    }
                } else {
                    createNewConnection = true;
                }

                if (createNewConnection) {
                    // 创建连接的地方
                    ChannelFuture channelFuture = this.bootstrap.connect(RemotingHelper.string2SocketAddress(addr));
                    log.info("createChannel: begin to connect remote host[{}] asynchronously", addr);
                    cw = new ChannelWrapper(channelFuture);
                    // 将连接放入缓存
                    this.channelTables.put(addr, cw);
                }
            } catch (Exception e) {
                log.error("createChannel: create channel exception", e);
            } finally {
                this.lockChannelTables.unlock();
            }
        } else {
            log.warn("createChannel: try to lock channel table, but timeout, {}ms", LOCK_TIMEOUT_MILLIS);
        }
      // 先不管
        if (cw != null) {
            ChannelFuture channelFuture = cw.getChannelFuture();
            if (channelFuture.awaitUninterruptibly(this.nettyClientConfig.getConnectTimeoutMillis())) {
                if (cw.isOK()) {
                    log.info("createChannel: connect remote host[{}] success, {}", addr, channelFuture.toString());
                    return cw.getChannel();
                } else {
                    log.warn("createChannel: connect remote host[" + addr + "] failed, " + channelFuture.toString(), channelFuture.cause());
                }
            } else {
                log.warn("createChannel: connect remote host[{}] timeout {}ms, {}", addr, this.nettyClientConfig.getConnectTimeoutMillis(),
                    channelFuture.toString());
            }
        }

        return null;
    }

发送网络请求的代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
// 发送网络请求的代码
   public RemotingCommand invokeSyncImpl(final Channel channel, final RemotingCommand request,
       final long timeoutMillis)
       throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException {
       final int opaque = request.getOpaque();

       try {
           final ResponseFuture responseFuture = new ResponseFuture(channel, opaque, timeoutMillis, null, null);
           this.responseTable.put(opaque, responseFuture);
           final SocketAddress addr = channel.remoteAddress();
           channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
               @Override
               public void operationComplete(ChannelFuture f) throws Exception {
                   if (f.isSuccess()) {
                       responseFuture.setSendRequestOK(true);
                       return;
                   } else {
                       responseFuture.setSendRequestOK(false);
                   }

                   responseTable.remove(opaque);
                   responseFuture.setCause(f.cause());
                   responseFuture.putResponse(null);
                   log.warn("send a request command to channel <" + addr + "> failed.");
               }
           });

           RemotingCommand responseCommand = responseFuture.waitResponse(timeoutMillis);
           if (null == responseCommand) {
               if (responseFuture.isSendRequestOK()) {
                   throw new RemotingTimeoutException(RemotingHelper.parseSocketAddressAddr(addr), timeoutMillis,
                       responseFuture.getCause());
               } else {
                   throw new RemotingSendRequestException(RemotingHelper.parseSocketAddressAddr(addr), responseFuture.getCause());
               }
           }

           return responseCommand;
       } finally {
           this.responseTable.remove(opaque);
       }
   }

心跳机制

为了让NameServer感知到Broker是否有异常，Broker会定时发送心跳到NameServer，告诉NameServer自己的服务正常。

BrokerController

回到start方法中，可以看到注册了一个定时任务，定时会向NameServer进行注册，默认每30s进行一次注册：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public void start() throws Exception {
    ...
    // 定时向NameServer注册
    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

            @Override
            public void run() {
                try {
                    // 向NameServer注册
                    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);
    ...
}

NameServer注册处理

Broker的注册请求处理

NamesrvController

回到NamesrvController的initialize方法中，有一个注册处理器的方法registerProcessor：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public boolean initialize() {
    ...
    // 注册请求处理器
    this.registerProcessor();
    ...
}
private void registerProcessor() {
    // 是否是集群测试
    if (namesrvConfig.isClusterTest()) {
        this.remotingServer.registerDefaultProcessor(new ClusterTestRequestProcessor(this, namesrvConfig.getProductEnvName()),
        this.remotingExecutor);
    } else {
         // 注册默认的处理器,创建了一个DefaultRequestProcessor
        this.remotingServer.registerDefaultProcessor(new DefaultRequestProcessor(this), this.remotingExecutor);
     }
 }

DefaultRequestProcessor

DefaultRequestProcessor的processRequest对请求进行处理，会判断请求类型，然后处理对应的请求，找到Broker的注册类型，可以看到调用registerBroker方法对Broker的请求进行处理，最终是调用RouteInfoManager的registerBroker进行注册的：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
public class DefaultRequestProcessor{
   @Override
    public RemotingCommand processRequest(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {

        if (ctx != null) {
            log.debug("receive request, {} {} {}",
                request.getCode(),
                RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
                request);
        }
        // 判断请求类型，处理不同的请求
        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);
            case RequestCode.REGISTER_BROKER: // 如果是Broker的注册请求
                Version brokerVersion = MQVersion.value2Version(request.getVersion());
                // 版本判断
                if (brokerVersion.ordinal() >= MQVersion.Version.V3_0_11.ordinal()) {
                    return this.registerBrokerWithFilterServer(ctx, request);
                } else {
                    // Broker注册
                    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.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;
        }
        return null;
    }
    // Broker的注册
    public RemotingCommand registerBroker(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        // 构造请求响应
        final RemotingCommand response = RemotingCommand.createResponseCommand(RegisterBrokerResponseHeader.class);
        final RegisterBrokerResponseHeader responseHeader = (RegisterBrokerResponseHeader) response.readCustomHeader();
        // 获取请求头
        final RegisterBrokerRequestHeader requestHeader =
            (RegisterBrokerRequestHeader) request.decodeCommandCustomHeader(RegisterBrokerRequestHeader.class);

        if (!checksum(ctx, request, requestHeader)) {
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark("crc32 not match");
            return response;
        }

        TopicConfigSerializeWrapper topicConfigWrapper;
        if (request.getBody() != null) {
            topicConfigWrapper = TopicConfigSerializeWrapper.decode(request.getBody(), TopicConfigSerializeWrapper.class);
        } else {
            topicConfigWrapper = new TopicConfigSerializeWrapper();
            topicConfigWrapper.getDataVersion().setCounter(new AtomicLong(0));
            topicConfigWrapper.getDataVersion().setTimestamp(0);
        }
        // 这里调用RouteInfoManager的registerBroker对Broker进行注册，可以看到从请求中获取了Broker的信息作为参数传入
        RegisterBrokerResult result = this.namesrvController.getRouteInfoManager().registerBroker(
            requestHeader.getClusterName(),
            requestHeader.getBrokerAddr(),
            requestHeader.getBrokerName(),
            requestHeader.getBrokerId(),
            requestHeader.getHaServerAddr(),
            topicConfigWrapper,
            null,
            ctx.channel()
        );
        // 设置响应头
        responseHeader.setHaServerAddr(result.getHaServerAddr());
        responseHeader.setMasterAddr(result.getMasterAddr());

        byte[] jsonValue = this.namesrvController.getKvConfigManager().getKVListByNamespace(NamesrvUtil.NAMESPACE_ORDER_TOPIC_CONFIG);
        response.setBody(jsonValue);
        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        // 返回响应
        return response;
    }

}

RouteInfoManager

RouteInfoManager的registerBroker对Broker进行了注册：

1.根据集群名称获取对应的broker集合，然后将当前的broker加入集合中；

2.根据Broker的name获取brokerAddrsMap，其中key为broker的id，value为IP+端口，然后处理重复数据，保证brokerAddrTable核心路由表同一个IP和端口只有一条记录；

3.判断是否是Master，如果是并且Topic发生了变化或者是Broker首次注册，将会创建或者更新当前broker的Topic队列信息；

4.记录当前Broker的LiveInfo，Broker每次定时发送请求作为心跳的时候，都会有一个新的BrokerLiveInfo记录Broker的注册信息，包括注册时间、连接channel等信息，后续可以用来感知Broker是否有异常；

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
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();
               // 根据集群名称获取对应的broker集合，这里用set集合，当Broker发送心跳机制的时候集合不会有重复数据
               Set<String> brokerNames = this.clusterAddrTable.get(clusterName);
               if (null == brokerNames) {
                   brokerNames = new HashSet<String>();
                   this.clusterAddrTable.put(clusterName, brokerNames);
               }
               // 向集合中添加当前的broker
               brokerNames.add(brokerName);

               boolean registerFirst = false;
               // 根据Broker name获取BrokerData
               BrokerData brokerData = this.brokerAddrTable.get(brokerName);
               // 首次注册的时候为空
               if (null == brokerData) {
                   registerFirst = true;
                   // 如果获取为空，构造BrokerData加入brokerAddrTable中
                   brokerData = new BrokerData(clusterName, brokerName, new HashMap<Long, String>());
                   this.brokerAddrTable.put(brokerName, brokerData);
               }
               // 根据broker name获取所有broker的地址，key为brokerId，value为IP:PROT
               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 brokerAddrTable
               // 这里处理重复数据，保证同一个IP和端口只在brokerAddrTable有一条记录
               Iterator<Entry<Long, String>> it = brokerAddrsMap.entrySet().iterator();
               while (it.hasNext()) {
                   Entry<Long, String> item = it.next();
                   // 如果IP和端口一致，但是brokerID不一致，移除旧的记录
                   if (null != brokerAddr && brokerAddr.equals(item.getValue()) && brokerId != item.getKey()) {
                       it.remove();
                   }
               }
               // 加入当前的broker信息
               String oldAddr = brokerData.getBrokerAddrs().put(brokerId, brokerAddr);
               // 判断是否是首次注册
               registerFirst = registerFirst || (null == oldAddr);
               // 如果topicConfigWrapper不为空，并且当前的broker是master
               if (null != topicConfigWrapper
                   && MixAll.MASTER_ID == brokerId) {
                   // 如果Topic有变化或者是首次注册
                   if (this.isBrokerTopicConfigChanged(brokerAddr, topicConfigWrapper.getDataVersion())
                       || registerFirst) {
                       // 获取当前broker的所有的Topic
                       ConcurrentMap<String, TopicConfig> tcTable =
                           topicConfigWrapper.getTopicConfigTable();
                       if (tcTable != null) {
                           // 遍历topic
                           for (Map.Entry<String, TopicConfig> entry : tcTable.entrySet()) {
                               // 创建或者更新当前broker的Topic队列信息
                               this.createAndUpdateQueueData(brokerName, entry.getValue());
                           }
                       }
                   }
               }
               // 构建BrokerLiveInfo,记录Broker的LiveInfo，当Broker定时发送请求作为心跳的时候，都会有一个新的BrokerLiveInfo记录Broker的注册信息，包括注册时间、连接channel等信息
               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);
                   }
               }
               // 如果不是master 
               if (MixAll.MASTER_ID != brokerId) {
                   // 获取master信息
                   String masterAddr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
                   if (masterAddr != null) {
                       // 获取master的LiveInfo
                       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;
   }

异常感知

在NamesrvController的初始化方法中启动了一个定时任务，调用了routeInfoManager的scanNotActiveBroker方法扫描不活跃的Broker，默认10s扫描一次：

1
2
3
4
5
6
7
8
9
10
11
12
13
public class NamesrvController {
      public boolean initialize() {
        ...
        // 定时任务线程池，启动了routeInfoManager的扫描不活跃的Broker方法
        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

            @Override
            public void run() {
                NamesrvController.this.routeInfoManager.scanNotActiveBroker();
            }
        }, 5, 10, TimeUnit.SECONDS);
        ...
}

RouteInfoManager

RouteInfoManager的scanNotActiveBroker会扫描超时未发送心跳的Broker，默认是120s，如果超过120s没有收到心跳注册，就会移除Broker的路由信息：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
// 扫描不活跃的Broker
public void scanNotActiveBroker() {
        // 遍历所有的BrokerLiveInfo
        Iterator<Entry<String, BrokerLiveInfo>> it = this.brokerLiveTable.entrySet().iterator();
        while (it.hasNext()) {
            Entry<String, BrokerLiveInfo> next = it.next();
            // 获取最后一次注册的时间戳
            long last = next.getValue().getLastUpdateTimestamp();
            // 判断是否超过设定的时间，默认是120s
            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);
                // 将移除Broker的路由信息
                this.onChannelDestroy(next.getKey(), next.getValue().getChannel());
            }
        }
    }

注：图片来自于儒猿技术窝-从 0 开始带你成为消息中间件实战高手

参考

儒猿技术窝：从 0 开始带你成为消息中间件实战高手