问题列表:

• Name Server 的作用是什么?

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

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

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

Name Server 作用

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 配置决定。

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

RouteInfoManager

//主题信息private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
//broker信息private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
//集群信息private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
//活跃broker信息private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
//过滤器信息private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;

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

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

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

    HashMap<String/* topic */, List<QueueData>> topicQueueTable;

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

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

public TopicRouteData pickupTopicRouteData(final String topic) {TopicRouteData topicRouteData = new TopicRouteData();boolean foundQueueData = false;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();//根据topic获取QueueData信息List<QueueData> queueDataList = this.topicQueueTable.get(topic);if (queueDataList != null) {topicRouteData.setQueueDatas(queueDataList);foundQueueData = true;Iterator<QueueData> 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<Long, String>) brokerData.getBrokerAddrs().clone());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);}if (log.isDebugEnabled()) {log.debug("pickupTopicRouteData {} {}", topic, topicRouteData);}if (foundBrokerData && foundQueueData) {return topicRouteData;}return null;}

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

Consumer需要哪些信息?

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

答

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

总结

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

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