RocketMQ源码学习(六)-Name Server

摘要：完全无状态，可集群部署与集群中的其中一个节点随机选择建立长连接，定期从取路由信息，并向提供服务的建立长连接，且定时向发送心跳。既可以从订阅消息，也可以从订阅消息，订阅规则由配置决定。

问题列表:

Name Server 的作用是什么?

Name Server 存储了Broker的什么信息?

Name Server 为Producer的提供些什么信息?

Name Server 为Consuner的提供些什么信息?

Name Server在RocketMQ中犹如如它名字一样,是提供Broker发现服务的.

Producer 与 Name Server 集群中的其中一个节点(随机选择)建立长连接，定期从 Name Server 取 Topic 路由信息，并向提供 Topic 服务的 Master 建立长连接，且定时向 Master 发送心跳。Producer 完全无状态，可集群部署

Consumer 与 Name Server 集群中的其中一个节点(随机选择)建立长连接，定期从 Name Server 取 Topic 路由信息，并向提供 Topic 服务的 Master、Slave 建立长连接，且定时向 Master、Slave 发送心跳。Consumer 既可以从 Master 订阅消息，也可以从 Slave 订阅消息，订阅规则由 Broker 配置决定。

RouteInfoManager

//主题信息
 private final HashMap> topicQueueTable;
//broker信息
    private final HashMap brokerAddrTable;
//集群信息
    private final HashMap> clusterAddrTable;
//活跃broker信息
    private final HashMap brokerLiveTable;
//过滤器信息
    private final HashMap/* Filter Server */> filterServerTable;

我们注意到保存broker的Map有两个，即brokerAddrTable用来保存所有的broker列表和brokerLiveTable用来保存当前活跃的broker列表，而BrokerData用来保存broker的主要新增，而BrokerLiveInfo只用来保存上次更新（心跳）时间，我们可以直接看看RouteInfoManager中扫描非活跃broker的方法：

    public void scanNotActiveBroker() {
        Iterator> it = this.brokerLiveTable.entrySet().iterator();
        while (it.hasNext()) {
            Entry 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());
            }
        }
    }

这个方法由在initialize的定时线程池加载,每十秒执行一次.可以看出，如果两分钟内都没收到一个broker的心跳数据，则直接将其从brokerLiveTable中移除，注意，这还会导致该broker从brokerAddrTable被删除，当然，如果该broker是Master，则它的所有Slave的broker都将被删除。具体细节可以参看RouteInfoManager的onChannelDestroy方法.

    HashMap> topicQueueTable;

    private String brokerName;  // broker的名称
    private int readQueueNums;  // 读队列数量
    private int writeQueueNums; // 写队列数量
    private int perm;           // 读写权限
    private int topicSynFlag;   // 同步复制还是异步复制标记

NameServer 维护了key为topic,List的数据为Producer提供服务.返回TopicRouteData信息
RouteInfoManager.pickupTopicRouteData

public TopicRouteData pickupTopicRouteData(final String topic) {
        TopicRouteData topicRouteData = new TopicRouteData();
        boolean foundQueueData = false;
        boolean foundBrokerData = false;
        Set brokerNameSet = new HashSet();
        List brokerDataList = new LinkedList();
        topicRouteData.setBrokerDatas(brokerDataList);

        HashMap> filterServerMap = new HashMap>();
        topicRouteData.setFilterServerTable(filterServerMap);

        try {
            try {
                this.lock.readLock().lockInterruptibly();
                //根据topic获取QueueData信息
                List queueDataList = this.topicQueueTable.get(topic);
                if (queueDataList != null) {
                    topicRouteData.setQueueDatas(queueDataList);
                    foundQueueData = true;

                    Iterator it = queueDataList.iterator();
                    while (it.hasNext()) {
                        QueueData qd = it.next();
                        brokerNameSet.add(qd.getBrokerName());
                    }

                    for (String brokerName : brokerNameSet) {
                        BrokerData brokerData = this.brokerAddrTable.get(brokerName);
                        if (null != brokerData) {
                        //根据broker名称获取其地址信息
                            BrokerData brokerDataClone = new BrokerData(brokerData.getCluster(), brokerData.getBrokerName(), (HashMap) brokerData
                                .getBrokerAddrs().clone());
                            brokerDataList.add(brokerDataClone);
                            foundBrokerData = true;
                            for (final String brokerAddr : brokerDataClone.getBrokerAddrs().values()) {
                                List filterServerList = this.filterServerTable.get(brokerAddr);
                                filterServerMap.put(brokerAddr, filterServerList);
                            }
                        }
                    }
                }
            } finally {
                this.lock.readLock().unlock();
            }
        } catch (Exception e) {
            log.error("pickupTopicRouteData Exception", e);
        }

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

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

        return null;
    }

Consumer需要哪些信息?

   
1. Consumer需要的topic的broker信息
2. 每一个consumer group都有哪些consumer，对应的topic是谁

答

1.如上节所述
2.此信息保存在Broker中

Name Server比较简单,如同一个简单的web服务,提供配置信息,只不过CRUD的不是数据库而是json文件.

此次RocketMQ学习就告一段落了,只描述了我比较关心的流程,很多细节没能涉及到,有时间再写吧,如有疑问和错误请在评论中指出,thx!