当前位置：首页 > news >正文

RocketMQ 源码梳理

news 2026/6/6 5:13:18

（一）：NameServer 启动原理 + Broker 启动原理

一、NameServer 启动原理（源码级 + 流程图）

1.1 NameServer 核心作用（回顾）

整个 RocketMQ 的路由中心 + 注册中心
管理 Broker 信息、Topic 路由信息
无状态、无主从、轻量高可用
生产者 / 消费者定期拉取路由信息

1.2 NameServer 启动入口类

核心启动类：

java

org.apache.rocketmq.namesrv.NamesrvStartup

启动命令：mqnamesrv最终执行的就是这个类的main方法。

1.3 NameServer 启动核心流程（源码步骤）

NameServer 启动分为5 大核心阶段：

阶段 1：解析命令行与配置

解析启动参数（-c 配置文件、-p 打印配置等）
创建NamesrvConfig、NettyServerConfig
加载配置文件namesrv.conf

阶段 2：创建 NamesrvController 核心控制器

NamesrvController是 NameServer 的大脑，统筹所有组件：

路由管理器（RouteInfoManager）
配置管理器
Netty 服务器

阶段 3：初始化核心组件

初始化路由信息管理器：创建 Broker、Topic、集群路由表
初始化 Netty 配置：绑定端口 9876
初始化定时任务：
- 定期扫描（默认 10s）移除过期不心跳的 Broker

阶段 4：启动 Netty 服务端

启动 Netty 监听端口 9876
等待 Broker、Producer、Consumer 连接

阶段 5：注册 JVM 钩子，优雅关闭

JVM 退出时释放资源、关闭线程池

1.4 NameServer 启动完整流程图

1.5 NameServer 启动关键源码说明

// NamesrvStartup 启动核心代码 public static NamesrvController createAndStartNamesrvController(String[] args) throws MQClientException, InterruptedException { // 1. 解析配置 NamesrvConfig namesrvConfig = new NamesrvConfig(); NettyServerConfig nettyServerConfig = new NettyServerConfig(); nettyServerConfig.setListenPort(9876); // 默认端口 // 2. 创建控制器 NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig); // 3. 初始化 controller.initialize(); // 4. 启动 controller.start(); // 5. 优雅关闭 Runtime.getRuntime().addShutdownHook(...); return controller; }

核心定时任务（10s 扫描一次）：

this.scheduledExecutorService.scheduleAtFixedRate(() -> { // 扫描并移除 120s 内未发送心跳的 Broker NamesrvController.this.routeInfoManager.scanNotActiveBroker(); }, 5, 10, TimeUnit.SECONDS);

二、Broker 启动原理（源码级 + 流程图）

2.1 Broker 核心作用（回顾）

消息接收、存储、投递的核心组件
向 NameServer 注册、发送心跳
管理 CommitLog、ConsumeQueue、Index
主从同步、副本同步、高可用

2.2 Broker 启动入口类

核心启动类：

org.apache.rocketmq.broker.BrokerStartup

启动命令：mqbroker -n localhost:9876

2.3 Broker 启动核心流程（源码步骤）

Broker 启动比 NameServer 复杂，分为7 个核心阶段：

阶段 1：解析启动参数与配置

解析-n指定 NameServer 地址
加载broker.conf
创建：
- BrokerConfig（Broker 配置）
- NettyServerConfig（端口 10911）
- NettyServerClient(客户端配置(broker向nameserver发送心跳等请求))
- MessageStoreConfig（存储配置）
- AuthConfig(认证 + 授权配置集合，Broker 启动阶段加载、初始化安全校验组件，管控所有客户端 / 集群节点接入权限)

阶段 2：创建 BrokerController 核心控制器

Broker 的总管家，统筹：

Netty 通信
消息存储
副本同步
注册中心交互

阶段 3：初始化所有组件

初始化消息存储（CommitLog、ConsumeQueue）
时间轮服务
初始化 Netty 服务
启动客户端
管理Topic路由信息
初始化心跳定时任务
服务状态

阶段 4：启动消息存储服务

加载 CommitLog、ConsumeQueue、Index
恢复异常宕机数据
启动刷盘线程（同步 / 异步）

阶段 5：启动 Netty 服务器

监听 10911 端口
处理 Producer、Consumer 请求

阶段 6：向 NameServer 注册 Broker

发送注册请求
启动定时心跳（每 30s 上报一次）

阶段 7：注册 JVM 钩子，启动完成

2.4 Broker 启动完整流程图

2.5 Broker 启动关键源码说明

// Broker 启动核心代码 public static BrokerController createAndStartBrokerController(String[] args) { // 1. 解析配置 BrokerConfig brokerConfig = new BrokerConfig(); NettyServerConfig nettyServerConfig = new NettyServerConfig(); nettyServerConfig.setListenPort(10911); // 2. 创建控制器 BrokerController controller = new BrokerController( brokerConfig, nettyServerConfig, nettyClientConfig, messageStoreConfig ); // 3. 初始化 controller.initialize(); // 4. 启动 controller.start(); return controller; }

Broker 注册与心跳（关键）：

// 启动后立即注册 this.brokerOuterAPI.registerBrokerAll(...); // 定时心跳（30s） this.scheduledExecutorService.scheduleAtFixedRate(() -> { BrokerController.this.brokerOuterAPI.registerBrokerAll(...); }, 0, 30, TimeUnit.SECONDS);