【RocketMQ】Broker的启动

Posted on

BrokerStartup

BrokerStartup是Broker的启动类：

1.与NameServer启动时一样，会初始化相关配置参数，与NameServer不一样的地方是，Broker不仅会创建Netty Server的配置， 也会创建Netty Client的配置，因为Broker作为Client，需要向NameServer发送请求进行注册，同时Broker又需要作为Server接受Producer和Consumer的请求生产/消费消息；

2.Broker会从配置文件中获取NameServer的地址列表，后续向NameServer注册；

3.创建BrokerController，BrokerController是Broker的核心组件；

4.注册关闭时的钩子函数，在关闭前处理资源的关闭。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
public class BrokerStartup {
    // 入口
    public static void main(String[] args) {
        start(createBrokerController(args));
    }

    public static BrokerController start(BrokerController controller) {
        try {
            // 启动
            controller.start();

            String tip = "The broker[" + controller.getBrokerConfig().getBrokerName() + ", "
                + controller.getBrokerAddr() + "] boot success. serializeType=" + RemotingCommand.getSerializeTypeConfigInThisServer();

            if (null != controller.getBrokerConfig().getNamesrvAddr()) {
                tip += " and name server is " + controller.getBrokerConfig().getNamesrvAddr();
            }

            log.info(tip);
            System.out.printf("%s%n", tip);
            return controller;
        } catch (Throwable e) {
            e.printStackTrace();
            System.exit(-1);
        }

        return null;
    }

    // 创建BrokerController
    public static BrokerController createBrokerController(String[] args) {
        System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));

        if (null == System.getProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_SOCKET_SNDBUF_SIZE)) {
            NettySystemConfig.socketSndbufSize = 131072;
        }

        if (null == System.getProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_SOCKET_RCVBUF_SIZE)) {
            NettySystemConfig.socketRcvbufSize = 131072;
        }

        try {
            // 处理命令行相关参数
            Options options = ServerUtil.buildCommandlineOptions(new Options());
            commandLine = ServerUtil.parseCmdLine("mqbroker", args, buildCommandlineOptions(options),
                new PosixParser());
            if (null == commandLine) {
                System.exit(-1);
            }
            // 创建Broker配置
            final BrokerConfig brokerConfig = new BrokerConfig();
            // 创建Netty Server配置
            final NettyServerConfig nettyServerConfig = new NettyServerConfig();
            // 创建Netty Client配置
            final NettyClientConfig nettyClientConfig = new NettyClientConfig();
            // 设置Netty Cliet TLS
            nettyClientConfig.setUseTLS(Boolean.parseBoolean(System.getProperty(TLS_ENABLE,
                String.valueOf(TlsSystemConfig.tlsMode == TlsMode.ENFORCING))));
            // 设置Netty Server默认端口
            nettyServerConfig.setListenPort(10911);
            // 创建MessageStoreConfig，从名字上可以看出是存储消息的一些配置
            final MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
            // 如果是SLAVE
            if (BrokerRole.SLAVE == messageStoreConfig.getBrokerRole()) {
                int ratio = messageStoreConfig.getAccessMessageInMemoryMaxRatio() - 10;
                messageStoreConfig.setAccessMessageInMemoryMaxRatio(ratio);
            }
            // 如果启动的时候命令行中带了-c参数
            if (commandLine.hasOption('c')) {
                String file = commandLine.getOptionValue('c');
                // 读取自定义配置文件
                if (file != null) {
                    configFile = file;
                    InputStream in = new BufferedInputStream(new FileInputStream(file));
                    properties = new Properties();
                    properties.load(in);

                    properties2SystemEnv(properties);
                    MixAll.properties2Object(properties, brokerConfig);
                    MixAll.properties2Object(properties, nettyServerConfig);
                    MixAll.properties2Object(properties, nettyClientConfig);
                    MixAll.properties2Object(properties, messageStoreConfig);

                    BrokerPathConfigHelper.setBrokerConfigPath(file);
                    in.close();
                }
            }
            // 将命令行中的参数设置到BrokerConfig中
            MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), brokerConfig);

            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);
            }
            // 获取NameServer地址
            String namesrvAddr = brokerConfig.getNamesrvAddr();
            if (null != namesrvAddr) {
                // 解析NameServer地址
                try {
                    String[] addrArray = namesrvAddr.split(";");
                    for (String addr : addrArray) {
                        RemotingUtil.string2SocketAddress(addr);
                    }
                } catch (Exception e) {
                    System.out.printf(
                        "The Name Server Address[%s] illegal, please set it as follows, \"127.0.0.1:9876;192.168.0.1:9876\"%n",
                        namesrvAddr);
                    System.exit(-3);
                }
            }
            // 判断Broker角色
            switch (messageStoreConfig.getBrokerRole()) {
                case ASYNC_MASTER:
                case SYNC_MASTER:
                    brokerConfig.setBrokerId(MixAll.MASTER_ID);
                    break;
                case SLAVE:
                    if (brokerConfig.getBrokerId() <= 0) {
                        System.out.printf("Slave's brokerId must be > 0");
                        System.exit(-3);
                    }

                    break;
                default:
                    break;
            }
            // 是否基于delger管理主从同步和commitlog
            if (messageStoreConfig.isEnableDLegerCommitLog()) {
                brokerConfig.setBrokerId(-1);
            }
            // 设置HA监听端口
            messageStoreConfig.setHaListenPort(nettyServerConfig.getListenPort() + 1);
            // 日志相关
            LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
            JoranConfigurator configurator = new JoranConfigurator();
            configurator.setContext(lc);
            lc.reset();
            configurator.doConfigure(brokerConfig.getRocketmqHome() + "/conf/logback_broker.xml");
            // 如果启动带了-p参数
            if (commandLine.hasOption('p')) {
                InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.BROKER_CONSOLE_NAME);
                MixAll.printObjectProperties(console, brokerConfig);
                MixAll.printObjectProperties(console, nettyServerConfig);
                MixAll.printObjectProperties(console, nettyClientConfig);
                MixAll.printObjectProperties(console, messageStoreConfig);
                System.exit(0);
            } else if (commandLine.hasOption('m')) {
                InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.BROKER_CONSOLE_NAME);
                MixAll.printObjectProperties(console, brokerConfig, true);
                MixAll.printObjectProperties(console, nettyServerConfig, true);
                MixAll.printObjectProperties(console, nettyClientConfig, true);
                MixAll.printObjectProperties(console, messageStoreConfig, true);
                System.exit(0);
            }
            // 打印配置参数
            log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
            MixAll.printObjectProperties(log, brokerConfig);
            MixAll.printObjectProperties(log, nettyServerConfig);
            MixAll.printObjectProperties(log, nettyClientConfig);
            MixAll.printObjectProperties(log, messageStoreConfig);
            // 创建BrokerController
            final BrokerController controller = new BrokerController(
                brokerConfig,
                nettyServerConfig,
                nettyClientConfig,
                messageStoreConfig);
            // remember all configs to prevent discard
            controller.getConfiguration().registerConfig(properties);
            // 初始化BrokerStartup
            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) {
                        log.info("Shutdown hook was invoked, {}", this.shutdownTimes.incrementAndGet());
                        if (!this.hasShutdown) {
                            this.hasShutdown = true;
                            long beginTime = System.currentTimeMillis();
                            // 调用BrokerController的shutdown方法进行关资源的关闭
                            controller.shutdown();
                            long consumingTimeTotal = System.currentTimeMillis() - beginTime;
                            log.info("Shutdown hook over, consuming total time(ms): {}", consumingTimeTotal);
                        }
                    }
                }
            }, "ShutdownHook"));

            return controller;
        } catch (Throwable e) {
            e.printStackTrace();
            System.exit(-1);
        }

        return null;
    }

}

BrokerController

BrokerController是Broker的核心组件，包含了一些核心的功能组件和线程池，管理Broker的请求处理、后台线程和磁盘数据。

初始化

1.从磁盘上加载一些相关配置，如Topic配置、Consumer的Offset等数据；

2.创建消息管理组件DefaultMessageStore，如果开启了Dleger，初始化DLeger相关配置；

3.创建Netty Server服务器；

4.创建了一系列的线程池，用来定时调度后台任务，如处理消息拉取的任务、处理发送消息的任务；

5.处理NameServer地址列表;

6.一些其他的操作，暂时先不管。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
public boolean initialize() throws CloneNotSupportedException {
      // 从磁盘上加载Topic的配置
      boolean result = this.topicConfigManager.load();
      // 加载Consumer的Offset
      result = result && this.consumerOffsetManager.load();
      // 加载订阅组
      result = result && this.subscriptionGroupManager.load();
      // 加载过滤器
      result = result && this.consumerFilterManager.load();

      if (result) {
          try {
              // 创建消息管理组件DefaultMessageStore
              this.messageStore =
                  new DefaultMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener,
                      this.brokerConfig);
              // 初始化DLeger相关配置
              if (messageStoreConfig.isEnableDLegerCommitLog()) {
                  DLedgerRoleChangeHandler roleChangeHandler = new DLedgerRoleChangeHandler(this, (DefaultMessageStore) messageStore);
                  ((DLedgerCommitLog)((DefaultMessageStore) messageStore).getCommitLog()).getdLedgerServer().getdLedgerLeaderElector().addRoleChangeHandler(roleChangeHandler);
              }
              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);
          }
      }

      result = result && this.messageStore.load();

      if (result) {
          // 创建Netty Server
          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);
          // 发送消息的线程池，处理发送过来的消息
          this.sendMessageExecutor = new BrokerFixedThreadPoolExecutor(
              this.brokerConfig.getSendMessageThreadPoolNums(),
              this.brokerConfig.getSendMessageThreadPoolNums(),
              1000 * 60,
              TimeUnit.MILLISECONDS,
              this.sendThreadPoolQueue,
              new ThreadFactoryImpl("SendMessageThread_"));
          // 拉取消息的线程池，处理消息的拉取
          this.pullMessageExecutor = new BrokerFixedThreadPoolExecutor(
              this.brokerConfig.getPullMessageThreadPoolNums(),
              this.brokerConfig.getPullMessageThreadPoolNums(),
              1000 * 60,
              TimeUnit.MILLISECONDS,
              this.pullThreadPoolQueue,
              new ThreadFactoryImpl("PullMessageThread_"));
          // 处理回复消息的线程池
          this.replyMessageExecutor = new BrokerFixedThreadPoolExecutor(
              this.brokerConfig.getProcessReplyMessageThreadPoolNums(),
              this.brokerConfig.getProcessReplyMessageThreadPoolNums(),
              1000 * 60,
              TimeUnit.MILLISECONDS,
              this.replyThreadPoolQueue,
              new ThreadFactoryImpl("ProcessReplyMessageThread_"));
          // 处理消息查询的线程池
          this.queryMessageExecutor = new BrokerFixedThreadPoolExecutor(
              this.brokerConfig.getQueryMessageThreadPoolNums(),
              this.brokerConfig.getQueryMessageThreadPoolNums(),
              1000 * 60,
              TimeUnit.MILLISECONDS,
              this.queryThreadPoolQueue,
              new ThreadFactoryImpl("QueryMessageThread_"));

          this.adminBrokerExecutor =
              Executors.newFixedThreadPool(this.brokerConfig.getAdminBrokerThreadPoolNums(), new ThreadFactoryImpl(
                  "AdminBrokerThread_"));
          // 管理客户端的线程池
          this.clientManageExecutor = new ThreadPoolExecutor(
              this.brokerConfig.getClientManageThreadPoolNums(),
              this.brokerConfig.getClientManageThreadPoolNums(),
              1000 * 60,
              TimeUnit.MILLISECONDS,
              this.clientManagerThreadPoolQueue,
              new ThreadFactoryImpl("ClientManageThread_"));
          // 处理心跳机制的线程池
          this.heartbeatExecutor = new BrokerFixedThreadPoolExecutor(
              this.brokerConfig.getHeartbeatThreadPoolNums(),
              this.brokerConfig.getHeartbeatThreadPoolNums(),
              1000 * 60,
              TimeUnit.MILLISECONDS,
              this.heartbeatThreadPoolQueue,
              new ThreadFactoryImpl("HeartbeatThread_", true));
          // 处理事务结束的线程池
          this.endTransactionExecutor = new BrokerFixedThreadPoolExecutor(
              this.brokerConfig.getEndTransactionThreadPoolNums(),
              this.brokerConfig.getEndTransactionThreadPoolNums(),
              1000 * 60,
              TimeUnit.MILLISECONDS,
              this.endTransactionThreadPoolQueue,
              new ThreadFactoryImpl("EndTransactionThread_"));
          // 管理消费者的线程池
          this.consumerManageExecutor =
              Executors.newFixedThreadPool(this.brokerConfig.getConsumerManageThreadPoolNums(), new ThreadFactoryImpl(
                  "ConsumerManageThread_"));

          this.registerProcessor();
          /**一些后台任务的定时调度线程池**/
          final long initialDelay = UtilAll.computeNextMorningTimeMillis() - System.currentTimeMillis();
          final long period = 1000 * 60 * 60 * 24;
          // 处理Broker统计任务
          this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
              @Override
              public void run() {
                  try {
                      BrokerController.this.getBrokerStats().record();
                  } catch (Throwable e) {
                      log.error("schedule record error.", e);
                  }
              }
          }, initialDelay, period, TimeUnit.MILLISECONDS);
          // 处理消费者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);
          // 处理过滤器持久化到磁盘的定时任务
          this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
              @Override
              public void run() {
                  try {
                      BrokerController.this.consumerFilterManager.persist();
                  } catch (Throwable e) {
                      log.error("schedule persist consumer filter error.", e);
                  }
              }
          }, 1000 * 10, 1000 * 10, TimeUnit.MILLISECONDS);
          // 对broker保护的定时任务
          this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
              @Override
              public void run() {
                  try {
                      BrokerController.this.protectBroker();
                  } catch (Throwable e) {
                      log.error("protectBroker error.", e);
                  }
              }
          }, 3, 3, TimeUnit.MINUTES);

          this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
              @Override
              public void run() {
                  try {
                      BrokerController.this.printWaterMark();
                  } catch (Throwable e) {
                      log.error("printWaterMark error.", e);
                  }
              }
          }, 10, 1, TimeUnit.SECONDS);
          // 定时进行落后commitlog分发的任务
          this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

              @Override
              public void run() {
                  try {
                      log.info("dispatch behind commit log {} bytes", BrokerController.this.getMessageStore().dispatchBehindBytes());
                  } catch (Throwable e) {
                      log.error("schedule dispatchBehindBytes error.", e);
                  }
              }
          }, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);

          if (this.brokerConfig.getNamesrvAddr() != null) {
              // 更新NameServer地址列表
              this.brokerOuterAPI.updateNameServerAddressList(this.brokerConfig.getNamesrvAddr());
              log.info("Set user specified name server address: {}", this.brokerConfig.getNamesrvAddr());
          } else if (this.brokerConfig.isFetchNamesrvAddrByAddressServer()) {
              this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

                  @Override
                  public void run() {
                      try {
                          // 拉取NameServer地址列表
                          BrokerController.this.brokerOuterAPI.fetchNameServerAddr();
                      } catch (Throwable e) {
                          log.error("ScheduledTask fetchNameServerAddr exception", e);
                      }
                  }
              }, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS);
          }
          // Dleger相关的处理代码
          if (!messageStoreConfig.isEnableDLegerCommitLog()) {
              if (BrokerRole.SLAVE == this.messageStoreConfig.getBrokerRole()) {
                  if (this.messageStoreConfig.getHaMasterAddress() != null && this.messageStoreConfig.getHaMasterAddress().length() >= 6) {
                      this.messageStore.updateHaMasterAddress(this.messageStoreConfig.getHaMasterAddress());
                      this.updateMasterHAServerAddrPeriodically = false;
                  } else {
                      this.updateMasterHAServerAddrPeriodically = true;
                  }
              } else {
                  this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
                      @Override
                      public void run() {
                          try {
                              BrokerController.this.printMasterAndSlaveDiff();
                          } catch (Throwable e) {
                              log.error("schedule printMasterAndSlaveDiff error.", e);
                          }
                      }
                  }, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);
              }
          }
          // 先不管
          if (TlsSystemConfig.tlsMode != TlsMode.DISABLED) {
              // Register a listener to reload SslContext
              try {
                  fileWatchService = new FileWatchService(
                      new String[] {
                          TlsSystemConfig.tlsServerCertPath,
                          TlsSystemConfig.tlsServerKeyPath,
                          TlsSystemConfig.tlsServerTrustCertPath
                      },
                      new FileWatchService.Listener() {
                          boolean certChanged, keyChanged = false;

                          @Override
                          public void onChanged(String path) {
                              if (path.equals(TlsSystemConfig.tlsServerTrustCertPath)) {
                                  log.info("The trust certificate changed, reload the ssl context");
                                  reloadServerSslContext();
                              }
                              if (path.equals(TlsSystemConfig.tlsServerCertPath)) {
                                  certChanged = true;
                              }
                              if (path.equals(TlsSystemConfig.tlsServerKeyPath)) {
                                  keyChanged = true;
                              }
                              if (certChanged && keyChanged) {
                                  log.info("The certificate and private key changed, reload the ssl context");
                                  certChanged = keyChanged = false;
                                  reloadServerSslContext();
                              }
                          }

                          private void reloadServerSslContext() {
                              ((NettyRemotingServer) remotingServer).loadSslContext();
                              ((NettyRemotingServer) fastRemotingServer).loadSslContext();
                          }
                      });
              } catch (Exception e) {
                  log.warn("FileWatchService created error, can't load the certificate dynamically");
              }
          }
          // 处理事务相关的代码
          initialTransaction();
          initialAcl();
          initialRpcHooks();
      }
      return result;
  }

启动

1.start方法中启动了一系列的组件，比较核心的是Netty Server服务器的启动和BrokerOuterAPI的启动。Netty启动之后就可以接收网络请求，BrokerOuterAPI处理一些核心的功能，比如通过Netty Client向外发送请求；

2.向NameServer注册；

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
public void start() throws Exception {
     // 启动消息存储组件
     if (this.messageStore != null) {
         this.messageStore.start();
     }
     // 启动Netty Server服务器
     if (this.remotingServer != null) {
         this.remotingServer.start();
     }

     if (this.fastRemotingServer != null) {
         this.fastRemotingServer.start();
     }
     // 文件相关的服务组件启动
     if (this.fileWatchService != null) {
         this.fileWatchService.start();
     }
     // 核心组件，通过Netty Client发送请求等
     if (this.brokerOuterAPI != null) {
         this.brokerOuterAPI.start();
     }
     // 先不关注
     if (this.pullRequestHoldService != null) {
         this.pullRequestHoldService.start();
     }

     if (this.clientHousekeepingService != null) {
         this.clientHousekeepingService.start();
     }

     if (this.filterServerManager != null) {
         this.filterServerManager.start();
     }

     if (!messageStoreConfig.isEnableDLegerCommitLog()) {
         startProcessorByHa(messageStoreConfig.getBrokerRole());
         handleSlaveSynchronize(messageStoreConfig.getBrokerRole());
         this.registerBrokerAll(true, false, true);
     }
     // 定时任务向NameServer注册
     this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

         @Override
         public void run() {
             try {
                 // 向NameServer注册
                 BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister());
             } catch (Throwable e) {
                 log.error("registerBrokerAll Exception", e);
             }
         }
     }, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), TimeUnit.MILLISECONDS);

     if (this.brokerStatsManager != null) {
         this.brokerStatsManager.start();
     }

     if (this.brokerFastFailure != null) {
         this.brokerFastFailure.start();
     }
 }

服务注册

在注册的代码中会判断是否需要注册，如果需要注册，将会通过brokerOuterAPI核心组件发送请求到NameServer进行注册，Broker会把自己注册给所有的NameServer，然后对注册的结果进行处理。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
public synchronized void registerBrokerAll(final boolean checkOrderConfig, boolean oneway, boolean forceRegister) {
     // Topic配置相关的东西
     TopicConfigSerializeWrapper topicConfigWrapper = this.getTopicConfigManager().buildTopicConfigSerializeWrapper();

     if (!PermName.isWriteable(this.getBrokerConfig().getBrokerPermission())
         || !PermName.isReadable(this.getBrokerConfig().getBrokerPermission())) {
         ConcurrentHashMap<String, TopicConfig> topicConfigTable = new ConcurrentHashMap<String, TopicConfig>();
         for (TopicConfig topicConfig : topicConfigWrapper.getTopicConfigTable().values()) {
             TopicConfig tmp =
                 new TopicConfig(topicConfig.getTopicName(), topicConfig.getReadQueueNums(), topicConfig.getWriteQueueNums(),
                     this.brokerConfig.getBrokerPermission());
             topicConfigTable.put(topicConfig.getTopicName(), tmp);
         }
         topicConfigWrapper.setTopicConfigTable(topicConfigTable);
     }
     // 判断是否需要注册
     if (forceRegister || needRegister(this.brokerConfig.getBrokerClusterName(),
         this.getBrokerAddr(),
         this.brokerConfig.getBrokerName(),
         this.brokerConfig.getBrokerId(),
         this.brokerConfig.getRegisterBrokerTimeoutMills())) {
         // 注册
         doRegisterBrokerAll(checkOrderConfig, oneway, topicConfigWrapper);
     }
 }

 private void doRegisterBrokerAll(boolean checkOrderConfig, boolean oneway,
     TopicConfigSerializeWrapper topicConfigWrapper) {
     // 通过brokerOuterAPI核心组件发送请求到NameServer进行注册，Broker会把自己注册给所有的NameServer
     List<RegisterBrokerResult> registerBrokerResultList = this.brokerOuterAPI.registerBrokerAll(
         this.brokerConfig.getBrokerClusterName(),
         this.getBrokerAddr(),
         this.brokerConfig.getBrokerName(),
         this.brokerConfig.getBrokerId(),
         this.getHAServerAddr(),
         topicConfigWrapper,
         this.filterServerManager.buildNewFilterServerList(),
         oneway,
         this.brokerConfig.getRegisterBrokerTimeoutMills(),
         this.brokerConfig.isCompressedRegister());
     // 对注册的结果进行处理
     if (registerBrokerResultList.size() > 0) {
         RegisterBrokerResult registerBrokerResult = registerBrokerResultList.get(0);
         if (registerBrokerResult != null) {
             if (this.updateMasterHAServerAddrPeriodically && registerBrokerResult.getHaServerAddr() != null) {
                 this.messageStore.updateHaMasterAddress(registerBrokerResult.getHaServerAddr());
             }

             this.slaveSynchronize.setMasterAddr(registerBrokerResult.getMasterAddr());

             if (checkOrderConfig) {
                 this.getTopicConfigManager().updateOrderTopicConfig(registerBrokerResult.getKvTable());
             }
         }
     }
 }

总结

Broker的启动过程简单总结：

1.初始化相关配置参数，包括Broker的配置、Netty Client、Netty Server的配置等；

2.创建核心组件BrokerController，初始化了一系列的组件，并创建Netty Server和一系列的后台任务线程池，Netty Server用来接收网络请求，线程池用来调度各种任务；

3.启动BrokerController中的一系列的组件，并通过BrokerOuterAPI向NameServer注册；

注：图片来自于儒猿技术窝-从 0 开始带你成为消息中间件实战高手

参考

儒猿技术窝：从 0 开始带你成为消息中间件实战高手