RocketMQ 源码分析（二）NameServer 源码分析

1 启动流程

首先看下 namesrv 模块下的结构，结构比较简单，只存在 config、processor、routeinfo 三个包再加上
namesrv控制器。NameServer 要作用是为消息生产者消息消费者提供关于主题 Topic 的路由信息，
那么 NameServer 要存储路由信息，还要能够管理 Broker 节点，包括路由注册，路由删除等功能。

NamesrvStartup 启动类，启动的方法很简单，只有十几行代码。

public static void main(String[] args) {
	main0(args);
}

public static NamesrvController main0(String[] args) {
	try {
		// 创建 namesrv 控制器
		NamesrvController controller = createNamesrvController(args);
		// 启动 namesrv 核心
		start(controller);
		String tip = "The Name Server boot success. serializeType=" + RemotingCommand.getSerializeTypeConfigInThisServer();
		log.info(tip);
		System.out.printf("%s%n", tip);
		return controller;
	} catch (Throwable e) {
		e.printStackTrace();
		System.exit(-1);
	}
	return null;
}

2 具体步骤 2.1 NamesrvStartup#createNamesrvController: 创建 namesrv 控制器

下面只会保留关键的代码

public static NamesrvController createNamesrvController(String[] args) throws IOException, JoranException {
	Options options = ServerUtil.buildCommandlineOptions(new Options());
	commandLine = ServerUtil.parseCmdLine("mqnamesrv", args, buildCommandlineOptions(options), new PosixParser());
	final NamesrvConfig namesrvConfig = new NamesrvConfig();
	final NettyServerConfig nettyServerConfig = new NettyServerConfig();
	// 修改 namerv rpc 监听端口为 9876， 默认为 8888
	nettyServerConfig.setListenPort(9876);
	// 根据启动命令配置 namesrvConfig
	MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);
	// 配置日志地址
	configurator.doConfigure(namesrvConfig.getRocketmqHome() + "/conf/logback_namesrv.xml");
	// 打印 namesrvConfig、nettyServerConfig 配置到日志中
	MixAll.printObjectProperties(log, namesrvConfig);
	MixAll.printObjectProperties(log, nettyServerConfig);
	//实例化 namesrv 控制器
	final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);
	// remember all configs to prevent discard
	controller.getConfiguration().registerConfig(properties);
	return controller;
}

2.2 NamesrvController#NamesrvController: 构造方法，其中包含了 broker、namesrv 等配置的初始化。

public NamesrvController(NamesrvConfig namesrvConfig, NettyServerConfig nettyServerConfig) {
	this.namesrvConfig = namesrvConfig;
	this.nettyServerConfig = nettyServerConfig;
	this.kvConfigManager = new KVConfigManager(this);
	// 初始化 broker、topicQueue
	this.routeInfoManager = new RouteInfoManager();
	// 创建一个 netty 的服务端
	this.brokerHousekeepingService = new BrokerHousekeepingService(this);
	// 设置配置信息：namesrvConfig，nettyServerConfig
	this.configuration = new Configuration(log,this.namesrvConfig, this.nettyServerConfig);
	// 设置存储路径
	this.configuration.setStorePathFromConfig(this.namesrvConfig, "configStorePath");
}
----------------------------------------------
RouteInfoManager#RouteInfoManager: 构造方法，初始化 broker、topicQueue 等信息
public RouteInfoManager() {
	//Topic 消息队列路由信息，消息发送时根据路由表进行负载均衡
	this.topicQueueTable = new HashMap>(1024);
	// Broker 基础信息， brokerName 所属集群名称 主备 Broker地址
	this.brokerAddrTable = new HashMap(128);
	// Broker 集群信息，存储集群中所有 Broker 名称
	this.clusterAddrTable = new HashMap>(32);
	// Broker 状态信息 NameServer 每次收到心跳包时会替换该信息
	this.brokerLiveTable = new HashMap(256);
	// Broker 上的 FilterServer 列表，用于类模式消息过滤
	this.filterServerTable = new HashMap>(256);
}

2.3 NamesrvStartup#start: 启动 namesrv 核心。

public static NamesrvController start(final NamesrvController controller) throws Exception {
	// initResult 启动结果
	boolean initResult = controller.initialize();
	// 添加 namesrv 关闭时的回调
	Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable() {
		@Override
		public Void call() throws Exception {
			controller.shutdown();
			return null;
		}
	}));
	//namesrv 控制器开启
	controller.start();
	return controller;
}

2.4 NamesrvController#initialize: 初始化 controller，主要是开启 rpc 远程连接、定时任务等。

public boolean initialize() {
	this.kvConfigManager.load();
	// 创建rpc，用于和 broker 连接
	this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
	// 创建远程线程池
	this.remotingExecutor =
		Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));
	// 注册处理器
	this.registerProcessor();
	// 定时任务：扫描不活跃的 broker
	this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
		@Override
		public void run() {
			NamesrvController.this.routeInfoManager.scanNotActiveBroker();
		}
	}, 5, 10, TimeUnit.SECONDS);
	this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
		@Override
		public void run() {
			NamesrvController.this.kvConfigManager.printAllPeriodically();
		}
	}, 1, 10, TimeUnit.MINUTES);
	return true;
}

2.5 NamesrvController#start: namesrv 控制器开启

public void start() throws Exception {
	// 开启 rpc 用于注册 broker
	this.remotingServer.start();
	// fileWatchService 开启文件监听，如果文件发现变更，则会进行操作
	if (this.fileWatchService != null) {
		this.fileWatchService.start();
	}
}

3 NameServer 路由注册、故障剔除 3.1 RoutelnfoManager: namesrv 路由实现类

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

public RouteInfoManager() {
	//Topic 消息队列路由信息，消息发送时根据路由表进行负载均衡
	this.topicQueueTable = new HashMap>(1024);
	// Broker 基础信息， brokerName 所属集群名称 主备 Broker地址
	this.brokerAddrTable = new HashMap(128);
	// Broker 集群信息，存储集群中所有 Broker 名称
	this.clusterAddrTable = new HashMap>(32);
	// Broker 状态信息 NameServer 每次收到心跳包时会替换该信息
	this.brokerLiveTable = new HashMap(256);
	// Broker 上的 FilterServer 列表，用于类模式消息过滤
	this.filterServerTable = new HashMap>(256);
}

public class QueueData implements Comparable {
	private String brokerName;
	private int readQueueNums;
	private int writeQueueNums;
	private int perm;
	private int topicSynFlag;
}

public class BrokerData implements Comparable {
	private String cluster;
	private String brokerName;
	private HashMap brokerAddrs;
}

class BrokerLiveInfo {
	private long lastUpdateTimestamp;
	private DataVersion dataVersion;
	private Channel channel;
	private String haServerAddr;
}

3.2 路由注册

上面提及到 namesrv 服务端启动后，当 broker 或者 producer 向 namesrv 发送请求时，namesrv 会进行处理，包括 broker 的注册。最终会定位到 DefaultRequestProcessor#processRequest，该方法会根据 requestCode，选择对应的方法，例如 broker注册的 requestCode = 103，调用 RouteInfoManager#registerBroker。

class NettyServerHandler extends SimpleChannelInboundHandler {
	@Override
	protected void channelRead0(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception {
		// 处理接收到的消息
		processMessageReceived(ctx, msg);
	}
}

public RegisterBrokerResult registerBroker(
    final String clusterName,
    final String brokerAddr,
    final String brokerName,
    final long brokerId,
    final String haServerAddr,
    final TopicConfigSerializeWrapper topicConfigWrapper,
    final List filterServerList,
    final Channel channel) {
    // 定义返回结果
    RegisterBrokerResult result = new RegisterBrokerResult();
    try {
        try {
            // 读写锁，由于是写操作，那么采用 writeLock，所以对应的属性采用的是 hashmap。
            this.lock.writeLock().lockInterruptibly();
            // 把 brokerName 加入 对应的集群中
            Set brokerNames = this.clusterAddrTable.get(clusterName);
            if (null == brokerNames) {
                brokerNames = new HashSet();
                this.clusterAddrTable.put(clusterName, brokerNames);
            }
            brokerNames.add(brokerName);
            // 是否为第一次注册
            boolean registerFirst = false;
            // 更新 brokerId 对应的 brokerAddress 集合中的属性
            BrokerData brokerData = this.brokerAddrTable.get(brokerName);
            if (null == brokerData) {
                registerFirst = true;
                brokerData = new BrokerData(clusterName, brokerName, new HashMap());
                this.brokerAddrTable.put(brokerName, brokerData);
            }
            String oldAddr = brokerData.getBrokerAddrs().put(brokerId, brokerAddr);
            registerFirst = registerFirst || (null == oldAddr);
            
            if (null != topicConfigWrapper
                && MixAll.MASTER_ID == brokerId) {
                if (this.isBrokerTopicConfigChanged(brokerAddr, topicConfigWrapper.getDataVersion())
                    || registerFirst) {
                    ConcurrentMap tcTable =
                        topicConfigWrapper.getTopicConfigTable();
                    if (tcTable != null) {
                        for (Map.Entry entry : tcTable.entrySet()) {
                            this.createAndUpdateQueueData(brokerName, entry.getValue());
                        }
                    }
                }
            }
            // 更新 broker 的存活表
            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);
            }
            //注册 Broker 的过滤器 Server 地址列表 ，1个Broker上会关联多个FilterServer消息过滤服务器
            if (filterServerList != null) {
                if (filterServerList.isEmpty()) {
                    this.filterServerTable.remove(brokerAddr);
                } else {
                    this.filterServerTable.put(brokerAddr, filterServerList);
                }
            }
            // 如果当前 broker 是从节点，那么更新 master 节点信息
            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);
                    }
                }
            }
        } finally {
            this.lock.writeLock().unlock();
        }
    } catch (Exception e) {
        log.error("registerBroker Exception", e);
    }

    return result;
}

Namesrv 与 broker 保持长连接，通过心跳检测更新 namesrv 中 broker 的状态及路由表等信息，采用读写锁，使得多个 producer 并发读，保证消息发送时的高并发，但同一时刻只会处理一个心跳包，多个心跳包串行执行。

3.3 路由删除

NamesrvController#initialize: namesrv 启动时，定义了两个定时线程池，其中一个是每 10s 扫描 brokerLiveTable，如果存在上次更新时间距离当前超过了过期时间（120s）时，移除 broker 与 namesrv 的连接，并且更新路由表等信息。

this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
        NamesrvController.this.routeInfoManager.scanNotActiveBroker();
    }
}

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

org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#onChannelDestroy: broker 与 namesrv 通道关闭时，执行路由表的更新操作。

this.lock.writeLock().lockInterruptibly();
// 从 brokerLiveTable、 filterServerTable中移除 broker
this.brokerLiveTable.remove(brokerAddrFound);
this.filterServerTable.remove(brokerAddrFound);
String brokerNameFound = null;
boolean removeBrokerName = false;
Iterator> itBrokerAddrTable =
    this.brokerAddrTable.entrySet().iterator();
while (itBrokerAddrTable.hasNext() && (null == brokerNameFound)) {
    BrokerData brokerData = itBrokerAddrTable.next().getValue();

    Iterator> it = brokerData.getBrokerAddrs().entrySet().iterator();
    while (it.hasNext()) {
        Entry entry = it.next();
        Long brokerId = entry.getKey();
        String brokerAddr = entry.getValue();
        if (brokerAddr.equals(brokerAddrFound)) {
            brokerNameFound = brokerData.getBrokerName();
            it.remove();
            log.info("remove brokerAddr[{}, {}] from brokerAddrTable, because channel destroyed",
                     brokerId, brokerAddr);
            break;
        }
    }

    if (brokerData.getBrokerAddrs().isEmpty()) {
        removeBrokerName = true;
        itBrokerAddrTable.remove();
        log.info("remove brokerName[{}] from brokerAddrTable, because channel destroyed",
                 brokerData.getBrokerName());
    }
}

if (brokerNameFound != null && removeBrokerName) {
    Iterator>> it = this.clusterAddrTable.entrySet().iterator();
    while (it.hasNext()) {
        Entry> entry = it.next();
        String clusterName = entry.getKey();
        Set brokerNames = entry.getValue();
        boolean removed = brokerNames.remove(brokerNameFound);
        if (removed) {
            log.info("remove brokerName[{}], clusterName[{}] from clusterAddrTable, because channel destroyed",
                     brokerNameFound, clusterName);

            if (brokerNames.isEmpty()) {
                log.info("remove the clusterName[{}] from clusterAddrTable, because channel destroyed and no broker in this cluster",
                         clusterName);
                it.remove();
            }

            break;
        }
    }
}
if (removeBrokerName) {
    // 遍历所有主题的队列，如果队列中包含当前 broker 的队列，则移除，如果 topic 是只包含 broker 的队列，那么全部移除该 topic
    Iterator>> itTopicQueueTable =
        this.topicQueueTable.entrySet().iterator();
    while (itTopicQueueTable.hasNext()) {
        Entry> entry = itTopicQueueTable.next();
        String topic = entry.getKey();
        List queueDataList = entry.getValue();

        Iterator itQueueData = queueDataList.iterator();
        while (itQueueData.hasNext()) {
            QueueData queueData = itQueueData.next();
            if (queueData.getBrokerName().equals(brokerNameFound)) {
                itQueueData.remove();
                log.info("remove topic[{} {}], from topicQueueTable, because channel destroyed",
                         topic, queueData);
            }
        }

        if (queueDataList.isEmpty()) {
            itTopicQueueTable.remove();
            log.info("remove topic[{}] all queue, from topicQueueTable, because channel destroyed",
                     topic);
        }
    }
}
this.lock.writeLock().unlock();

3.4 路由发现

调用org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor#getRouteInfoByTopic: 根据 topic 找到所在的路由信息，返回信息为 TopicRouteData。RocketMQ 路由发现是非实时的，当 Topic 路由出现变化后， NameServer 不主动推送给客户端而是由客户端定时拉取主题最新的路由。根据主题名称拉取路由信息的命令编码为： GET ROUTEINTO BY_TOPIC。

public RemotingCommand getRouteInfoByTopic(ChannelHandlerContext ctx,
                                           RemotingCommand request) throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);
    final GetRouteInfoRequestHeader requestHeader =
        (GetRouteInfoRequestHeader) request.decodeCommandCustomHeader(GetRouteInfoRequestHeader.class);
// 获取 topic 路由信息
TopicRouteData topicRouteData = this.namesrvController.getRouteInfoManager().pickupTopicRouteData(requestHeader.getTopic());
}

org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#pickupTopicRouteData：核心方法，这个才是实际获取 topic 路由信息的方法。

// 返回结果
public class TopicRouteData extends RemotingSerializable {
    // 顺序消息配置内容，来自于 kvConfig
    private String orderTopicConf;
    // topic 对应的队列数据
    private List queueDatas;
    // broker 信息
    private List brokerDatas;
    // broker 上对应的过滤服务器的地址
    private HashMap> filterServerTable;
}

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);
    // 读锁
    this.lock.readLock().lockInterruptibly();
    List queueDataList = this.topicQueueTable.get(topic);
    // 得到所有的队列，然后筛选出对应的 broker
    if (queueDataList != null) {
        topicRouteData.setQueueDatas(queueDataList);
        foundQueueData = true;

        Iterator it = queueDataList.iterator();
        while (it.hasNext()) {
            QueueData qd = it.next();
            brokerNameSet.add(qd.getBrokerName());
        }
        // 根据 brokerName 集合找到对应的 brokerData 以及对应的 filterServer
        for (String brokerName : brokerNameSet) {
            BrokerData brokerData = this.brokerAddrTable.get(brokerName);
            if (null != brokerData) {
                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);
                }
            }
        }
    }
this.lock.readLock().unlock();
if (foundBrokerData && foundQueueData) {
	return topicRouteData;
}
	return null;
}

RocketMQ 源码分析（二）NameServer 源码分析

大数据系统相关栏目本月热门文章