java开发RocketMQ之NameServer路由管理源码分析

目录
  • 1.前言
  • 2.路由元信息
  • 3.路由注册
    • 3.1Broker路由注册
    • 3.2NameServer处理路由注册
    • 3.3路由删除
      • 3.3.1Broker异常关闭
      • 3.3.2Broker正常关闭
    • 3.4路由发现
  • 3.5总结

1.前言

NameServer主要作用是为消息消费者和消息生产者提供关于主题Topic的路由信息,那么NameServer需要存储路由的基本信息,还要管理Broker节点,包括路由注册、路由删除等。

2.路由元信息

路由元信息主要由RouteInfoManager来进行管理,这个类在上一篇文章中已经介绍过,这里再做一下简单的介绍。

private final static long BROKER_CHANNEL_EXPIRED_TIME = 1000 * 60 * 2;
private final ReadWriteLock lock = new ReentrantReadWriteLock();
private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;

BROKER_CHANNEL_EXPIRED_TIME:NameServer与Broker空闲连接时长,在2minNameServer之内没有收到Broker的心跳包,则NameServer会关闭与该Broker的连接并删除Broker的路由信息。

lock:读写锁,用来保护以下用于存储关键信息的非线程安全容器HashMap。

topicQueueTable:Topic消息队列路由信息,消息发送时根据路由表进行负载均衡

brokerAddrTable:Broker基础信息,包括brokerName、所属集群名称、主备Broker地址

clusterAddrTable:Broker集群信息,存储集群中所有Broker名称

brokerLiveTable:Broker状态信息,NameServer每次收到心跳包会对该表进行信息维护。

filterServerTable:Broker上的FilterServer列表,用于类模式消息过滤。

RocketMQ基于定于发布机制,一个Topic拥有多个消息队列,一个Broker为每一个主题创建4个读队列和4个写队列。多个Broker组成一个集群,集群由相同的多台Broker组成Master-Slave架构,brokerId为0代表Master,大于0为Slave。BrokerLiveInfo中的lastUpdateTimestamp存储上次收到Broker心跳包的时间。

3.路由注册

以下是路由注册的时序图:

RocketMQ的路由注册功能是通过Broker与NameServer的心跳机制实现的。Broker启动时向NameServer集群发送心跳信息,每隔30s向集群所有NameServer发送心跳包,NameServer接收到之后会对brokerLiveTable中的BrokerLiveInfo的lastUpdateTimeStamp属性值进行更新。此外NameServer会每隔10s扫描brokerLiveTable,如果检测到有Broker连续120s没有发送心跳包,则NameServer将移除Broker的路由信息同时关闭Socket连接。

源码

3.1Broker路由注册

BrokerController#start

//...进行检查
if (!messageStoreConfig.isEnableDLegerCommitLog()) {
        startProcessorByHa(messageStoreConfig.getBrokerRole());
        handleSlaveSynchronize(messageStoreConfig.getBrokerRole());
        //注册Broker信息
        this.registerBrokerAll(true, false, true);
    }
    //创建定时任务线程池
    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
            try {
                //每隔三十秒注册Broker信息
                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);
    if (this.brokerStatsManager != null) {
        this.brokerStatsManager.start();
    }

    if (this.brokerFastFailure != null) {
        this.brokerFastFailure.start();
    }
}

BrokerController#registerBrokerAll

if (forceRegister || needRegister(this.brokerConfig.getBrokerClusterName(),
    this.getBrokerAddr(),
    this.brokerConfig.getBrokerName(),
    this.brokerConfig.getBrokerId(),
    this.brokerConfig.getRegisterBrokerTimeoutMills())) {
    //开始注册
    doRegisterBrokerAll(checkOrderConfig, oneway, topicConfigWrapper);
}

BrokerController#doRegisterBrokerAll

//进入brokerOuterAPI#registerBrokerAll
List<RegisterBrokerResult> registerBrokerResultList = this.brokerOuterAPI.registerBrokerAll(
    this.brokerConfig.getBrokerClusterName(),
    this.getBrokerAddr(),
    this.brokerConfig.getBrokerName(),
    this.brokerConfig.getBrokerId(),
    this.getHAServerAddr(),
    topicConfigWrapper,
    this.filterServerManager.buildNewFilterServerList(),
    oneway,
    this.brokerConfig.getRegisterBrokerTimeoutMills(),
    this.brokerConfig.isCompressedRegister());

BrokerOuterAPI#registerBrokerAll

final List<RegisterBrokerResult> registerBrokerResultList = new CopyOnWriteArrayList<>();
//获取nameServer地址信息
List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
if (nameServerAddressList != null && nameServerAddressList.size() > 0) {
    //进行请求头的封装
    final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();
    requestHeader.setBrokerAddr(brokerAddr);
    requestHeader.setBrokerId(brokerId);
    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);
    final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());
    //遍历nameServer地址 --->  分别向各个nameServer注册Broker
    for (final String namesrvAddr : nameServerAddressList) {
        brokerOuterExecutor.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    //向nameServer注册Broker并返回注册结果
                    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 {
        countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS);
    } catch (InterruptedException e) {
    }
}

return registerBrokerResultList;

BrokerOuterAPI#registerBroker

//如果是oneway发送 无需关注返回结果 直接返回null
if (oneway) {
    try {
        this.remotingClient.invokeOneway(namesrvAddr, request, timeoutMills);
    } catch (RemotingTooMuchRequestException e) {
        // Ignore
        log.error("RemotingTooMuchRequestException!!!");
    }
    return null;
}
//返回远程调用的响应结果
RemotingCommand response = this.remotingClient.invokeSync(namesrvAddr, request, timeoutMills);

3.2NameServer处理路由注册

以下是NameServer处理路由注册的时序图:

请求发送至NameServer,会由默认的网络处理解析类DefaultRequestProcessor进行处理解析,如果请求类型是REGISTER_BROKER则会将请求转发至RouteInfoManager进行路由的注册。

源码

DefaultRequestProcessor#processRequest

switch (request.getCode()) {
    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);
        }
}

DefaultRequestProcessor#registerBrokerWithFilterServer

//转发到NameServerController中的RouteInfoManager进行路由注册
RegisterBrokerResult result = this.namesrvController.getRouteInfoManager().registerBroker(
    requestHeader.getClusterName(),
    requestHeader.getBrokerAddr(),
    requestHeader.getBrokerName(),
    requestHeader.getBrokerId(),
    requestHeader.getHaServerAddr(),
    registerBrokerBody.getTopicConfigSerializeWrapper(),
    registerBrokerBody.getFilterServerList(),
    ctx.channel());

RouteInfoManager#registerBroker

//加入写锁-保护HashTable线程安全
this.lock.writeLock().lockInterruptibly();
//维护clusterAddrTable
Set<String> brokerNames = this.clusterAddrTable.get(clusterName);
if (null == brokerNames) {
    brokerNames = new HashSet<String>();
    this.clusterAddrTable.put(clusterName, brokerNames);
}
brokerNames.add(brokerName);
//维护brokerAddrTable
//是否是第一次注册 默认为false
boolean registerFirst = false;
BrokerData brokerData = this.brokerAddrTable.get(brokerName);
//如果在brokerAddrTable没有该Broker信息
if (null == brokerData) {
    //设置为第一次注册
    registerFirst = true;
    //创建对应的BrokerData信息
    brokerData = new BrokerData(clusterName, brokerName, new HashMap<Long, String>());
    this.brokerAddrTable.put(brokerName, brokerData);
}
//不是第一次注册-对Broker信息进行更新
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
Iterator<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()) {
        it.remove();
    }
}

String oldAddr = brokerData.getBrokerAddrs().put(brokerId, brokerAddr);
registerFirst = registerFirst || (null == oldAddr);
//维护topicQueueTable
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());
            }
        }
    }
}
//维护brokerLiveTable
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);
}
//维护filterServerTable
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);
        }
    }
}

RouteInfoManager#createAndUpdateQueueData

//创建并封装新的队列信息
QueueData queueData = new QueueData();
//Broker名称
queueData.setBrokerName(brokerName);
//读写队列个数
queueData.setWriteQueueNums(topicConfig.getWriteQueueNums());
queueData.setReadQueueNums(topicConfig.getReadQueueNums());
//操作权限
queueData.setPerm(topicConfig.getPerm());
//同步复制还是异步复制的标识
queueData.setTopicSysFlag(topicConfig.getTopicSysFlag());

//根据Topic从topicQueueTable中获取队列集合
List<QueueData> queueDataList = this.topicQueueTable.get(topicConfig.getTopicName());
//如果没有获取到
if (null == queueDataList) {
    //创建并将新的queueData添加到集合中
    queueDataList = new LinkedList<QueueData>();
    queueDataList.add(queueData);
    //放入topicQueueTable
    this.topicQueueTable.put(topicConfig.getTopicName(), queueDataList);
    log.info("new topic registered, {} {}", topicConfig.getTopicName(), queueData);
}
//queueDataList不为空
else {
    boolean addNewOne = true;

    Iterator<QueueData> it = queueDataList.iterator();
    //对该主题已有队列信息进行遍历
    while (it.hasNext()) {
        QueueData qd = it.next();
        //如果找到相同主题
        if (qd.getBrokerName().equals(brokerName)) {
            //已经存在
            if (qd.equals(queueData)) {
                addNewOne = false;
            } else {
                //进行更改-先删除后新增
                log.info("topic changed, {} OLD: {} NEW: {}", topicConfig.getTopicName(), qd,
                    queueData);
                it.remove();
            }
        }
    }
	//添加一个新的队列信息
    if (addNewOne) {
        queueDataList.add(queueData);
    }
}

3.3路由删除

NameServer会每隔10s扫描brokerLiveTable状态表,如果BrokerLivelastUpdateTimestamp的时间戳距当前时间超过120s,则认为Broker失效,移除该Broker,关闭与Broker连接,同时更新topicQueueTablebrokerAddrTablebrokerLiveTablefilterServerTable

删除路由信息的两个触发条件:

  • NameServer定期扫描brokerLiveTable检测上次心跳包发送时间与当前系统的时间差,如果时间超过120s,则需要移除该Broker路由信息
  • Broker正常关闭,执行unregisterBroker

3.3.1Broker异常关闭

NamesrvController#initialize

//定时任务线程池--->每隔十秒扫描活跃状态异常的Broker信息
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    /**
     * 对Not Active Broker 进行扫描
     */
    @Override
    public void run() {
        //在routeInfoManager中进行路由删除
        NamesrvController.this.routeInfoManager.scanNotActiveBroker();
    }
}, 5, 10, TimeUnit.SECONDS);

RouteInfoManager#scanNotActiveBroker

//遍历brokerLiveTable 查看是否有broker状态长时间没有更新
Iterator<Entry<String, BrokerLiveInfo>> it = this.brokerLiveTable.entrySet().iterator();
//遍历brokerLiveTable
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());
        //移除broker
        it.remove();
        log.warn("The broker channel expired, {} {}ms", next.getKey(), BROKER_CHANNEL_EXPIRED_TIME);
        //维护路由表
        this.onChannelDestroy(next.getKey(), next.getValue().getChannel());
    }
}

RouteInfoManager#onChannelDestroy

//待删除的BrokerAddress
String brokerAddrFound = null;
if (channel != null) {
    try {
        try {
            //加入读锁
            this.lock.readLock().lockInterruptibly();
            Iterator<Entry<String, BrokerLiveInfo>> itBrokerLiveTable =
                this.brokerLiveTable.entrySet().iterator();
            //遍历查找待删除的Broker
            while (itBrokerLiveTable.hasNext()) {
                Entry<String, BrokerLiveInfo> entry = itBrokerLiveTable.next();
                if (entry.getValue().getChannel() == channel) {
                    brokerAddrFound = entry.getKey();
                    break;
                }
            }
        } finally {
            //释放锁
            this.lock.readLock().unlock();
        }
    } catch (Exception e) {
        log.error("onChannelDestroy Exception", e);
    }
}
//加入写锁
//根据brokerAddrFound从brokerLiveTable和filterServerTable移除对应的Broker信息
this.lock.writeLock().lockInterruptibly();
this.brokerLiveTable.remove(brokerAddrFound);
this.filterServerTable.remove(brokerAddrFound);
//维护brokerAddrTable
String brokerNameFound = null;
boolean removeBrokerName = false;
Iterator<Entry<String, BrokerData>> itBrokerAddrTable =
    this.brokerAddrTable.entrySet().iterator();
//遍历brokerAddrTable
while (itBrokerAddrTable.hasNext() && (null == brokerNameFound)) {
    BrokerData brokerData = itBrokerAddrTable.next().getValue();
	//遍历brokerAddress
    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();
        //遍历到匹配的broker
        if (brokerAddr.equals(brokerAddrFound)) {
            brokerNameFound = brokerData.getBrokerName();
            //根据broker地址移除broker
            it.remove();
            log.info("remove brokerAddr[{}, {}] from brokerAddrTable, because channel destroyed",
                brokerId, brokerAddr);
            break;
        }
    }
	//如果当前BrokerData中的Broker列表为空
    if (brokerData.getBrokerAddrs().isEmpty()) {
        removeBrokerName = true;
        //从BrokerAddrTable移除该BrokerData
        itBrokerAddrTable.remove();
        log.info("remove brokerName[{}] from brokerAddrTable, because channel destroyed",
            brokerData.getBrokerName());
    }
}
//维护clusterAddrTable
if (brokerNameFound != null && removeBrokerName) {
    Iterator<Entry<String, Set<String>>> it = this.clusterAddrTable.entrySet().iterator();
    //遍历clusterAddrTable
    while (it.hasNext()) {
        Entry<String, Set<String>> entry = it.next();
        String clusterName = entry.getKey();
        Set<String> brokerNames = entry.getValue();
        //根据brokerName删除cluster集群中的broker并返回是否删除成功
        boolean removed = brokerNames.remove(brokerNameFound);
        if (removed) {
            log.info("remove brokerName[{}], clusterName[{}] from clusterAddrTable, because channel destroyed",
                brokerNameFound, clusterName);

            //如果cluster已经不包含任何broker则移除该集群
            if (brokerNames.isEmpty()) {
                log.info("remove the clusterName[{}] from clusterAddrTable, because channel destroyed and no broker in this cluster",
                    clusterName);
                it.remove();
            }
            break;
        }
    }
}
//维护TopicQueueTable
if (removeBrokerName) {
    Iterator<Entry<String, List<QueueData>>> itTopicQueueTable =
        this.topicQueueTable.entrySet().iterator();
    //遍历TopicQueueTable
    while (itTopicQueueTable.hasNext()) {
        Entry<String, List<QueueData>> entry = itTopicQueueTable.next();
        //主题
        String topic = entry.getKey();
        //主题对应的Broker信息
        List<QueueData> queueDataList = entry.getValue();

        Iterator<QueueData> itQueueData = queueDataList.iterator();
        //遍历Broker
        while (itQueueData.hasNext()) {
            QueueData queueData = itQueueData.next();
            //通过brokerNameFound在queueData中查找对应的Broker信息
            if (queueData.getBrokerName().equals(brokerNameFound)) {
                //移除该Broker
                itQueueData.remove();
                log.info("remove topic[{} {}], from topicQueueTable, because channel destroyed",
                    topic, queueData);
            }
        }

        //如果当前主题已经没有任何Broker,则移除该主题Topic
        if (queueDataList.isEmpty()) {
            itTopicQueueTable.remove();
            log.info("remove topic[{}] all queue, from topicQueueTable, because channel destroyed",
                topic);
        }
    }
}
//释放写锁
finally {
    this.lock.writeLock().unlock();
}

3.3.2Broker正常关闭

关闭入口-BrokerController

//BrokerController#shutdown
this.unregisterBrokerAll();

//BrokerController#unregisterBrokerAll
this.brokerOuterAPI.unregisterBrokerAll(
    this.brokerConfig.getBrokerClusterName(),
    this.getBrokerAddr(),
    this.brokerConfig.getBrokerName(),
    this.brokerConfig.getBrokerId());

BrokerOuterAPI#unregisterBrokerAll

//获取NameServerAddress列表
List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
if (nameServerAddressList != null) {
    //遍历NameServer
    for (String namesrvAddr : nameServerAddressList) {
        try {
            //注销指定Broker
            this.unregisterBroker(namesrvAddr, clusterName, brokerAddr, brokerName, brokerId);
            log.info("unregisterBroker OK, NamesrvAddr: {}", namesrvAddr);
        } catch (Exception e) {
            log.warn("unregisterBroker Exception, {}", namesrvAddr, e);
        }
    }
}

BrokerOuterAPI#unregisterBroker

//封装注销Broker请求头
UnRegisterBrokerRequestHeader requestHeader = new UnRegisterBrokerRequestHeader();
requestHeader.setBrokerAddr(brokerAddr);
requestHeader.setBrokerId(brokerId);
requestHeader.setBrokerName(brokerName);
requestHeader.setClusterName(clusterName);
//封装注销请求
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UNREGISTER_BROKER, requestHeader);
//远程调用并发送请求-返回响应结果
RemotingCommand response = this.remotingClient.invokeSync(namesrvAddr, request, 3000);
assert response != null;
switch (response.getCode()) {
    case ResponseCode.SUCCESS: {
        return;
    }
    default:
        break;
}
throw new MQBrokerException(response.getCode(), response.getRemark(), brokerAddr);

忽略Netty远程调用过程

DefaultRequestProcess#processRequest

//根据请求状态码-进行不同的处理
switch (request.getCode()) {
	case RequestCode.UNREGISTER_BROKER:
        return this.unregisterBroker(ctx, request);
}

DefaultRequestProcess#unregisterBroker

final RemotingCommand response = RemotingCommand.createResponseCommand(null);
//解码注销Broker的请求头
final UnRegisterBrokerRequestHeader requestHeader =
    (UnRegisterBrokerRequestHeader) request.decodeCommandCustomHeader(UnRegisterBrokerRequestHeader.class);
//转发至RouteInfoManager注销Broker
this.namesrvController.getRouteInfoManager().unregisterBroker(
    requestHeader.getClusterName(),
    requestHeader.getBrokerAddr(),
    requestHeader.getBrokerName(),
    requestHeader.getBrokerId());

//设置响应状态码
response.setCode(ResponseCode.SUCCESS);
response.setRemark(null);
//返回响应结果
return response;

RouteInfoManager#unregisterBroker

try {
    try {
        //加入写锁-保护HashMap线程安全
        this.lock.writeLock().lockInterruptibly();
        //由于是从BrokerController远程调用来的请求-故不对BrokerAddr进行检查-直接进行移除
        //移除brokerLiveTable中的Broker信息
        BrokerLiveInfo brokerLiveInfo = this.brokerLiveTable.remove(brokerAddr);
        log.info("unregisterBroker, remove from brokerLiveTable {}, {}",
            brokerLiveInfo != null ? "OK" : "Failed",
            brokerAddr
        );

        //维护filterServerTable
        this.filterServerTable.remove(brokerAddr);
        boolean removeBrokerName = false;
        //根据brokername获取对应的BrokerData信息
        BrokerData brokerData = this.brokerAddrTable.get(brokerName);
        if (null != brokerData) {
            //删除对应Broker信息
            String addr = brokerData.getBrokerAddrs().remove(brokerId);
            log.info("unregisterBroker, remove addr from brokerAddrTable {}, {}",
                addr != null ? "OK" : "Failed",
                brokerAddr
            );
            //如果当前BrokerData中已经没有其他Broker
            if (brokerData.getBrokerAddrs().isEmpty()) {
                //从BrokerAddrTable移除相应的Broker
                this.brokerAddrTable.remove(brokerName);
                log.info("unregisterBroker, remove name from brokerAddrTable OK, {}",
                    brokerName
                );
                removeBrokerName = true;
            }
        }

        if (removeBrokerName) {
            //在clusterAddrTable中通过集群名称获取对应的Broker集合
            Set<String> nameSet = this.clusterAddrTable.get(clusterName);
            if (nameSet != null) {
                //移除该集合中的Broker
                boolean removed = nameSet.remove(brokerName);
                log.info("unregisterBroker, remove name from clusterAddrTable {}, {}",
                    removed ? "OK" : "Failed",
                    brokerName);
                //如果该集群中的Broker集合为空-移除该集群
                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);
}

3.4路由发现

RocketMQ路由发现是非实时的,当Topic路由出现变化后,NameServer不会主动推送给客户端,而是由客户端定时拉取主题最新的路由。

DefaultRequestProcessor#processRequest

switch (request.getCode()) {
    case RequestCode.GET_ROUTEINFO_BY_TOPIC:
         return this.getRouteInfoByTopic(ctx, request);
}

DefaultRequestProcessor#getRouteInfoByTopic

//远程调用返回的响应结果
final RemotingCommand response = RemotingCommand.createResponseCommand(null);
//解析获取路由信息的请求头
final GetRouteInfoRequestHeader requestHeader =
    (GetRouteInfoRequestHeader) request.decodeCommandCustomHeader(GetRouteInfoRequestHeader.class);
//调用RouteInfoManager中的方法,从路由表topicQueueTable、brokerAddrTable、filterServerTable中
//分别填充TopicRouteData的List<QueueData>、List<BrokerData>、filterServer
TopicRouteData topicRouteData = this.namesrvController.getRouteInfoManager().pickupTopicRouteData(requestHeader.getTopic());

//如果找到主题对应你的路由信息并且该主题为顺序消息
//则从NameServer KVConfig中获取关于顺序消息相关的配置填充路由信息
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;

RouteInfoManager#pickupTopicRouteData

//创建主题路由信息对象
TopicRouteData topicRouteData = new TopicRouteData();
//是否找到对应队列的标识
boolean foundQueueData = false;
//是否找到对应Broker的标识
boolean foundBrokerData = false;
Set<String> brokerNameSet = new HashSet<String>();
List<BrokerData> brokerDataList = new LinkedList<BrokerData>();
topicRouteData.setBrokerDatas(brokerDataList);
HashMap<String, List<String>> filterServerMap = new HashMap<String, List<String>>();
topicRouteData.setFilterServerTable(filterServerMap);
try {
    try {
        //加入读锁
        this.lock.readLock().lockInterruptibly();
        //从topicQueueTable中通过主题topic查找Broker列表
        List<QueueData> queueDataList = this.topicQueueTable.get(topic);
        if (queueDataList != null) {
            //填充Broker信息到topicRouteData.QueueDatas
            topicRouteData.setQueueDatas(queueDataList);
            foundQueueData = true;

            Iterator<QueueData> it = queueDataList.iterator();
            while (it.hasNext()) {
                QueueData qd = it.next();
                brokerNameSet.add(qd.getBrokerName());
            }
            //遍历Broker集合
            for (String brokerName : brokerNameSet) {
                //从brokerAddrTable中通过BrokerName进行查找
                BrokerData brokerData = this.brokerAddrTable.get(brokerName);
                if (null != brokerData) {
                    BrokerData brokerDataClone = new BrokerData(brokerData.getCluster(), brokerData.getBrokerName(), (HashMap<Long, String>) brokerData.getBrokerAddrs().clone());
                    //填充BrokerData信息到brokerDatas
                    brokerDataList.add(brokerDataClone);
                    foundBrokerData = true;
                    for (final String brokerAddr : brokerDataClone.getBrokerAddrs().values()) {
                        List<String> filterServerList = this.filterServerTable.get(brokerAddr);
                        filterServerMap.put(brokerAddr, filterServerList);
                    }
                }
            }
        }
    } finally {
        //释放读锁
        this.lock.readLock().unlock();
    }
} catch (Exception e) {
    log.error("pickupTopicRouteData Exception", e);
}

log.debug("pickupTopicRouteData {} {}", topic, topicRouteData);

if (foundBrokerData && foundQueueData) {
    return topicRouteData;
}
return null;

3.5总结

路由注册:Broker启动之后会将自己的信息注册到NameServer,NameServer收到Broker心跳包之后会更新BrokerLiveTable中对应的Broker的相关信息,特别记录lastUpdateTimestamp

心跳机制:Broker会每隔30s向NameServer发送心跳包,告诉NameServer自己的运行状态良好。

路由发现:NameServer中的主题Topic信息更新之后,NameServer并不会实时将数据推送给订阅方Producer,而需要Producer自行通过Topic来查询路由信息。

路由删除:

  • Broker异常宕机:NameServer每隔10s扫描BrokerLiveTable,检查表中各个Broker最新的发送心跳包时间,如果这个时间与当前系统时间差超过120s,则该Broker会被视作已掉线,NameServer会将这个Broker从表中删除,并且更新各个表中与该Broker有关的所有信息。
  • Broker正常下线:NameServer接受到BrokerController发送来的UNREGISTER_BROKER请求之后,会执行上述步骤,对路由信息进行更新。

本文仅作为个人学习使用,如有不足或者错误请指正!

以上就是java开发RocketMQ之NameServer路由管理源码分析的详细内容,更多关于RocketMQ之NameServer路由管理的资料请关注我们其它相关文章!

(0)

相关推荐

  • springBoot整合RocketMQ及坑的示例代码

    版本: JDK:1.8 springBoot:1.5.10 rocketMQ:4.2.0 pom 配置: <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>1.5.10.RELEASE</version> </parent> <d

  • Java RocketMQ 路由注册与删除的实现

    简介 RocketMQ路由注册与删除是通过Broker与NameServer的心跳功能实现的.Broker启动时向集群中所有的NameServer发送心跳语句,每隔30s向集群中所有NameServer发送心跳包,NameServer收到Broker心跳包时会更新brokerLiveTable中的lastUpdateTimestamp,然后NameServer每隔10s扫描brokerLiveTable,如果连续120s没有收到心跳包,NameServer将移除该Broker的路由信息. 路由信

  • java rocketmq--消息的产生(普通消息)

    前言 与消息发送紧密相关的几行代码: 1. DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName"); 2. producer.start(); 3. Message msg = new Message(...) 4. SendResult sendResult = producer.send(msg); 5. producer.shutdown(); 那这几行代码执行时,背后都做了什么? 一. 首先

  • RocketMQ获取指定消息的实现方法(源码)

    概要 消息查询是什么? 消息查询就是根据用户提供的msgId从MQ中取出该消息 RocketMQ如果有多个节点如何查询? 问题:RocketMQ分布式结构中,数据分散在各个节点,即便是同一Topic的数据,也未必都在一个broker上.客户端怎么知道数据该去哪个节点上查? 猜想1:逐个访问broker节点查询数据 猜想2:有某种数据中心存在,该中心知道所有消息存储的位置,只要向该中心查询即可得到消息具体位置,进而取得消息内容 实际: 1.消息Id中含有消息所在的broker的地址信息(IP\Po

  • java开发RocketMQ之NameServer路由管理源码分析

    目录 1.前言 2.路由元信息 3.路由注册 3.1Broker路由注册 3.2NameServer处理路由注册 3.3路由删除 3.3.1Broker异常关闭 3.3.2Broker正常关闭 3.4路由发现 3.5总结 1.前言 NameServer主要作用是为消息消费者和消息生产者提供关于主题Topic的路由信息,那么NameServer需要存储路由的基本信息,还要管理Broker节点,包括路由注册.路由删除等. 2.路由元信息 路由元信息主要由RouteInfoManager来进行管理,这

  • java开发RocketMQ消息中间件原理基础详解

    RocketMQ 是什么 Github 上关于 RocketMQ 的介绍: RcoketMQ 是一款低延迟.高可靠.可伸缩.易于使用的消息中间件.具有以下特性: 支持发布/订阅(Pub/Sub)和点对点(P2P)消息模型 在一个队列中可靠的先进先出(FIFO)和严格的顺序传递 支持拉(pull)和推(push)两种消息模式 单一队列百万消息的堆积能力 支持多种消息协议,如 JMS.MQTT 等 分布式高可用的部署架构,满足至少一次消息传递语义 提供 docker 镜像用于隔离测试和云集群部署 提

  • java开发RocketMQ生产者高可用示例详解

    目录 引言 1 消息 1.1 topic 1.2 Body 1.3 tag 1.4 key 1.5 延迟级别 2 生产者高可用 2.1 客户端保证生产者高可用 2.1.1 重试机制 2.1.2 客户端容错 2.2 Broker端保证生产者高可用 引言 前边两章说了点基础的,从这章开始,我们挖挖源码.看看RocketMQ是怎么工作的. 首先呢,这个生产者就是送孩子去码头的家长,孩子们呢,就是消息了. 我们看看消息孩子们都长啥样. 1 消息 public class Message implemen

  • spring-session简介及实现原理源码分析

    一:spring-session介绍 1.简介 session一直都是我们做集群时需要解决的一个难题,过去我们可以从serlvet容器上解决,比如开源servlet容器-tomcat提供的tomcat-redis-session-manager.memcached-session-manager. 或者通过nginx之类的负载均衡做ip_hash,路由到特定的服务器上.. 但是这两种办法都存在弊端. spring-session是spring旗下的一个项目,把servlet容器实现的httpSe

  • Flink时间和窗口逻辑处理源码分析

    目录 概览 时间 重要类 WatermarkStrategy WatermarkGenerator TimerService 处理逻辑 窗口 重要类 Window WindowAssigner Triger Evictor WindowOperator InternalAppendingState 处理逻辑 总结 概览 计算模型 DataStream基础框架 事件时间和窗口 部署&调度 存储体系 底层支撑 在实时计算处理时,需要跟时间来打交道,如实时风控场景的时间行为序列,实时分析场景下的时间窗

  • Android ViewGroup事件分发和处理源码分析

    目录 正文 处理ACTION_DOWN事件 检测是否截断事件 不截断ACTION_DOWN事件 寻找处理事件的子View 事件分发给子View ViewGroup自己处理ACTION_DOWN事件 处理ACTION_DOWN总结 处理ACTION_MOVE事件 检测是否截断ACTION_MOVE事件 不截断ACTION_MOVE 事件分发给mFirstTouchTarget.child 截断ACTION_MOVE 处理 ACTION_UP 和 ACTION_CANCEL 事件 正确地使用requ

  • java使用websocket,并且获取HttpSession 源码分析(推荐)

    一:本文使用范围 此文不仅仅局限于spring boot,普通的spring工程,甚至是servlet工程,都是一样的,只不过配置一些监听器的方法不同而已. 本文经过作者实践,确认完美运行. 二:Spring boot使用websocket 2.1:依赖包 websocket本身是servlet容器所提供的服务,所以需要在web容器中运行,像我们所使用的tomcat,当然,spring boot中已经内嵌了tomcat. websocket遵循了javaee规范,所以需要引入javaee的包 <

  • java并发容器CopyOnWriteArrayList实现原理及源码分析

    CopyOnWriteArrayList是Java并发包中提供的一个并发容器,它是个线程安全且读操作无锁的ArrayList,写操作则通过创建底层数组的新副本来实现,是一种读写分离的并发策略,我们也可以称这种容器为"写时复制器",Java并发包中类似的容器还有CopyOnWriteSet.本文会对CopyOnWriteArrayList的实现原理及源码进行分析. 实现原理 我们都知道,集合框架中的ArrayList是非线程安全的,Vector虽是线程安全的,但由于简单粗暴的锁同步机制,

  • jdk动态代理源码分析过程

    代理对象的生成方法是: Proxy.newProxyInstance(...) ,进入这个方法内部,一步一步往下走会发现会调用 ProxyGenerator.generateProxyClass() ,这个方法用来生成代理类的字节码. 下面通过调用 ProxyGenerator.generateProxyClass()方法在本地生成代理类. 1.首先要有一个接口 2.生成代理类的方法如下 3.将生成的代理类导入到idea中查看是长这样 // // Source code recreated fr

  • java开发线上事故理解RocketMQ异步精髓

    目录 引言 1 业务场景 2 线程池模式 3 本地内存 + 定时任务 4 MQ 模式 5 Agent 服务 + MQ 模式 6 总结 第一层:什么场景下需要异步 第二层:异步的外功心法 第三层:异步的本质 引言 在高并发的场景下,异步是一个极其重要的优化方向. 前段时间,生产环境发生一次事故,笔者认为事故的场景非常具备典型性 . 写这篇文章,笔者想和大家深入探讨该场景的架构优化方案.希望大家读完之后,可以对异步有更深刻的理解. 1 业务场景 老师登录教研平台,会看到课程列表,点击课程后,课程会以

随机推荐