RocketMQ源码 — 二、 NameServer

NameServer

作用：Producer和Consumer获取Broker的地址
目的：解耦Broker和Producer、Consumer
原理：使用netty作为通信工具，监听指定端口，如果是broker注册，将broker的信息保存在内存中并保存到文件中，producer和consumer获取broker地址的请求

RocketMQ包含的组件

NameServer：单点，供Producer和Consumer获取Broker地址
Producer：产生并发送消息
Consumer：接受并消费消息
Broker：消息暂存，消息转发

NamesrvController包含的组件

namesrvConfig：nameServer的配置
nettyServerConfig：NameServer的netty配置
remotingServer：NameServer 的netty服务器
scheduledExecutorService：routeInfoManager和kvCOnfigManager使用的定时线程池
remotingExecutor：netty使用的线程池
brokerHosekeppingService：
kvConfigManager：kv配置管理
routeInfoManager：包含broker的ip和对应的队列信息，说明producer可以往哪一个broker发送消息，consumer从哪一个broker pull消息

NameServer启动

NamesrvStartup.main0

NettySystemConfig配置
解析命令行参数，NettyServerConfig配置
logback配置
NamesrvController初始化，initialize
注册shutdown钩子
NamesrvController.start

NamesrvController.initialize

KVConfigManager.load加载原来的key-value文件到内存中
初始化NettyRemotingServer
注册requestProcessor，默认为DefaultRequestProcessor，用来处理netty接收到的信息
启动定时线程，每隔10s判断broker是否依然存活
启动定时线程，每隔10min打印出所有k-v

NamesrvController.start

启动netty server，使用NettyServerHandler进行read和write，最终调用到DefaultRequestProcessor.processRequest

NameServer处理信息

netty收到的所有消息都是DefaultRequestProcessor.processRequest处理的,根据不同的RequestCode执行不同的操作

public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    if (log.isDebugEnabled()) {
        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.REGISTER_BROKER:	// 注册borker信息
            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);
            }
        case RequestCode.UNREGISTER_BROKER:		// 取消注册broker
            return this.unregisterBroker(ctx, request);
        case RequestCode.GET_ROUTEINTO_BY_TOPIC:
            return this.getRouteInfoByTopic(ctx, request);		// 根据topic获取路由信息，在producer发送消息和consumer在pull消息的时候的时候会从nameServer 中获取
        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);
        default:
            break;
    }
    return null;
}

这里主要看注册broker，依次调用DefaultRequestProcessor.registerBroker->RouteInfoManager.registerBroker；
进行的主要操作是解析request，依次填充以下集合，供producer和consumer获取

// 某一个topic对应的逻辑队列
private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
// 每个broker 对应的信息
private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
// 某一个集群下对应的所有broker
private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
// 某一个broker对应的filter
private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;