Rocketmq源码分析04：broker 启动流程
java技术探秘
共 53386字，需浏览 107分钟
·
2021-04-22 21:34
注：本系列源码分析基于RocketMq 4.8.0，gitee仓库链接：https://gitee.com/funcy/rocketmq.git.
前面我们已经分析完了NameServer，从本文开始，我们将分析Broker。
1. 启动入口
broker的启动类为org.apache.rocketmq.broker.BrokerStartup，代码如下：
public class BrokerStartup {
...
public static void main(String[] args) {
start(createBrokerController(args));
}
...
}
在main()方法中，仅有一行代码，这行代码包含了两个操作:
createBrokerController(...)：创建BrokerControllerstart(...)：启动Broker
接下来我们就来分析这两个操作。
2. 创建BrokerController
创建BrokerController的方法为BrokerStartup#createBrokerController，代码如下：
/**
* 创建 broker 的配置参数
* @param args
* @return
*/
public static BrokerController createBrokerController(String[] args) {
...
try {
//解析命令行参数
Options options = ServerUtil.buildCommandlineOptions(new Options());
commandLine = ServerUtil.parseCmdLine("mqbroker", args, buildCommandlineOptions(options),
new PosixParser());
if (null == commandLine) {
System.exit(-1);
}
// 处理配置
final BrokerConfig brokerConfig = new BrokerConfig();
final NettyServerConfig nettyServerConfig = new NettyServerConfig();
final NettyClientConfig nettyClientConfig = new NettyClientConfig();
// tls安全相关
nettyClientConfig.setUseTLS(Boolean.parseBoolean(System.getProperty(TLS_ENABLE,
String.valueOf(TlsSystemConfig.tlsMode == TlsMode.ENFORCING))));
// 配置端口
nettyServerConfig.setListenPort(10911);
// 消息存储的配置
final MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
...
// 将命令行中的配置设置到brokerConfig对象中
MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), brokerConfig);
// 检查环境变量：ROCKETMQ_HOME
if (null == brokerConfig.getRocketmqHome()) {
System.out.printf("Please set the %s variable in your environment to match
the location of the RocketMQ installation", MixAll.ROCKETMQ_HOME_ENV);
System.exit(-2);
}
//省略一些配置
...
// 创建 brokerController
final BrokerController controller = new BrokerController(
brokerConfig,
nettyServerConfig,
nettyClientConfig,
messageStoreConfig);
controller.getConfiguration().registerConfig(properties);
// 初始化
boolean initResult = controller.initialize();
if (!initResult) {
controller.shutdown();
System.exit(-3);
}
// 关闭钩子，在关闭前处理一些操作
Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
private volatile boolean hasShutdown = false;
private AtomicInteger shutdownTimes = new AtomicInteger(0);
@Override
public void run() {
synchronized (this) {
if (!this.hasShutdown) {
...
// 这里会发一条注销消息给nameServer
controller.shutdown();
...
}
}
}
}, "ShutdownHook"));
return controller;
} catch (Throwable e) {
e.printStackTrace();
System.exit(-1);
}
return null;
}
这个方法的代码有点长，但功能并不多，总的来说就三个功能：
处理配置：主要是处理 nettyServerConfig与nettyClientConfig的配置，这块就是一些配置解析的操作，处理方式与NameServer很类似，这里就不多说了。创建及初始化 controller：调用方法controller.initialize()，这块内容我们后面分析。注册关闭钩子：调用 Runtime.getRuntime().addShutdownHook(...)，可以在jvm进程关闭前进行一些操作。
2.1 controller实例化
BrokerController的创建及初始化是在BrokerStartup#createBrokerController方法中进行，我们先来看看它的构造方法：
public BrokerController(
final BrokerConfig brokerConfig,
final NettyServerConfig nettyServerConfig,
final NettyClientConfig nettyClientConfig,
final MessageStoreConfig messageStoreConfig
) {
// 4个核心配置信息
this.brokerConfig = brokerConfig;
this.nettyServerConfig = nettyServerConfig;
this.nettyClientConfig = nettyClientConfig;
this.messageStoreConfig = messageStoreConfig;
// 管理consumer消费消息的offset
this.consumerOffsetManager = new ConsumerOffsetManager(this);
// 管理topic配置
this.topicConfigManager = new TopicConfigManager(this);
// 处理 consumer 拉消息请求的
this.pullMessageProcessor = new PullMessageProcessor(this);
this.pullRequestHoldService = new PullRequestHoldService(this);
// 消息送达的监听器
this.messageArrivingListener
= new NotifyMessageArrivingListener(this.pullRequestHoldService);
...
// 往外发消息的组件
this.brokerOuterAPI = new BrokerOuterAPI(nettyClientConfig);
...
}
BrokerController的构造方法很长，基本都是一些赋值操作，代码中已列出关键项，这些包括：
核心配置赋值：主要是 brokerConfig/nettyServerConfig/nettyClientConfig/messageStoreConfig四个配置ConsumerOffsetManager：管理consumer消费消息位置的偏移量，偏移量表示消费者组消费该topic消息的位置，后面再消费时，就从该位置后消费，避免重复消费消息，也避免了漏消费消息。topicConfigManager：topic配置管理器，就是用来管理topic配置的，如topic名称，topic队列数量pullMessageProcessor：消息处理器，用来处理消费者拉消息messageArrivingListener：消息送达的监听器，当生产者的消息送达时，由该监听器监听brokerOuterAPI：往外发消息的组件，如向NameServer发送注册/注销消息
以上这些组件的用处，这里先混个脸熟，我们后面再分析。
2.2 初始化controller
我们再来看看初始化操作，方法为BrokerController#initialize：
public boolean initialize() throws CloneNotSupportedException {
// 加载配置文件中的配置
boolean result = this.topicConfigManager.load();
result = result && this.consumerOffsetManager.load();
result = result && this.subscriptionGroupManager.load();
result = result && this.consumerFilterManager.load();
if (result) {
try {
// 消息存储管理组件，管理磁盘上的消息
this.messageStore =
new DefaultMessageStore(this.messageStoreConfig, this.brokerStatsManager,
this.messageArrivingListener, this.brokerConfig);
// 启用了DLeger,就创建DLeger相关组件
if (messageStoreConfig.isEnableDLegerCommitLog()) {
...
}
// broker统计组件
this.brokerStats = new BrokerStats((DefaultMessageStore) this.messageStore);
//load plugin
MessageStorePluginContext context = new MessageStorePluginContext(messageStoreConfig,
brokerStatsManager, messageArrivingListener, brokerConfig);
this.messageStore = MessageStoreFactory.build(context, this.messageStore);
this.messageStore.getDispatcherList().addFirst(
new CommitLogDispatcherCalcBitMap(this.brokerConfig, this.consumerFilterManager));
} catch (IOException e) {
result = false;
log.error("Failed to initialize", e);
}
}
// 加载磁盘上的记录，如commitLog写入的位置、消费者主题/队列的信息
result = result && this.messageStore.load();
if (result) {
// 处理 nettyServer
this.remotingServer = new NettyRemotingServer(
this.nettyServerConfig, this.clientHousekeepingService);
NettyServerConfig fastConfig = (NettyServerConfig) this.nettyServerConfig.clone();
fastConfig.setListenPort(nettyServerConfig.getListenPort() - 2);
this.fastRemotingServer = new NettyRemotingServer(
fastConfig, this.clientHousekeepingService);
// 创建线程池start... 这里会创建多种类型的线程池
...
// 处理consumer pull操作的线程池
this.pullMessageExecutor = new BrokerFixedThreadPoolExecutor(
this.brokerConfig.getPullMessageThreadPoolNums(),
this.brokerConfig.getPullMessageThreadPoolNums(),
1000 * 60,
TimeUnit.MILLISECONDS,
this.pullThreadPoolQueue,
new ThreadFactoryImpl("PullMessageThread_"));
...
// 创建线程池end...
// 注册处理器
this.registerProcessor();
// 启动定时任务start... 这里会启动好多的定时任务
...
// 定时将consumer消费到的offset进行持久化操作，即将数据保存到磁盘上
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
try {
BrokerController.this.consumerOffsetManager.persist();
} catch (Throwable e) {
log.error("schedule persist consumerOffset error.", e);
}
}
}, 1000 * 10, this.brokerConfig.getFlushConsumerOffsetInterval(), TimeUnit.MILLISECONDS);
...
// 启动定时任务end...
...
// 开启 DLeger 的一些操作
if (!messageStoreConfig.isEnableDLegerCommitLog()) {
...
}
// 处理tls配置
if (TlsSystemConfig.tlsMode != TlsMode.DISABLED) {
...
}
// 初始化一些操作
initialTransaction();
initialAcl();
initialRpcHooks();
}
return result;
}
这个还是很长，关键部分都做了注释，该方法所做的工作如下：
加载配置文件中的配置 赋值与初始化操作 创建线程池 注册处理器 启动定时任务
这里我们来看下注册处理器的操作this.registerProcessor():
1. 注册处理器：BrokerController#registerProcessor
this.registerProcessor()实际调用的方法是BrokerController#registerProcessor，代码如下：
public void registerProcessor() {
/**
* SendMessageProcessor
*/
SendMessageProcessor sendProcessor = new SendMessageProcessor(this);
sendProcessor.registerSendMessageHook(sendMessageHookList);
sendProcessor.registerConsumeMessageHook(consumeMessageHookList);
this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE, sendProcessor,
this.sendMessageExecutor);
this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE_V2, sendProcessor,
this.sendMessageExecutor);
this.remotingServer.registerProcessor(RequestCode.SEND_BATCH_MESSAGE, sendProcessor,
this.sendMessageExecutor);
this.remotingServer.registerProcessor(RequestCode.CONSUMER_SEND_MSG_BACK, sendProcessor,
this.sendMessageExecutor);
...
/**
* PullMessageProcessor
*/
this.remotingServer.registerProcessor(RequestCode.PULL_MESSAGE, this.pullMessageProcessor,
this.pullMessageExecutor);
this.pullMessageProcessor.registerConsumeMessageHook(consumeMessageHookList);
/**
* ReplyMessageProcessor
*/
ReplyMessageProcessor replyMessageProcessor = new ReplyMessageProcessor(this);
replyMessageProcessor.registerSendMessageHook(sendMessageHookList);
...
}
这个方法里注册了许许多多的处理器，这里仅列出了与消息相关的内容，如发送消息、回复消息、拉取消息等，后面在处理producer/consumer的消息时，就会用到这些处理器，这里先不展开分析。
2. remotingServer注册处理器：NettyRemotingServer#registerProcessor
我们来看下remotingServer注册处理器的操作，方法为NettyRemotingServer#registerProcessor：
public class NettyRemotingServer extends NettyRemotingAbstract implements RemotingServer {
...
@Override
public void registerProcessor(int requestCode, NettyRequestProcessor processor,
ExecutorService executor) {
ExecutorService executorThis = executor;
if (null == executor) {
executorThis = this.publicExecutor;
}
Pair<NettyRequestProcessor, ExecutorService> pair = new Pair<NettyRequestProcessor,
ExecutorService>(processor, executorThis);
this.processorTable.put(requestCode, pair);
}
...
}
最终，这些处理器注册到了processorTable中，它是NettyRemotingAbstract的成员变量，定义如下：
HashMap<Integer/* request code */, Pair<NettyRequestProcessor, ExecutorService>>
这是一个hashMap的结构，key为code，value为Pair，该类中有两个成员变量：NettyRequestProcessor、ExecutorService，code与NettyRequestProcessor的映射关系就是在hashMap里存储的。
2.3 注册关闭钩子：Runtime.getRuntime().addShutdownHook(...)
接着我们来看看注册关闭钩子的操作：
// 关闭钩子，在关闭前处理一些操作
Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
private volatile boolean hasShutdown = false;
private AtomicInteger shutdownTimes = new AtomicInteger(0);
@Override
public void run() {
synchronized (this) {
if (!this.hasShutdown) {
...
// 这里会发一条注销消息给nameServer
controller.shutdown();
...
}
}
}
}, "ShutdownHook"));
跟进BrokerController#shutdown方法：
public void shutdown() {
// 调用各组件的shutdown方法
...
// 发送注销消息到NameServer
this.unregisterBrokerAll();
...
// 持久化consumer的消费偏移量
this.consumerOffsetManager.persist();
// 又是调用各组件的shutdown方法
...
这个方法里会调用各组件的shutdown()方法、发送注销消息给NameServer、持久化consumer的消费偏移量，这里我们主要看发送注销消息的方法BrokerController#unregisterBrokerAll:
private void unregisterBrokerAll() {
// 发送一条注销消息给nameServer
this.brokerOuterAPI.unregisterBrokerAll(
this.brokerConfig.getBrokerClusterName(),
this.getBrokerAddr(),
this.brokerConfig.getBrokerName(),
this.brokerConfig.getBrokerId());
}
继续进入BrokerOuterAPI#unregisterBrokerAll：
public void unregisterBrokerAll(
final String clusterName,
final String brokerAddr,
final String brokerName,
final long brokerId
) {
// 获取所有的 nameServer，遍历发送注销消息
List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
if (nameServerAddressList != null) {
for (String namesrvAddr : nameServerAddressList) {
try {
this.unregisterBroker(namesrvAddr, clusterName, brokerAddr, brokerName, brokerId);
log.info("unregisterBroker OK, NamesrvAddr: {}", namesrvAddr);
} catch (Exception e) {
log.warn("unregisterBroker Exception, {}", namesrvAddr, e);
}
}
}
}
这个方法里，会获取到所有的nameServer，然后逐个发送注销消息，继续进入BrokerOuterAPI#unregisterBroker方法：
public void unregisterBroker(
final String namesrvAddr,
final String clusterName,
final String brokerAddr,
final String brokerName,
final long brokerId
) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException,
InterruptedException, MQBrokerException {
UnRegisterBrokerRequestHeader requestHeader = new UnRegisterBrokerRequestHeader();
requestHeader.setBrokerAddr(brokerAddr);
requestHeader.setBrokerId(brokerId);
requestHeader.setBrokerName(brokerName);
requestHeader.setClusterName(clusterName);
// 发送的注销消息：RequestCode.UNREGISTER_BROKER
RemotingCommand request = RemotingCommand.createRequestCommand(
c, requestHeader);
RemotingCommand response = this.remotingClient.invokeSync(namesrvAddr, request, 3000);
assert response != null;
switch (response.getCode()) {
case ResponseCode.SUCCESS: {
return;
}
default:
break;
}
throw new MQBrokerException(response.getCode(), response.getRemark(), brokerAddr);
}
最终调用的是RemotingClient#invokeSync进行消息发送，请求code是RequestCode.UNREGISTER_BROKER，这就与NameServer接收broker的注销消息对应上了。
3. 启动Broker：start(...)
我们再来看看Broker的启动流程，处理方法为BrokerController#start：
public void start() throws Exception {
// 启动各组件
// 启动消息存储相关组件
if (this.messageStore != null) {
this.messageStore.start();
}
// 启动 remotingServer，其实就是启动一个netty服务，用来接收producer传来的消息
if (this.remotingServer != null) {
this.remotingServer.start();
}
...
// broker对外发放消息的组件，向nameServer上报存活消息时使用了它，也是一个netty服务
if (this.brokerOuterAPI != null) {
this.brokerOuterAPI.start();
}
...
// broker 核心的心跳注册任务
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
try {
BrokerController.this.registerBrokerAll(true, false,
brokerConfig.isForceRegister());
} catch (Throwable e) {
log.error("registerBrokerAll Exception", e);
}
}
// brokerConfig.getRegisterNameServerPeriod() 值为 1000 * 30，最终计算得到默认30秒执行一次
}, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)),
TimeUnit.MILLISECONDS);
...
}
这个方法主要就是启动各组件了，这里列出了几大重要组件的启动：
messageStore：消息存储组件，在这个组件里，会启动消息存储相关的线程，如消息的投递操作、commitLog文件的flush操作、comsumeQueue文件的flush操作等remotingServer：netty服务，用来接收请求消息，如producer发送过来的消息brokerOuterAPI：也是一个netty服务，用来对外发送消息，如向nameServer上报心跳消息启动定时任务： broker向nameServer发送注册消息
这里我们重点来看定时任务是如何发送心跳发送的。
处理注册消息发送的时间间隔如下：
Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)
这行代码看着长，但意思就一句话：时间间隔可以自行配置，但不能小于10s，不能大于60s，默认是30s.
处理消息注册的方法为BrokerController#registerBrokerAll(...)，代码如下：
public synchronized void registerBrokerAll(final boolean checkOrderConfig,
boolean oneway, boolean forceRegister) {
TopicConfigSerializeWrapper topicConfigWrapper
= this.getTopicConfigManager().buildTopicConfigSerializeWrapper();
// 处理topic相关配置
if (!PermName.isWriteable(this.getBrokerConfig().getBrokerPermission())
|| !PermName.isReadable(this.getBrokerConfig().getBrokerPermission())) {
...
}
// 这里会判断是否需要进行注册
if (forceRegister || needRegister(this.brokerConfig.getBrokerClusterName(),
this.getBrokerAddr(),
this.brokerConfig.getBrokerName(),
this.brokerConfig.getBrokerId(),
this.brokerConfig.getRegisterBrokerTimeoutMills())) {
// 进行注册操作
doRegisterBrokerAll(checkOrderConfig, oneway, topicConfigWrapper);
}
}
这个方法就是用来处理注册操作的，不过注册前会先验证下是否需要注册，验证是否需要注册的方法为BrokerController#needRegister, 代码如下：
private boolean needRegister(final String clusterName,
final String brokerAddr,
final String brokerName,
final long brokerId,
final int timeoutMills) {
TopicConfigSerializeWrapper topicConfigWrapper
= this.getTopicConfigManager().buildTopicConfigSerializeWrapper();
// 判断是否需要进行注册
List<Boolean> changeList = brokerOuterAPI.needRegister(clusterName, brokerAddr, brokerName,
brokerId, topicConfigWrapper, timeoutMills);
// 有一个发生了变化，就表示需要注册了
boolean needRegister = false;
for (Boolean changed : changeList) {
if (changed) {
needRegister = true;
break;
}
}
return needRegister;
}
这个方法调用了brokerOuterAPI.needRegister(...)来判断broker是否发生了变化，只要一个NameServer上发生了变化，就说明需要进行注册操作。
brokerOuterAPI.needRegister(...)是如何判断broker是否发生了变化的呢？继续跟进BrokerOuterAPI#needRegister：
public List<Boolean> needRegister(
final String clusterName,
final String brokerAddr,
final String brokerName,
final long brokerId,
final TopicConfigSerializeWrapper topicConfigWrapper,
final int timeoutMills) {
final List<Boolean> changedList = new CopyOnWriteArrayList<>();
// 获取所有的 nameServer
List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
if (nameServerAddressList != null && nameServerAddressList.size() > 0) {
final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());
// 遍历所有的nameServer,逐一发送请求
for (final String namesrvAddr : nameServerAddressList) {
brokerOuterExecutor.execute(new Runnable() {
@Override
public void run() {
try {
QueryDataVersionRequestHeader requestHeader
= new QueryDataVersionRequestHeader();
...
// 向nameServer发送消息，命令是 RequestCode.QUERY_DATA_VERSION
RemotingCommand request = RemotingCommand
.createRequestCommand(RequestCode.QUERY_DATA_VERSION, requestHeader);
// 把当前的 DataVersion 发到 nameServer
request.setBody(topicConfigWrapper.getDataVersion().encode());
// 发请求到nameServer
RemotingCommand response = remotingClient
.invokeSync(namesrvAddr, request, timeoutMills);
DataVersion nameServerDataVersion = null;
Boolean changed = false;
switch (response.getCode()) {
case ResponseCode.SUCCESS: {
QueryDataVersionResponseHeader queryDataVersionResponseHeader =
(QueryDataVersionResponseHeader) response
.decodeCommandCustomHeader(QueryDataVersionResponseHeader.class);
changed = queryDataVersionResponseHeader.getChanged();
byte[] body = response.getBody();
if (body != null) {
// 拿到 DataVersion
nameServerDataVersion = DataVersion.decode(body, D
ataVersion.class);
// 这里是判断的关键
if (!topicConfigWrapper.getDataVersion()
.equals(nameServerDataVersion)) {
changed = true;
}
}
if (changed == null || changed) {
changedList.add(Boolean.TRUE);
}
}
default:
break;
}
...
} catch (Exception e) {
...
} finally {
countDownLatch.countDown();
}
}
});
}
try {
countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
log.error("query dataversion from nameserver countDownLatch await Exception", e);
}
}
return changedList;
}
这个方法里，先是遍历所有的nameServer，向每个nameServer都发送一条code为RequestCode.QUERY_DATA_VERSION的参数，参数为当前broker的DataVersion，当nameServer收到消息后，就返回nameServer中保存的、与当前broker对应的DataVersion，当两者版本不相等时，就表明当前broker发生了变化，需要重新注册。
DataVersion是个啥呢？它的部分代码如下：
public class DataVersion extends RemotingSerializable {
// 时间戳
private long timestamp = System.currentTimeMillis();
// 计数器，可以理解为最近的版本号
private AtomicLong counter = new AtomicLong(0);
public void nextVersion() {
this.timestamp = System.currentTimeMillis();
this.counter.incrementAndGet();
}
/**
* equals 方法，当 timestamp 与 counter 都相等时，则两者相等
*/
@Override
public boolean equals(final Object o) {
if (this == o)
return true;
if (o == null || getClass() != o.getClass())
return false;
final DataVersion that = (DataVersion) o;
if (timestamp != that.timestamp) {
return false;
}
if (counter != null && that.counter != null) {
return counter.longValue() == that.counter.longValue();
}
return (null == counter) && (null == that.counter);
}
...
}
从DataVersion的equals()方法来看，只有当timestamp与counter都相等时，两个DataVersion对象才相等。那这两个值会在哪里被修改呢？从DataVersion#nextVersion方法的调用情况来看，引起这两个值的变化主要有两种：
broker上新创建了一个topictopic的发了的变化
在这两种情况下，DataVersion#nextVersion方法被调用，从而引起DataVersion的改变。DataVersion改变了，就表明当前broker需要向nameServer注册了。
让我们再回到BrokerController#registerBrokerAll(...)方法：
public synchronized void registerBrokerAll(final boolean checkOrderConfig,
boolean oneway, boolean forceRegister) {
...
// 这里会判断是否需要进行注册
if (forceRegister || needRegister(this.brokerConfig.getBrokerClusterName(),
this.getBrokerAddr(),
this.brokerConfig.getBrokerName(),
this.brokerConfig.getBrokerId(),
this.brokerConfig.getRegisterBrokerTimeoutMills())) {
// 进行注册操作
doRegisterBrokerAll(checkOrderConfig, oneway, topicConfigWrapper);
}
}
处理注册的方法为BrokerController#doRegisterBrokerAll，稍微看下它的流程：
private void doRegisterBrokerAll(boolean checkOrderConfig, boolean oneway,
TopicConfigSerializeWrapper topicConfigWrapper) {
// 注册
List<RegisterBrokerResult> registerBrokerResultList = this.brokerOuterAPI.registerBrokerAll(
this.brokerConfig.getBrokerClusterName(),
this.getBrokerAddr(),
this.brokerConfig.getBrokerName(),
this.brokerConfig.getBrokerId(),
this.getHAServerAddr(),
// 这个对象里就包含了当前broker的版本信息
topicConfigWrapper,
this.filterServerManager.buildNewFilterServerList(),
oneway,
this.brokerConfig.getRegisterBrokerTimeoutMills(),
this.brokerConfig.isCompressedRegister());
...
}
继续跟下去，最终调用的是BrokerOuterAPI#registerBroker方法：
private RegisterBrokerResult registerBroker(
final String namesrvAddr,
final boolean oneway,
final int timeoutMills,
final RegisterBrokerRequestHeader requestHeader,
final byte[] body
) throws RemotingCommandException, MQBrokerException, RemotingConnectException,
RemotingSendRequestException, RemotingTimeoutException, InterruptedException {
// 构建请求
RemotingCommand request = RemotingCommand
.createRequestCommand(RequestCode.REGISTER_BROKER, requestHeader);
request.setBody(body);
// 处理发送操作：sendOneWay
if (oneway) {
try {
// 注册操作
this.remotingClient.invokeOneway(namesrvAddr, request, timeoutMills);
} catch (RemotingTooMuchRequestException e) {
// Ignore
}
return null;
...
}
....
}
所以，所谓的注册操作，就是当nameServer发送一条code为RequestCode.REGISTER_BROKER的消息，消息里会带上当前broker的topic信息、版本号等。
4.总结
本文主要分析了broker的启动流程，总的来说，启动流程分为3个：
解析配置文件，这一步会解析各种配置，并将其赋值到对应的对象中 BrokerController创建及初始化：创建了BrokerController对象，并进行初始化操作，所谓的初始化，就是加载配置文件中配置、创建线程池、注册请求处理器、启动定时任务等BrokerController启动：这一步是启动broker的核心组件，如messageStore(消息存储)、remotingServer(netty服务，用来处理producer与consumer请求)、brokerOuterAPI(netty服务，用来向nameServer上报当前broker信息)等。
在分析启动过程中，重点分析了两类消息的发送：
在 ShutdownHook中，broker会向nameServer发送注销消息，这表明在broker关闭前，nameServer会清除当前broker的注册信息broker启动后，会启动一个定时任务，定期判断是否需要向nameServer注册，判断是否需要注册时，会向nameServer发送code为QUERY_DATA_VERSION的消息，从nameServer得到当前broker的版本号，该版本号与本地版本号不一致，就表示需要向broker重新注册了，即发送注册消息。
限于篇幅，本文就先分析到这里了，下一篇继续分析broker相关内容。
限于作者个人水平，文中难免有错误之处，欢迎指正！原创不易，商业转载请联系作者获得授权，非商业转载请注明出处。
本文首发于微信公众号 Java技术探秘，如果您喜欢本文，欢迎关注该公众号，让我们一起在技术的世界里探秘吧！
评论