RocketMQ源码分析【三】Broker启动流程源码分析

曾经终败给现在 2022-09-12 12:54 432阅读 0赞

文章目录

    • 入口
    • 核心流程
    • BrokerController是如何被创建的
    • Broker是如何注册到NameServer的
      • registerBrokerAll
      • doRegisterBrokerAll
      • registerBrokerAll()
    • 总结

入口

Broker启动的源码入口是BrokerStartup.java
在这里插入图片描述

核心流程

主要流程和NameServer启动有些类似

  1. 创建BrokerController

  2. 启动BrokerController

    public static void main(String[] args) { 

    1.     start(createBrokerController(args));
    2. }
    4. public static BrokerController start(BrokerController controller) { 
    5.     try { 
    7.         controller.start();
    8.         String tip = "The broker[" + controller.getBrokerConfig().getBrokerName() + ", "
    9.             + controller.getBrokerAddr() + "] boot success. serializeType=" + RemotingCommand.getSerializeTypeConfigInThisServer();
    11.         if (null != controller.getBrokerConfig().getNamesrvAddr()) { 
    12.             tip += " and name server is " + controller.getBrokerConfig().getNamesrvAddr();
    13.         }
    14.         log.info(tip);
    15.         System.out.printf("%s%n", tip);
    16.         return controller;
    17.     } catch (Throwable e) { 
    18.         e.printStackTrace();
    19.         System.exit(-1);
    20.     }
    22.     return null;
    23. }

BrokerController是如何被创建的

  1. public static BrokerController createBrokerController(String[] args) { 
  2.         System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
  3.         if (null == System.getProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_SOCKET_SNDBUF_SIZE)) { 
  4.             NettySystemConfig.socketSndbufSize = 131072;
  5.         }
  6.         // 设置socket 发送缓冲大小
  7.         if (null == System.getProperty(NettySystemConfig.COM_ROCKETMQ_REMOTING_SOCKET_RCVBUF_SIZE)) { 
  8.             NettySystemConfig.socketRcvbufSize = 131072;
  9.         }
  11.         try { 
  12.             //PackageConflictDetect.detectFastjson();
  13.             // 解析启动参数 args
  14.             Options options = ServerUtil.buildCommandlineOptions(new Options());
  15.             commandLine = ServerUtil.parseCmdLine("mqbroker", args, buildCommandlineOptions(options),
  16.                 new PosixParser());
  17.             if (null == commandLine) { 
  18.                 System.exit(-1);
  19.             }
  20.             // Broker核心配置类
  21.             final BrokerConfig brokerConfig = new BrokerConfig();
  22.             // Netty服务器配置
  23.             final NettyServerConfig nettyServerConfig = new NettyServerConfig();
  24.             // Netty客户端配置TLS
  25.             final NettyClientConfig nettyClientConfig = new NettyClientConfig();
  26.             // 设置Netty客户端是否使用
  27.             nettyClientConfig.setUseTLS(Boolean.parseBoolean(System.getProperty(TLS_ENABLE,
  28.                 String.valueOf(TlsSystemConfig.tlsMode == TlsMode.ENFORCING))));
  29.             // 设置Netty服务器监听端口为 10911
  30.             nettyServerConfig.setListenPort(10911);
  31.             // Broker 消息配置类
  32.             final MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
  33.             // broker是否为 slave broker 如果是 则将 消息占用内存百度分 减10%, 默认40%,超过内存的消息将置换出内存
  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);
  47.                     properties2SystemEnv(properties);
  48.                     MixAll.properties2Object(properties, brokerConfig);
  49.                     MixAll.properties2Object(properties, nettyServerConfig);
  50.                     MixAll.properties2Object(properties, nettyClientConfig);
  51.                     MixAll.properties2Object(properties, messageStoreConfig);
  53.                     BrokerPathConfigHelper.setBrokerConfigPath(file);
  54.                     in.close();
  55.                 }
  56.             }
  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.             // 获取Nameserver地址，;分割解析为数组 因为Nameserver可能为集群，所以有多个
  65.             String namesrvAddr = brokerConfig.getNamesrvAddr();
  66.             if (null != namesrvAddr) { 
  67.                 try { 
  68.                     String[] addrArray = namesrvAddr.split(";");
  69.                     for (String addr : addrArray) { 
  70.                         RemotingUtil.string2SocketAddress(addr);
  71.                     }
  72.                 } catch (Exception e) { 
  73.                     System.out.printf(
  74.                         "The Name Server Address[%s] illegal, please set it as follows, \"127.0.0.1:9876;192.168.0.1:9876\"%n",
  75.                         namesrvAddr);
  76.                     System.exit(-3);
  77.                 }
  78.             }
  79.             // 判断Broker角色，作相应的处理 master有两种同步消息方式
  80.             switch (messageStoreConfig.getBrokerRole()) { 
  81.                 // 异步复制:生产者写入消息到Master后无需等待消息复制到slave即可返回，消息的复制由旁路线程进行异步复制
  82.                 case ASYNC_MASTER:
  83.                 /* 同步复制的方式，表现出来的是类似同步双写的策略。即Master写入完消息之后，需要等待Slave的复制成功。 注，这里只需要有一个Slave复制成功并成功应答即算成功，所以在这种模式下，如果有3个Slave，当生产者获得SEND_OK的应答时， 代表消息已经达到Maser和一个Slave（注：这里并不代表已经持久化到磁盘，而只能证明肯定到了PageCache， 是否能刷到磁盘取决于刷盘策略是同步刷盘还是异步刷盘），而还有两个Slave实际上是无法保证的， 并且这里也不支持配置，即不支持如“”同步半数以上”之类的设置 */
  84.                 case SYNC_MASTER:
  85.                     brokerConfig.setBrokerId(MixAll.MASTER_ID);
  86.                     break;
  87.                 /** * 消息发送的状态除了SEND_OK外，还会多出以下的状态： * FLUSH_SLAVE_TIMEOUT ： 同步到slave等待超时，即一直等Slave上报同步的进度，但过了超时时间都没有成功没有同步完。 * SLAVE_NOT_AVAILABLE：当前没有可用的Slave。注：如果slave落后master实在太多， * 那个slave也会认为是暂时不可用的slave，直到它同步到接近的范围为止， * 这个不可用的阈值由broker配置haSlaveFallbehindMax（默认是1024 * 1024 * 256）决定 */
  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.             }
  98.             // 是否基于dleger技术管理主从同步和CommitLog
  99.             if (messageStoreConfig.isEnableDLegerCommitLog()) { 
  100.                 brokerConfig.setBrokerId(-1);
  101.             }
  102.             // 设置HA监听端口
  103.             messageStoreConfig.setHaListenPort(nettyServerConfig.getListenPort() + 1);
  105.             // 日志相关
  106.             LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
  107.             JoranConfigurator configurator = new JoranConfigurator();
  108.             configurator.setContext(lc);
  109.             lc.reset();
  110.             configurator.doConfigure(brokerConfig.getRocketmqHome() + "/conf/logback_broker.xml");
  112.             // 打印启动参数信息
  113.             if (commandLine.hasOption('p')) { 
  114.                 InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.BROKER_CONSOLE_NAME);
  115.                 MixAll.printObjectProperties(console, brokerConfig);
  116.                 MixAll.printObjectProperties(console, nettyServerConfig);
  117.                 MixAll.printObjectProperties(console, nettyClientConfig);
  118.                 MixAll.printObjectProperties(console, messageStoreConfig);
  119.                 System.exit(0);
  120.             } else if (commandLine.hasOption('m')) { 
  121.                 InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.BROKER_CONSOLE_NAME);
  122.                 MixAll.printObjectProperties(console, brokerConfig, true);
  123.                 MixAll.printObjectProperties(console, nettyServerConfig, true);
  124.                 MixAll.printObjectProperties(console, nettyClientConfig, true);
  125.                 MixAll.printObjectProperties(console, messageStoreConfig, true);
  126.                 System.exit(0);
  127.             }
  129.             log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
  130.             MixAll.printObjectProperties(log, brokerConfig);
  131.             MixAll.printObjectProperties(log, nettyServerConfig);
  132.             MixAll.printObjectProperties(log, nettyClientConfig);
  133.             MixAll.printObjectProperties(log, messageStoreConfig);
  134.             // 基于上面的配置参数 创建 BrokerController
  135.             final BrokerController controller = new BrokerController(
  136.                 brokerConfig,
  137.                 nettyServerConfig,
  138.                 nettyClientConfig,
  139.                 messageStoreConfig);
  140.             // remember all configs to prevent discard
  141.             controller.getConfiguration().registerConfig(properties);
  142.             // 初始化 BrokerController
  143.             boolean initResult = controller.initialize();
  144.             if (!initResult) { 
  145.                 controller.shutdown();
  146.                 System.exit(-3);
  147.             }
  149.             // 钩子线程 JVM退出时打印消费时间
  150.             Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() { 
  151.                 private volatile boolean hasShutdown = false;
  152.                 private AtomicInteger shutdownTimes = new AtomicInteger(0);
  154.                 @Override
  155.                 public void run() { 
  156.                     synchronized (this) { 
  157.                         log.info("Shutdown hook was invoked, {}", this.shutdownTimes.incrementAndGet());
  158.                         if (!this.hasShutdown) { 
  159.                             this.hasShutdown = true;
  160.                             long beginTime = System.currentTimeMillis();
  161.                             controller.shutdown();
  162.                             long consumingTimeTotal = System.currentTimeMillis() - beginTime;
  163.                             log.info("Shutdown hook over, consuming total time(ms): {}", consumingTimeTotal);
  164.                         }
  165.                     }
  166.                 }
  167.             }, "ShutdownHook"));
  169.             return controller;
  170.         } catch (Throwable e) { 
  171.             e.printStackTrace();
  172.             System.exit(-1);
  173.         }
  175.         return null;
  176.     }

上面代码可以看到主要是读取一些配置文件，根据配置文件设置不同的一些枚举状态，然后在上面代码中又有一个比较核心的方法

  1. boolean initResult = controller.initialize();

我们进去看看

下面代码很长，可以简单过一下，这里先作一个总结吧
下面的代码主要是配置启动一些线程池，然后作内存持久化到磁盘的操作，然后就是一些Netty的配置，大致就是干了这些事。这里不用太注意细节，源码每个点都看懂太难了，我们主要是挑核心的看，下面代码我也作了一些注释

  1. public boolean initialize() throws CloneNotSupportedException { 
  3.         // 从磁盘中初始化 topicConfigManager
  4.         boolean result = this.topicConfigManager.load();
  5.         // 加载消费偏移
  6.         result = result && this.consumerOffsetManager.load();
  7.         // 加载消费组信息
  8.         result = result && this.subscriptionGroupManager.load();
  9.         // 加载过滤器配置
  10.         result = result && this.consumerFilterManager.load();
  12.         if (result) { 
  13.             try { 
  14.                 // 创建消息存储管理组件,默认使用commitLog 管理磁盘消息
  15.                 this.messageStore =
  16.                     new DefaultMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener,
  17.                         this.brokerConfig);
  18.                 // 如果使用Dleger 管理 commitlog 初始化相关信息
  19.                 if (messageStoreConfig.isEnableDLegerCommitLog()) { 
  20.                     DLedgerRoleChangeHandler roleChangeHandler = new DLedgerRoleChangeHandler(this, (DefaultMessageStore) messageStore);
  21.                     ((DLedgerCommitLog)((DefaultMessageStore) messageStore).getCommitLog()).getdLedgerServer().getdLedgerLeaderElector().addRoleChangeHandler(roleChangeHandler);
  22.                 }
  23.                 this.brokerStats = new BrokerStats((DefaultMessageStore) this.messageStore);
  24.                 //load plugin
  25.                 MessageStorePluginContext context = new MessageStorePluginContext(messageStoreConfig, brokerStatsManager, messageArrivingListener, brokerConfig);
  26.                 this.messageStore = MessageStoreFactory.build(context, this.messageStore);
  27.                 this.messageStore.getDispatcherList().addFirst(new CommitLogDispatcherCalcBitMap(this.brokerConfig, this.consumerFilterManager));
  28.             } catch (IOException e) { 
  29.                 result = false;
  30.                 log.error("Failed to initialize", e);
  31.             }
  32.         }
  34.         result = result && this.messageStore.load();
  36.         if (result) { 
  37.             // 启动Netty服务器
  38.             this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.clientHousekeepingService);
  39.             NettyServerConfig fastConfig = (NettyServerConfig) this.nettyServerConfig.clone();
  40.             fastConfig.setListenPort(nettyServerConfig.getListenPort() - 2);
  41.             this.fastRemotingServer = new NettyRemotingServer(fastConfig, this.clientHousekeepingService);
  42.             this.sendMessageExecutor = new BrokerFixedThreadPoolExecutor(
  43.                 this.brokerConfig.getSendMessageThreadPoolNums(),
  44.                 this.brokerConfig.getSendMessageThreadPoolNums(),
  45.                 1000 * 60,
  46.                 TimeUnit.MILLISECONDS,
  47.                 this.sendThreadPoolQueue,
  48.                 new ThreadFactoryImpl("SendMessageThread_"));
  50.             this.pullMessageExecutor = new BrokerFixedThreadPoolExecutor(
  51.                 this.brokerConfig.getPullMessageThreadPoolNums(),
  52.                 this.brokerConfig.getPullMessageThreadPoolNums(),
  53.                 1000 * 60,
  54.                 TimeUnit.MILLISECONDS,
  55.                 this.pullThreadPoolQueue,
  56.                 new ThreadFactoryImpl("PullMessageThread_"));
  58.             this.replyMessageExecutor = new BrokerFixedThreadPoolExecutor(
  59.                 this.brokerConfig.getProcessReplyMessageThreadPoolNums(),
  60.                 this.brokerConfig.getProcessReplyMessageThreadPoolNums(),
  61.                 1000 * 60,
  62.                 TimeUnit.MILLISECONDS,
  63.                 this.replyThreadPoolQueue,
  64.                 new ThreadFactoryImpl("ProcessReplyMessageThread_"));
  66.             this.queryMessageExecutor = new BrokerFixedThreadPoolExecutor(
  67.                 this.brokerConfig.getQueryMessageThreadPoolNums(),
  68.                 this.brokerConfig.getQueryMessageThreadPoolNums(),
  69.                 1000 * 60,
  70.                 TimeUnit.MILLISECONDS,
  71.                 this.queryThreadPoolQueue,
  72.                 new ThreadFactoryImpl("QueryMessageThread_"));
  74.             this.adminBrokerExecutor =
  75.                 Executors.newFixedThreadPool(this.brokerConfig.getAdminBrokerThreadPoolNums(), new ThreadFactoryImpl(
  76.                     "AdminBrokerThread_"));
  78.             this.clientManageExecutor = new ThreadPoolExecutor(
  79.                 this.brokerConfig.getClientManageThreadPoolNums(),
  80.                 this.brokerConfig.getClientManageThreadPoolNums(),
  81.                 1000 * 60,
  82.                 TimeUnit.MILLISECONDS,
  83.                 this.clientManagerThreadPoolQueue,
  84.                 new ThreadFactoryImpl("ClientManageThread_"));
  86.             this.heartbeatExecutor = new BrokerFixedThreadPoolExecutor(
  87.                 this.brokerConfig.getHeartbeatThreadPoolNums(),
  88.                 this.brokerConfig.getHeartbeatThreadPoolNums(),
  89.                 1000 * 60,
  90.                 TimeUnit.MILLISECONDS,
  91.                 this.heartbeatThreadPoolQueue,
  92.                 new ThreadFactoryImpl("HeartbeatThread_", true));
  94.             this.endTransactionExecutor = new BrokerFixedThreadPoolExecutor(
  95.                 this.brokerConfig.getEndTransactionThreadPoolNums(),
  96.                 this.brokerConfig.getEndTransactionThreadPoolNums(),
  97.                 1000 * 60,
  98.                 TimeUnit.MILLISECONDS,
  99.                 this.endTransactionThreadPoolQueue,
  100.                 new ThreadFactoryImpl("EndTransactionThread_"));
  102.             this.consumerManageExecutor =
  103.                 Executors.newFixedThreadPool(this.brokerConfig.getConsumerManageThreadPoolNums(), new ThreadFactoryImpl(
  104.                     "ConsumerManageThread_"));
  106.             this.registerProcessor();
  108.             final long initialDelay = UtilAll.computeNextMorningTimeMillis() - System.currentTimeMillis();
  109.             final long period = 1000 * 60 * 60 * 24;
  110.             // 定时记录broker状态 默认 86400s 一次
  111.             this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
  112.                 @Override
  113.                 public void run() { 
  114.                     try { 
  115.                         BrokerController.this.getBrokerStats().record();
  116.                     } catch (Throwable e) { 
  117.                         log.error("schedule record error.", e);
  118.                     }
  119.                 }
  120.             }, initialDelay, period, TimeUnit.MILLISECONDS);
  121.             // 定时持久化 offset, 5s 持久化一次
  122.             this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
  123.                 @Override
  124.                 public void run() { 
  125.                     try { 
  126.                         BrokerController.this.consumerOffsetManager.persist();
  127.                     } catch (Throwable e) { 
  128.                         log.error("schedule persist consumerOffset error.", e);
  129.                     }
  130.                 }
  131.             }, 1000 * 10, this.brokerConfig.getFlushConsumerOffsetInterval(), TimeUnit.MILLISECONDS);
  132.             // 定时持久化filter, 10s同步一次
  133.             this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
  134.                 @Override
  135.                 public void run() { 
  136.                     try { 
  137.                         BrokerController.this.consumerFilterManager.persist();
  138.                     } catch (Throwable e) { 
  139.                         log.error("schedule persist consumer filter error.", e);
  140.                     }
  141.                 }
  142.             }, 1000 * 10, 1000 * 10, TimeUnit.MILLISECONDS);
  144.             this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
  145.                 @Override
  146.                 public void run() { 
  147.                     try { 
  148.                         BrokerController.this.protectBroker();
  149.                     } catch (Throwable e) { 
  150.                         log.error("protectBroker error.", e);
  151.                     }
  152.                 }
  153.             }, 3, 3, TimeUnit.MINUTES);
  155.             this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
  156.                 @Override
  157.                 public void run() { 
  158.                     try { 
  159.                         BrokerController.this.printWaterMark();
  160.                     } catch (Throwable e) { 
  161.                         log.error("printWaterMark error.", e);
  162.                     }
  163.                 }
  164.             }, 10, 1, TimeUnit.SECONDS);
  166.             this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
  168.                 @Override
  169.                 public void run() { 
  170.                     try { 
  171.                         log.info("dispatch behind commit log {} bytes", BrokerController.this.getMessageStore().dispatchBehindBytes());
  172.                     } catch (Throwable e) { 
  173.                         log.error("schedule dispatchBehindBytes error.", e);
  174.                     }
  175.                 }
  176.             }, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);
  178.             if (this.brokerConfig.getNamesrvAddr() != null) { 
  179.                 this.brokerOuterAPI.updateNameServerAddressList(this.brokerConfig.getNamesrvAddr());
  180.                 log.info("Set user specified name server address: {}", this.brokerConfig.getNamesrvAddr());
  181.             } else if (this.brokerConfig.isFetchNamesrvAddrByAddressServer()) { 
  182.                 this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
  184.                     @Override
  185.                     public void run() { 
  186.                         try { 
  187.                             BrokerController.this.brokerOuterAPI.fetchNameServerAddr();
  188.                         } catch (Throwable e) { 
  189.                             log.error("ScheduledTask fetchNameServerAddr exception", e);
  190.                         }
  191.                     }
  192.                 }, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS);
  193.             }
  195.             if (!messageStoreConfig.isEnableDLegerCommitLog()) { 
  196.                 if (BrokerRole.SLAVE == this.messageStoreConfig.getBrokerRole()) { 
  197.                     if (this.messageStoreConfig.getHaMasterAddress() != null && this.messageStoreConfig.getHaMasterAddress().length() >= 6) { 
  198.                         this.messageStore.updateHaMasterAddress(this.messageStoreConfig.getHaMasterAddress());
  199.                         this.updateMasterHAServerAddrPeriodically = false;
  200.                     } else { 
  201.                         this.updateMasterHAServerAddrPeriodically = true;
  202.                     }
  203.                 } else { 
  204.                     this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
  205.                         @Override
  206.                         public void run() { 
  207.                             try { 
  208.                                 BrokerController.this.printMasterAndSlaveDiff();
  209.                             } catch (Throwable e) { 
  210.                                 log.error("schedule printMasterAndSlaveDiff error.", e);
  211.                             }
  212.                         }
  213.                     }, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);
  214.                 }
  215.             }
  217.             if (TlsSystemConfig.tlsMode != TlsMode.DISABLED) { 
  218.                 // Register a listener to reload SslContext
  219.                 try { 
  220.                     fileWatchService = new FileWatchService(
  221.                         new String[] { 
  222.                             TlsSystemConfig.tlsServerCertPath,
  223.                             TlsSystemConfig.tlsServerKeyPath,
  224.                             TlsSystemConfig.tlsServerTrustCertPath
  225.                         },
  226.                         new FileWatchService.Listener() { 
  227.                             boolean certChanged, keyChanged = false;
  229.                             @Override
  230.                             public void onChanged(String path) { 
  231.                                 if (path.equals(TlsSystemConfig.tlsServerTrustCertPath)) { 
  232.                                     log.info("The trust certificate changed, reload the ssl context");
  233.                                     reloadServerSslContext();
  234.                                 }
  235.                                 if (path.equals(TlsSystemConfig.tlsServerCertPath)) { 
  236.                                     certChanged = true;
  237.                                 }
  238.                                 if (path.equals(TlsSystemConfig.tlsServerKeyPath)) { 
  239.                                     keyChanged = true;
  240.                                 }
  241.                                 if (certChanged && keyChanged) { 
  242.                                     log.info("The certificate and private key changed, reload the ssl context");
  243.                                     certChanged = keyChanged = false;
  244.                                     reloadServerSslContext();
  245.                                 }
  246.                             }
  248.                             private void reloadServerSslContext() { 
  249.                                 ((NettyRemotingServer) remotingServer).loadSslContext();
  250.                                 ((NettyRemotingServer) fastRemotingServer).loadSslContext();
  251.                             }
  252.                         });
  253.                 } catch (Exception e) { 
  254.                     log.warn("FileWatchService created error, can't load the certificate dynamically");
  255.                 }
  256.             }
  257.             initialTransaction();
  258.             initialAcl();
  259.             initialRpcHooks();
  260.         }
  261.         return result;
  262.     }

Broker是如何注册到NameServer的

比较关键的一个点就是Broker是如何注册到NameServer的,我们来看看这段代码吧
在创建出BrokerController后调用了start方法
在这里插入图片描述
我们进入这个方法会看到一个定时线程池

  1. this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
  3.             @Override
  4.             public void run() { 
  5.                 try { 
  6.                     BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister());
  7.                 } catch (Throwable e) { 
  8.                     log.error("registerBrokerAll Exception", e);
  9.                 }
  10.             }
  11.         }, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), TimeUnit.MILLISECONDS);

registerBrokerAll

然后进入方法registerBrokerAll

  1. BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister());
  3. public synchronized void registerBrokerAll(final boolean checkOrderConfig, boolean oneway, boolean forceRegister) { 
  4.         // Topic 相关配置信息
  5.         TopicConfigSerializeWrapper topicConfigWrapper = this.getTopicConfigManager().buildTopicConfigSerializeWrapper();
  6.         // 处理 TopicConfig信息
  7.         if (!PermName.isWriteable(this.getBrokerConfig().getBrokerPermission())
  8.             || !PermName.isReadable(this.getBrokerConfig().getBrokerPermission())) { 
  9.             ConcurrentHashMap<String, TopicConfig> topicConfigTable = new ConcurrentHashMap<String, TopicConfig>();
  10.             for (TopicConfig topicConfig : topicConfigWrapper.getTopicConfigTable().values()) { 
  11.                 TopicConfig tmp =
  12.                     new TopicConfig(topicConfig.getTopicName(), topicConfig.getReadQueueNums(), topicConfig.getWriteQueueNums(),
  13.                         this.brokerConfig.getBrokerPermission());
  14.                 topicConfigTable.put(topicConfig.getTopicName(), tmp);
  15.             }
  16.             topicConfigWrapper.setTopicConfigTable(topicConfigTable);
  17.         }
  18.         // 判断是否进行Broker注册
  19.         if (forceRegister || needRegister(this.brokerConfig.getBrokerClusterName(),
  20.             this.getBrokerAddr(),
  21.             this.brokerConfig.getBrokerName(),
  22.             this.brokerConfig.getBrokerId(),
  23.             this.brokerConfig.getRegisterBrokerTimeoutMills())) { 
  24.             // 真正注册Broker关键入口
  25.             doRegisterBrokerAll(checkOrderConfig, oneway, topicConfigWrapper);
  26.         }
  27.     }

doRegisterBrokerAll

我们进入到doRegisterBrokerAll看看，这下面就是注册Broker到NameServer的核心代码，涉及到Netty网络请求相关的代码被封装在registerBrokerAll()方法中，这段代码看完了我们可以进去看看

  1. private void doRegisterBrokerAll(boolean checkOrderConfig, boolean oneway,
  2.         TopicConfigSerializeWrapper topicConfigWrapper) { 
  4.         // 将Broker 注册到所有NameServer上，NameServer可能是集群有多个，所以这里是List
  5.         List<RegisterBrokerResult> registerBrokerResultList = this.brokerOuterAPI.registerBrokerAll(
  6.             this.brokerConfig.getBrokerClusterName(),
  7.             this.getBrokerAddr(),
  8.             this.brokerConfig.getBrokerName(),
  9.             this.brokerConfig.getBrokerId(),
  10.             this.getHAServerAddr(),
  11.             topicConfigWrapper,
  12.             this.filterServerManager.buildNewFilterServerList(),
  13.             oneway,
  14.             this.brokerConfig.getRegisterBrokerTimeoutMills(),
  15.             this.brokerConfig.isCompressedRegister());
  17.         // 如果注册结果大于 0 则继续处理相关逻辑
  18.         if (registerBrokerResultList.size() > 0) { 
  19.             RegisterBrokerResult registerBrokerResult = registerBrokerResultList.get(0);
  20.             if (registerBrokerResult != null) { 
  21.                 if (this.updateMasterHAServerAddrPeriodically && registerBrokerResult.getHaServerAddr() != null) { 
  22.                     this.messageStore.updateHaMasterAddress(registerBrokerResult.getHaServerAddr());
  23.                 }
  25.                 this.slaveSynchronize.setMasterAddr(registerBrokerResult.getMasterAddr());
  27.                 if (checkOrderConfig) { 
  28.                     this.getTopicConfigManager().updateOrderTopicConfig(registerBrokerResult.getKvTable());
  29.                 }
  30.             }
  31.         }
  32.     }

registerBrokerAll()

这里我们来看看registerBrokerAll()方法

  1. public List<RegisterBrokerResult> registerBrokerAll(
  2.         final String clusterName,
  3.         final String brokerAddr,
  4.         final String brokerName,
  5.         final long brokerId,
  6.         final String haServerAddr,
  7.         final TopicConfigSerializeWrapper topicConfigWrapper,
  8.         final List<String> filterServerList,
  9.         final boolean oneway,
  10.         final int timeoutMills,
  11.         final boolean compressed) { 
  13.         // 用于存放每个远程注册的返回结果
  14.         final List<RegisterBrokerResult> registerBrokerResultList = new CopyOnWriteArrayList<>();
  16.         // NameServer的注册地址
  17.         List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
  18.         if (nameServerAddressList != null && nameServerAddressList.size() > 0) { 
  21.             // 构建网络请求头
  22.             final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();
  23.             requestHeader.setBrokerAddr(brokerAddr);
  24.             requestHeader.setBrokerId(brokerId);
  25.             requestHeader.setBrokerName(brokerName);
  26.             requestHeader.setClusterName(clusterName);
  27.             requestHeader.setHaServerAddr(haServerAddr);
  28.             requestHeader.setCompressed(compressed);
  30.             // 构建网络请求 body
  31.             RegisterBrokerBody requestBody = new RegisterBrokerBody();
  32.             requestBody.setTopicConfigSerializeWrapper(topicConfigWrapper);
  33.             requestBody.setFilterServerList(filterServerList);
  34.             final byte[] body = requestBody.encode(compressed);
  35.             final int bodyCrc32 = UtilAll.crc32(body);
  36.             requestHeader.setBodyCrc32(bodyCrc32);
  38.             // 多线程注册提高效率
  39.             final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());
  40.             for (final String namesrvAddr : nameServerAddressList) { 
  41.                 //自定义线程池
  42.                 brokerOuterExecutor.execute(() -> { 
  43.                     try { 
  44.                         RegisterBrokerResult result = registerBroker(namesrvAddr,oneway, timeoutMills,requestHeader,body);
  45.                         if (result != null) { 
  46.                             registerBrokerResultList.add(result);
  47.                         }
  49.                         log.info("register broker[{}]to name server {} OK", brokerId, namesrvAddr);
  50.                     } catch (Exception e) { 
  51.                         log.warn("registerBroker Exception, {}", namesrvAddr, e);
  52.                     } finally { 
  53.                         countDownLatch.countDown();
  54.                     }
  55.                 });
  56.             }
  58.             try { 
  59.                 countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS);
  60.             } catch (InterruptedException e) { 
  61.             }
  62.         }
  64.         return registerBrokerResultList;
  65.     }

总结

我们这里就把Broker的启动流程分析完了，我们这里并没有一行代码一行代码去看，因为有些代码不知道实际的业务场景，我们也不知道是干嘛的，总的来说看了整个源码的一个核心流程，相关细节并没有很深入，主要是对Broker的启动有了一个大致的了解，后续如果需要修改或者遇到了相关问题，可以很好的找到入库，问题排查也会更轻松

发表评论

表情：
评论列表 （有 0 条评论，432人围观）

还没有评论，来说两句吧...

相关阅读