RocketMQ（1.NameServer源码）

NameServer功能简述

主要功能如下

1. 服务注册与发现：Nameserver扮演了RocketMQ集群中服务注册中心的角色。当RocketMQ中的Broker、Producer和Consumer启动时，它们会向Nameserver注册自己的网络地址和角色信息。Nameserver维护着集群中所有活跃实例的信息，并提供查询接口供其他组件发现这些实例。这样，Producer和Consumer可以根据Nameserver提供的信息找到Broker，实现消息的发布和消费。
2. Topic路由信息管理：RocketMQ中的消息按照Topic进行分类。Nameserver维护着Topic到Broker之间的路由关系，包括哪些Topic被哪些Broker负责。这样，当Producer发送消息时，它会根据Topic在Nameserver查询路由信息，得知应该将消息发送到哪些Broker上。
3. 负载均衡：Nameserver通过周期性地检查Broker的状态信息，包括负载、存活状态等，来维护一个Broker的列表，供Producer和Consumer使用。这样做的目的是为了在消息生产和消费过程中实现负载均衡，避免某个Broker过载，影响整个系统性能。
4. 权限控制：Nameserver支持对连接到集群的客户端进行权限控制。通过配置，管理员可以限制哪些客户端可以连接到Broker，增加系统的安全性。
5. 集群管理：Nameserver允许管理员对Broker和Topic进行动态管理，包括增加、删除Broker节点，创建和删除Topic等操作。这使得RocketMQ集群能够根据实际业务需求进行扩展和调整。

综上所述，RockerMQ Nameserver是RocketMQ集群中起着至关重要、不可或缺的作用。其服务注册与发现、Topic路由信息管理、负载均衡、权限控制和集群管理等功能，Nameserver确保整个RocketMQ系统的稳定、高效和安全运行。

核心源码

main0

public static void main(String[] args) {// 本地启动需设置环境变量System.setProperty(MixAll.ROCKETMQ_HOME_PROPERTY, "/Users/config/rocketmq");main0(args);controllerManagerMain();}

1.parseCommandlineAndConfigFile（解析命令行和配置文件，初始化namesrvConfig、nettyServerConfig、nettyClientConfig等配置对象）

public static void parseCommandlineAndConfigFile(String[] args) throws Exception {System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));Options options = ServerUtil.buildCommandlineOptions(new Options());CommandLine commandLine = ServerUtil.parseCmdLine("mqnamesrv", args, buildCommandlineOptions(options), new DefaultParser());if (null == commandLine) {System.exit(-1);return;}namesrvConfig = new NamesrvConfig();nettyServerConfig = new NettyServerConfig();nettyClientConfig = new NettyClientConfig();nettyServerConfig.setListenPort(9876);if (commandLine.hasOption('c')) {String file = commandLine.getOptionValue('c');if (file != null) {InputStream in = new BufferedInputStream(Files.newInputStream(Paths.get(file)));properties = new Properties();properties.load(in);MixAll.properties2Object(properties, namesrvConfig);MixAll.properties2Object(properties, nettyServerConfig);MixAll.properties2Object(properties, nettyClientConfig);if (namesrvConfig.isEnableControllerInNamesrv()) {controllerConfig = new ControllerConfig();MixAll.properties2Object(properties, controllerConfig);}namesrvConfig.setConfigStorePath(file);System.out.printf("load config properties file OK, %s%n", file);in.close();}}MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);if (commandLine.hasOption('p')) {MixAll.printObjectProperties(logConsole, namesrvConfig);MixAll.printObjectProperties(logConsole, nettyServerConfig);MixAll.printObjectProperties(logConsole, nettyClientConfig);if (namesrvConfig.isEnableControllerInNamesrv()) {MixAll.printObjectProperties(logConsole, controllerConfig);}System.exit(0);}if (null == namesrvConfig.getRocketmqHome()) {System.out.printf("Please set the %s variable in your environment to match the location of the RocketMQ installation%n", MixAll.ROCKETMQ_HOME_ENV);System.exit(-2);}MixAll.printObjectProperties(log, namesrvConfig);MixAll.printObjectProperties(log, nettyServerConfig);}

2.createAndStartNamesrvController（创建并运行NamesrvController）

public static NamesrvController start(final NamesrvController controller) throws Exception {if (null == controller) {throw new IllegalArgumentException("NamesrvController is null");}boolean initResult = controller.initialize();if (!initResult) {controller.shutdown();System.exit(-3);}Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, (Callable<Void>) () -> {controller.shutdown();return null;}));controller.start();return controller;}

一、controller.initialize()初始化

1. 加载配置

private final HashMap<String/* Namespace */, HashMap<String/* Key */, String/* Value */>> configTable =new HashMap<>();
public void load() {String content = null;try {content = MixAll.file2String(this.namesrvController.getNamesrvConfig().getKvConfigPath());} catch (IOException e) {log.warn("Load KV config table exception", e);}if (content != null) {KVConfigSerializeWrapper kvConfigSerializeWrapper =KVConfigSerializeWrapper.fromJson(content, KVConfigSerializeWrapper.class);if (null != kvConfigSerializeWrapper) {this.configTable.putAll(kvConfigSerializeWrapper.getConfigTable());log.info("load KV config table OK");}}}

2. 初始化网络组件

private void initiateNetworkComponents() {// remotingServer是一个NettyRemotingServer对象，用于接收和处理来自其他服务的请求或响应// BrokerHouseKeepingService对象用于处理Broker的连接和断开事件this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);// remotingClient是一个NettyRemotingClient对象，它用于向其他服务发送请求或响应this.remotingClient = new NettyRemotingClient(this.nettyClientConfig);}

3. 初始化线程执行器

private void initiateThreadExecutors() {this.defaultThreadPoolQueue = new LinkedBlockingQueue<>(this.namesrvConfig.getDefaultThreadPoolQueueCapacity());// 用于处理默认的远程请求this.defaultExecutor = new ThreadPoolExecutor(this.namesrvConfig.getDefaultThreadPoolNums(), this.namesrvConfig.getDefaultThreadPoolNums(), 1000 * 60, TimeUnit.MILLISECONDS, this.defaultThreadPoolQueue, new ThreadFactoryImpl("RemotingExecutorThread_")) {@Overrideprotected <T> RunnableFuture<T> newTaskFor(final Runnable runnable, final T value) {return new FutureTaskExt<>(runnable, value);}};this.clientRequestThreadPoolQueue = new LinkedBlockingQueue<>(this.namesrvConfig.getClientRequestThreadPoolQueueCapacity());// 处理客户端的路由信息请求this.clientRequestExecutor = new ThreadPoolExecutor(this.namesrvConfig.getClientRequestThreadPoolNums(), this.namesrvConfig.getClientRequestThreadPoolNums(), 1000 * 60, TimeUnit.MILLISECONDS, this.clientRequestThreadPoolQueue, new ThreadFactoryImpl("ClientRequestExecutorThread_")) {@Overrideprotected <T> RunnableFuture<T> newTaskFor(final Runnable runnable, final T value) {return new FutureTaskExt<>(runnable, value);}};}

4. 注册remoting server处理器

private void registerProcessor() {if (namesrvConfig.isClusterTest()) {// ClusterTestRequestProcessor是一个用于集群测试的处理器，它会在请求前后添加一些环境信息，比如产品环境名称、请求时间等this.remotingServer.registerDefaultProcessor(new ClusterTestRequestProcessor(this, namesrvConfig.getProductEnvName()), this.defaultExecutor);} else {// Support get route info only temporarily// 在 namesrvConfig.isClusterTest() = false 时如果收到请求的 requestCode 等于 RequestCode.GET_ROUTEINFO_BY_TOPIC 则会使用ClientRequestProcessor来处理；当收到其他请求时，会使用DefaultRequestProcessor来处理。ClientRequestProcessor clientRequestProcessor = new ClientRequestProcessor(this);this.remotingServer.registerProcessor(RequestCode.GET_ROUTEINFO_BY_TOPIC, clientRequestProcessor, this.clientRequestExecutor);// DefaultRequestProcessor是一个用于正常运行的处理器，它会根据请求的类型，调用不同的方法来处理，比如注册Broker、获取路由信息、更新配置等。this.remotingServer.registerDefaultProcessor(new DefaultRequestProcessor(this), this.defaultExecutor);}}

5. 启动定时任务

private void startScheduleService() {// 首次延迟5毫秒执行，后续执行间隔5秒执行（分两种情况）// 后续执行：1.当执行任务时间小于间隔时间时，延迟（间隔时间-任务执行时间）执行//         2.当任务执行时间大于间隔时间，则任务结束立即执行下一次任务// 间隔扫描不活跃的Brokerthis.scanExecutorService.scheduleAtFixedRate(NamesrvController.this.routeInfoManager::scanNotActiveBroker,5, this.namesrvConfig.getScanNotActiveBrokerInterval(), TimeUnit.MILLISECONDS);//每隔10分钟打印所有KV配置this.scheduledExecutorService.scheduleAtFixedRate(NamesrvController.this.kvConfigManager::printAllPeriodically,1, 10, TimeUnit.MINUTES);// 每隔1分钟打印WaterMarkthis.scheduledExecutorService.scheduleAtFixedRate(() -> {try {NamesrvController.this.printWaterMark();} catch (Throwable e) {LOGGER.error("printWaterMark error.", e);}}, 10, 1, TimeUnit.SECONDS);}

6. 初始化ssl上下文（配置remotingServer使用TLS协议进行安全通信）
7. 注册RPC钩子

private void initiateRpcHooks() {// 注册RPC钩子 在remotingServer处理请求之前或之后执行一些自定义的逻辑this.remotingServer.registerRPCHook(new ZoneRouteRPCHook());}

二、注册JVM钩子

1. controllerManager.shutdown()

// 注册一个ShutdownHookThread对象，JVM钩子，在程序终止时调用controllerManager.shutdown()，释放资源
Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, (Callable<Void>) () -> {controllerManager.shutdown();return null;}));
public void shutdown() {this.heartbeatManager.shutdown();this.controllerRequestExecutor.shutdown();this.notifyService.shutdown();this.controller.shutdown();this.remotingClient.shutdown();}

三、controller.start()运行

1. 运行NettyRemotingServer，启动一个NettyRemotingServer，用于接收和处理客户端的请求

2. 运行remotingClient，启动一个NettyRemotingClient，用于向其他服务发送请求

3. 运行FileWatch，调用它的start方法，启动一个文件监视服务，用于动态加载证书文件

用来跟踪SSL

4. 运行RouteInfoManager，启动一个路由信息管理器，用于维护Broker和Topic的路由关系

unRegisterService.start() 提供了一种以批处理方式注销Broker的机制。扫描离线的Broker BlockingQueue，take()获取数据，while循环获取不活跃的broker。take()是阻塞的，drainTo()方法获取队列中的全部数据。

registerBroker() 接收Broker每隔30秒上报Broker信息