上篇环境搭建及NamesrvController分析： RocketMQ源码环境搭建&NamesrvController分析

BrokerController 分析

BrokerStartup#main -> 调用

BrokerStartup#start(createBrokerController(args))

createBrokerController(args)

  1. public static BrokerController createBrokerController(String[] args) {
  2.     // NamesrvController一样，设置环境变量
  3.     System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
  4.     // 数据校验，设默认值
  5.     if (null == System.getProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_SOCKET_SNDBUF_SIZE)) {
  6.         NettySystemConfig.socketSndbufSize = 131072;
  7.     }
  9.     if (null == System.getProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_SOCKET_RCVBUF_SIZE)) {
  10.         NettySystemConfig.socketRcvbufSize = 131072;
  11.     }
  13.     try {
  14.         // 创建命令行的选项
  15.         Options options = ServerUtil.buildCommandlineOptions(new Options());
  16.         commandLine = ServerUtil.parseCmdLine("mqbroker", args, buildCommandlineOptions(options),
  17.                                               new PosixParser());
  18.         if (null == commandLine) {
  19.             System.exit(-1);
  20.         }
  21.         // 创建BrokerConfig，NettyServerConfig，NettyClientConfig 三个配置类
  22.         final BrokerConfig brokerConfig = new BrokerConfig();
  23.         final NettyServerConfig nettyServerConfig = new NettyServerConfig();
  24.         final NettyClientConfig nettyClientConfig = new NettyClientConfig();
  25.         // 设置是否使用安全链接
  26.         nettyClientConfig.setUseTLS(Boolean.parseBoolean(System.getProperty(TLS_ENABLE,
  27.                                                                             String.valueOf(TlsSystemConfig.tlsMode == TlsMode.ENFORCING))));
  28.         // 设置Broker的监听端口
  29.         nettyServerConfig.setListenPort(10911);
  30.         // 设置消息的存储配置类
  31.         final MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
  32.         // 刚刚创建，messageStoreConfig的默认是是ASYNC_MASTER,这个判断永远不成立
  33.         // 有点不明白，为什么会出现这个东西
  34.         if (BrokerRole.SLAVE == messageStoreConfig.getBrokerRole()) {
  35.             int ratio = messageStoreConfig.getAccessMessageInMemoryMaxRatio() - 10;
  36.             messageStoreConfig.setAccessMessageInMemoryMaxRatio(ratio);
  37.         }
  38.         // 加载配置文件
  39.         if (commandLine.hasOption('c')) {
  40.             String file = commandLine.getOptionValue('c');
  41.             if (file != null) {
  42.                 configFile = file;
  43.                 InputStream in = new BufferedInputStream(new FileInputStream(file));
  44.                 properties = new Properties();
  45.                 properties.load(in);
  46.                 // 将properties中的内容，通过反射，设置到各个配置类中
  47.                 properties2SystemEnv(properties);
  48.                 MixAll.properties2Object(properties, brokerConfig);
  49.                 MixAll.properties2Object(properties, nettyServerConfig);
  50.                 MixAll.properties2Object(properties, nettyClientConfig);
  51.                 MixAll.properties2Object(properties, messageStoreConfig);
  52.                 // 然后，设置Broker配置文件的路径
  53.                 BrokerPathConfigHelper.setBrokerConfigPath(file);
  54.                 in.close();
  55.             }
  56.         }
  57.         // 把命令行的参数，先弄成一个properties，然后设置到BrokerConfig中
  58.         MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), brokerConfig);
  59.         // RocketMQ home没设置，报错
  60.         if (null == brokerConfig.getRocketmqHome()) {
  61.             System.out.printf("Please set the %s variable in your environment to match the location of the RocketMQ installation", MixAll.ROCKETMQ_HOME_ENV);
  62.             System.exit(-2);
  63.         }
  64.         // 获取到配置文件中配置的Namesrv的地址
  65.         String namesrvAddr = brokerConfig.getNamesrvAddr();
  66.         if (null != namesrvAddr) {
  67.             try {
  68.                 // 根据分号进行分割namesrv的地址(因为可以配置多个)
  69.                 String[] addrArray = namesrvAddr.split(";");
  70.                 for (String addr : addrArray) {
  71.                     // 尝试着把他封装成一个SocketAddress，如果转化不了，就会抛出异常
  72.                     // 说明配置的ip不会，就直接退出
  73.                     RemotingUtil.string2SocketAddress(addr);
  74.                 }
  75.             } catch (Exception e) {
  76.                 System.out.printf(
  77.                     "The Name Server Address[%s] illegal, please set it as follows, \"127.0.0.1:9876;192.168.0.1:9876\"%n",
  78.                     namesrvAddr);
  79.                 System.exit(-3);
  80.             }
  81.         }
  82.         // 在上面，已经把配置文件中的值，设置到配置文件中的，根据配置文件中配置的角色，设置BrokerId
  83.         switch (messageStoreConfig.getBrokerRole()) {
  84.             case ASYNC_MASTER:
  85.             case SYNC_MASTER:
  86.                 brokerConfig.setBrokerId(MixAll.MASTER_ID);
  87.                 break;
  88.             case SLAVE:
  89.                 if (brokerConfig.getBrokerId() <= 0) {
  90.                     System.out.printf("Slave's brokerId must be > 0");
  91.                     System.exit(-3);
  92.                 }
  94.                 break;
  95.             default:
  96.                 break;
  97.         }
  99.         if (messageStoreConfig.isEnableDLegerCommitLog()) {
  100.             brokerConfig.setBrokerId(-1);
  101.         }
  103.         messageStoreConfig.setHaListenPort(nettyServerConfig.getListenPort() + 1);
  104.         // 获取上下文，读取配置文件
  105.         LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
  106.         JoranConfigurator configurator = new JoranConfigurator();
  107.         configurator.setContext(lc);
  108.         lc.reset();
  109.         configurator.doConfigure(brokerConfig.getRocketmqHome() + "/conf/logback_broker.xml");
  110.         // 根据命令行设置的东西，读取到配置类中
  111.         if (commandLine.hasOption('p')) {
  112.             InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.BROKER_CONSOLE_NAME);
  113.             MixAll.printObjectProperties(console, brokerConfig);
  114.             MixAll.printObjectProperties(console, nettyServerConfig);
  115.             MixAll.printObjectProperties(console, nettyClientConfig);
  116.             MixAll.printObjectProperties(console, messageStoreConfig);
  117.             System.exit(0);
  118.         } else if (commandLine.hasOption('m')) {
  119.             InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.BROKER_CONSOLE_NAME);
  120.             MixAll.printObjectProperties(console, brokerConfig, true);
  121.             MixAll.printObjectProperties(console, nettyServerConfig, true);
  122.             MixAll.printObjectProperties(console, nettyClientConfig, true);
  123.             MixAll.printObjectProperties(console, messageStoreConfig, true);
  124.             System.exit(0);
  125.         }
  126.         // 将配置信息，打印到日志中
  127.         log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
  128.         MixAll.printObjectProperties(log, brokerConfig);
  129.         MixAll.printObjectProperties(log, nettyServerConfig);
  130.         MixAll.printObjectProperties(log, nettyClientConfig);
  131.         MixAll.printObjectProperties(log, messageStoreConfig);
  132.         // 创建BrokerController
  133.         final BrokerController controller = new BrokerController(
  134.             brokerConfig,
  135.             nettyServerConfig,
  136.             nettyClientConfig,
  137.             messageStoreConfig);
  138.         // remember all configs to prevent discard
  139.         controller.getConfiguration().registerConfig(properties);
  141.         boolean initResult = controller.initialize();
  142.         if (!initResult) {
  143.             controller.shutdown();
  144.             System.exit(-3);
  145.         }
  146.         // 添加一个Hook，用来关闭BrokerController
  147.         Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
  148.             private volatile boolean hasShutdown = false;
  149.             private AtomicInteger shutdownTimes = new AtomicInteger(0);
  151.             @Override
  152.             public void run() {
  153.                 synchronized (this) {
  154.                     log.info("Shutdown hook was invoked, {}", this.shutdownTimes.incrementAndGet());
  155.                     if (!this.hasShutdown) {
  156.                         this.hasShutdown = true;
  157.                         long beginTime = System.currentTimeMillis();
  158.                         controller.shutdown();
  159.                         long consumingTimeTotal = System.currentTimeMillis() - beginTime;
  160.                         log.info("Shutdown hook over, consuming total time(ms): {}", consumingTimeTotal);
  161.                     }
  162.                 }
  163.             }
  164.         }, "ShutdownHook"));
  166.         return controller;
  167.     } catch (Throwable e) {
  168.         e.printStackTrace();
  169.         System.exit(-1);
  170.     }
  172.     return null;
  173. }

controller.start()

  1. public void start() throws Exception {
  2.     // 启动消息持久化，启动Netty服务等等等等，一堆启动
  3.     if (this.messageStore != null) {
  4.         this.messageStore.start();
  5.     }
  7.     if (this.remotingServer != null) {
  8.         this.remotingServer.start();
  9.     }
  11.     if (this.fastRemotingServer != null) {
  12.         this.fastRemotingServer.start();
  13.     }
  15.     if (this.fileWatchService != null) {
  16.         this.fileWatchService.start();
  17.     }
  18.     // 
  19.     if (this.brokerOuterAPI != null) {
  20.         this.brokerOuterAPI.start();
  21.     }
  23.     if (this.pullRequestHoldService != null) {
  24.         this.pullRequestHoldService.start();
  25.     }
  27.     if (this.clientHousekeepingService != null) {
  28.         this.clientHousekeepingService.start();
  29.     }
  31.     if (this.filterServerManager != null) {
  32.         this.filterServerManager.start();
  33.     }
  35.     if (!messageStoreConfig.isEnableDLegerCommitLog()) {
  36.         startProcessorByHa(messageStoreConfig.getBrokerRole());
  37.         handleSlaveSynchronize(messageStoreConfig.getBrokerRole());
  38.         this.registerBrokerAll(true, false, true);
  39.     }
  40.     // 定时线程池，定时将Broker注册到Namesrv上
  41.     // 通过BrokerOuterAPI，与Namesrv交互，注册Broker
  42.     this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
  44.         @Override
  45.         public void run() {
  46.             try {
  47.                 BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister());
  48.             } catch (Throwable e) {
  49.                 log.error("registerBrokerAll Exception", e);
  50.             }
  51.         }
  52.     }, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), TimeUnit.MILLISECONDS);
  54.     if (this.brokerStatsManager != null) {
  55.         this.brokerStatsManager.start();
  56.     }
  57.     // 定时清理过期的请求
  58.     if (this.brokerFastFailure != null) {
  59.         this.brokerFastFailure.start();
  60.     }
  63. }

打印日志，结束