RocketMQ源码解析二（Broker启动流程）

本是古典 何须时尚 2022-09-12 07:46 72阅读 0赞

RocketMQ版本4.6.0，记录自己看源码的过程

启动流程

broker入口在BrokerStartup类中main()方法

  1. public static void main(String[] args) { 
  2.     // 1、解析并创建相关配置，根据配置创建BrokerController组件
  3.     // 2、初始化BrokerController组件，，主要是创建一个netty服务器以及一些线程池
  4.     // 3、启动BrokerController中的各个功能组件，开启注册和发送心跳的定时任务
  5.     start(createBrokerController(args));
  6. }

嗯，看起来跟namesrv启动过程差不多。
接下来就看上面的3个步骤
首先是解析并创建BrokerController组件，也是有一个核心组件：

  1. /** * 解析并创建相关配置，根据配置创建BrokerController组件并初始化 */
  2. public static BrokerController createBrokerController(String[] args) { 
  3.     System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
  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.         //PackageConflictDetect.detectFastjson();
  16.         Options options = ServerUtil.buildCommandlineOptions(new Options());
  17.         commandLine = ServerUtil.parseCmdLine("mqbroker", args, buildCommandlineOptions(options),
  18.             new PosixParser());
  19.         if (null == commandLine) { 
  20.             System.exit(-1);
  21.         }
  23.         // Broker核心配置
  24.         final BrokerConfig brokerConfig = new BrokerConfig();
  25.         // Broker分别作为netty服务端（与生产者消费者通信）和客户端的配置（与namesrv通信）
  26.         final NettyServerConfig nettyServerConfig = new NettyServerConfig();
  27.         final NettyClientConfig nettyClientConfig = new NettyClientConfig();
  29.         nettyClientConfig.setUseTLS(Boolean.parseBoolean(System.getProperty(TLS_ENABLE,
  30.             String.valueOf(TlsSystemConfig.tlsMode == TlsMode.ENFORCING))));
  31.         // 设置netty服务器监听端口10911
  32.         nettyServerConfig.setListenPort(10911);
  33.         // 消息存储的相关配置
  34.         final MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
  36.         // 如果是slave，就设置一个accessMessageInMemoryMaxRatio属性，暂时不懂
  37.         if (BrokerRole.SLAVE == messageStoreConfig.getBrokerRole()) { 
  38.             int ratio = messageStoreConfig.getAccessMessageInMemoryMaxRatio() - 10;
  39.             messageStoreConfig.setAccessMessageInMemoryMaxRatio(ratio);
  40.         }
  42.         // 如果启动broker带了-c选项，读取-c后面的配置文件地址
  43.         if (commandLine.hasOption('c')) { 
  44.             String file = commandLine.getOptionValue('c');
  45.             if (file != null) { 
  46.                 configFile = file;
  47.                 InputStream in = new BufferedInputStream(new FileInputStream(file));
  48.                 properties = new Properties();
  49.                 properties.load(in);
  51.                 properties2SystemEnv(properties);
  52.                 MixAll.properties2Object(properties, brokerConfig);
  53.                 MixAll.properties2Object(properties, nettyServerConfig);
  54.                 MixAll.properties2Object(properties, nettyClientConfig);
  55.                 MixAll.properties2Object(properties, messageStoreConfig);
  57.                 BrokerPathConfigHelper.setBrokerConfigPath(file);
  58.                 in.close();
  59.             }
  60.         }
  61.         // 将配置解析到brokerConfig中
  62.         MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), brokerConfig);
  64.         // 没有ROCKETMQ_HOME这个环境变量，无法启动
  65.         if (null == brokerConfig.getRocketmqHome()) { 
  66.             System.out.printf("Please set the %s variable in your environment to match the location of the RocketMQ installation", MixAll.ROCKETMQ_HOME_ENV);
  67.             System.exit(-2);
  68.         }
  70.         // 读取配置的namesrvAddr，按；进行分割
  71.         String namesrvAddr = brokerConfig.getNamesrvAddr();
  72.         if (null != namesrvAddr) { 
  73.             try { 
  74.                 String[] addrArray = namesrvAddr.split(";");
  75.                 for (String addr : addrArray) { 
  76.                     RemotingUtil.string2SocketAddress(addr);
  77.                 }
  78.             } catch (Exception e) { 
  79.                 System.out.printf(
  80.                     "The Name Server Address[%s] illegal, please set it as follows, \"127.0.0.1:9876;192.168.0.1:9876\"%n",
  81.                     namesrvAddr);
  82.                 System.exit(-3);
  83.             }
  84.         }
  86.         // 根据角色判断
  87.         switch (messageStoreConfig.getBrokerRole()) { 
  88.             case ASYNC_MASTER:
  89.             case SYNC_MASTER:
  90.                 // master就设置id为0
  91.                 brokerConfig.setBrokerId(MixAll.MASTER_ID);
  92.                 break;
  93.             case SLAVE:
  94.                 // slave的id必须大于0
  95.                 if (brokerConfig.getBrokerId() <= 0) { 
  96.                     System.out.print("Slave's brokerId must be > 0");
  97.                     System.exit(-3);
  98.                 }
  100.                 break;
  101.             default:
  102.                 break;
  103.         }
  105.         // 如果配置了DLeger技术，则id设为-1
  106.         if (messageStoreConfig.isEnableDLegerCommitLog()) { 
  107.             brokerConfig.setBrokerId(-1);
  108.         }
  110.         // 设置HA监听端口
  111.         messageStoreConfig.setHaListenPort(nettyServerConfig.getListenPort() + 1);
  112.         LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
  113.         JoranConfigurator configurator = new JoranConfigurator();
  114.         configurator.setContext(lc);
  115.         lc.reset();
  116.         configurator.doConfigure(brokerConfig.getRocketmqHome() + "/conf/logback_broker.xml");
  118.         // 解析启动参数中的-p和-m选项
  119.         if (commandLine.hasOption('p')) { 
  120.             InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.BROKER_CONSOLE_NAME);
  121.             MixAll.printObjectProperties(console, brokerConfig);
  122.             MixAll.printObjectProperties(console, nettyServerConfig);
  123.             MixAll.printObjectProperties(console, nettyClientConfig);
  124.             MixAll.printObjectProperties(console, messageStoreConfig);
  125.             System.exit(0);
  126.         } else if (commandLine.hasOption('m')) { 
  127.             InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.BROKER_CONSOLE_NAME);
  128.             MixAll.printObjectProperties(console, brokerConfig, true);
  129.             MixAll.printObjectProperties(console, nettyServerConfig, true);
  130.             MixAll.printObjectProperties(console, nettyClientConfig, true);
  131.             MixAll.printObjectProperties(console, messageStoreConfig, true);
  132.             System.exit(0);
  133.         }
  135.         // 日志相关，跳过
  136.         log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
  137.         MixAll.printObjectProperties(log, brokerConfig);
  138.         MixAll.printObjectProperties(log, nettyServerConfig);
  139.         MixAll.printObjectProperties(log, nettyClientConfig);
  140.         MixAll.printObjectProperties(log, messageStoreConfig);
  142.         // 根据上面的4个配置创建BrokerController
  143.         final BrokerController controller = new BrokerController(
  144.             brokerConfig,
  145.             nettyServerConfig,
  146.             nettyClientConfig,
  147.             messageStoreConfig);
  148.         // remember all configs to prevent discard
  149.         controller.getConfiguration().registerConfig(properties);
  151.         // 初始化BrokerController组件，主要是创建一个netty服务器以及一些线程池
  152.         boolean initResult = controller.initialize();
  153.         if (!initResult) { 
  154.             controller.shutdown();
  155.             System.exit(-3);
  156.         }
  158.         Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() { 
  159.             private volatile boolean hasShutdown = false;
  160.             private AtomicInteger shutdownTimes = new AtomicInteger(0);
  162.             @Override
  163.             public void run() { 
  164.                 synchronized (this) { 
  165.                     log.info("Shutdown hook was invoked, {}", this.shutdownTimes.incrementAndGet());
  166.                     if (!this.hasShutdown) { 
  167.                         this.hasShutdown = true;
  168.                         long beginTime = System.currentTimeMillis();
  169.                         controller.shutdown();
  170.                         long consumingTimeTotal = System.currentTimeMillis() - beginTime;
  171.                         log.info("Shutdown hook over, consuming total time(ms): {}", consumingTimeTotal);
  172.                     }
  173.                 }
  174.             }
  175.         }, "ShutdownHook"));
  177.         return controller;
  178.     } catch (Throwable e) { 
  179.         e.printStackTrace();
  180.         System.exit(-1);
  181.     }
  183.     return null;
  184. }

也是有几个核心配置，BrokerConfig，NettyServerConfig，NettyClientConfig，MessageStoreConfig，其中BrokerConfig是broker自己本身的配置，NettyServerConfig是broker需要作为服务器端与生产者和消费端进行通信的配置，NettyClientConfig是broker作为客户端与namesrv通信的配置，至于里面的具体配置内容就不分析了，等具体用到了再说明。
然后根据上面的4个配置创建BrokerController：

  1. final BrokerController controller = new BrokerController(
  2.     brokerConfig,
  3.     nettyServerConfig,
  4.     nettyClientConfig,
  5.     messageStoreConfig);
  7. public BrokerController(
  8.     final BrokerConfig brokerConfig,
  9.     final NettyServerConfig nettyServerConfig,
  10.     final NettyClientConfig nettyClientConfig,
  11.     final MessageStoreConfig messageStoreConfig
  12. ) { 
  13.     this.brokerConfig = brokerConfig;
  14.     this.nettyServerConfig = nettyServerConfig;
  15.     this.nettyClientConfig = nettyClientConfig;
  16.     this.messageStoreConfig = messageStoreConfig;
  17.     // 创建broker各个功能对应的组件
  18.     this.consumerOffsetManager = new ConsumerOffsetManager(this);
  19.     this.topicConfigManager = new TopicConfigManager(this);
  20.     this.pullMessageProcessor = new PullMessageProcessor(this);
  21.     this.pullRequestHoldService = new PullRequestHoldService(this);
  22.     this.messageArrivingListener = new NotifyMessageArrivingListener(this.pullRequestHoldService);
  23.     this.consumerIdsChangeListener = new DefaultConsumerIdsChangeListener(this);
  24.     this.consumerManager = new ConsumerManager(this.consumerIdsChangeListener);
  25.     this.consumerFilterManager = new ConsumerFilterManager(this);
  26.     this.producerManager = new ProducerManager();
  27.     this.clientHousekeepingService = new ClientHousekeepingService(this);
  28.     this.broker2Client = new Broker2Client(this);
  29.     this.subscriptionGroupManager = new SubscriptionGroupManager(this);
  30.     this.brokerOuterAPI = new BrokerOuterAPI(nettyClientConfig);
  31.     this.filterServerManager = new FilterServerManager(this);
  33.     this.slaveSynchronize = new SlaveSynchronize(this);
  35.     // 创建各线程池的等待队列
  36.     this.sendThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getSendThreadPoolQueueCapacity());
  37.     this.pullThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getPullThreadPoolQueueCapacity());
  38.     this.replyThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getReplyThreadPoolQueueCapacity());
  39.     this.queryThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getQueryThreadPoolQueueCapacity());
  40.     this.clientManagerThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getClientManagerThreadPoolQueueCapacity());
  41.     this.consumerManagerThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getConsumerManagerThreadPoolQueueCapacity());
  42.     this.heartbeatThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getHeartbeatThreadPoolQueueCapacity());
  43.     this.endTransactionThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getEndTransactionPoolQueueCapacity());
  45.     this.brokerStatsManager = new BrokerStatsManager(this.brokerConfig.getBrokerClusterName());
  46.     this.setStoreHost(new InetSocketAddress(this.getBrokerConfig().getBrokerIP1(), this.getNettyServerConfig().getListenPort()));
  48.     this.brokerFastFailure = new BrokerFastFailure(this);
  49.     this.configuration = new Configuration(
  50.         log,
  51.         BrokerPathConfigHelper.getBrokerConfigPath(),
  52.         this.brokerConfig, this.nettyServerConfig, this.nettyClientConfig, this.messageStoreConfig
  53.     );
  54. }

在BrokerController的构造方法里会同时创建各个功能组件。
然后就需要对BrokerController进行初始化：

  1. boolean initResult = controller.initialize();
  3. /** * 对BrokerController进行初始化，主要是创建一个netty服务器以及一些线程池 */
  4. public boolean initialize() throws CloneNotSupportedException { 
  5.     // 加载配置
  6.     boolean result = this.topicConfigManager.load();
  8.     result = result && this.consumerOffsetManager.load();
  9.     result = result && this.subscriptionGroupManager.load();
  10.     result = result && this.consumerFilterManager.load();
  12.     if (result) { 
  13.         try { 
  14.             // 创建消息存储相关组件及服务
  15.             this.messageStore =
  16.                 new DefaultMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener,
  17.                     this.brokerConfig);
  18.             if (messageStoreConfig.isEnableDLegerCommitLog()) { 
  19.                 DLedgerRoleChangeHandler roleChangeHandler = new DLedgerRoleChangeHandler(this, (DefaultMessageStore) messageStore);
  20.                 ((DLedgerCommitLog)((DefaultMessageStore) messageStore).getCommitLog()).getdLedgerServer().getdLedgerLeaderElector().addRoleChangeHandler(roleChangeHandler);
  21.             }
  22.             this.brokerStats = new BrokerStats((DefaultMessageStore) this.messageStore);
  23.             //load plugin
  24.             MessageStorePluginContext context = new MessageStorePluginContext(messageStoreConfig, brokerStatsManager, messageArrivingListener, brokerConfig);
  25.             this.messageStore = MessageStoreFactory.build(context, this.messageStore);
  26.             this.messageStore.getDispatcherList().addFirst(new CommitLogDispatcherCalcBitMap(this.brokerConfig, this.consumerFilterManager));
  27.         } catch (IOException e) { 
  28.             result = false;
  29.             log.error("Failed to initialize", e);
  30.         }
  31.     }
  33.     // 加载消息存储相关信息
  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.         // 下面是一些处理不同请求的线程池
  43.         // 创建处理发送过来的消息的线程池
  44.         this.sendMessageExecutor = new BrokerFixedThreadPoolExecutor(
  45.             this.brokerConfig.getSendMessageThreadPoolNums(),
  46.             this.brokerConfig.getSendMessageThreadPoolNums(),
  47.             1000 * 60,
  48.             TimeUnit.MILLISECONDS,
  49.             this.sendThreadPoolQueue,
  50.             new ThreadFactoryImpl("SendMessageThread_"));
  51.         // 创建处理consumer过来拉取消息的线程池
  52.         this.pullMessageExecutor = new BrokerFixedThreadPoolExecutor(
  53.             this.brokerConfig.getPullMessageThreadPoolNums(),
  54.             this.brokerConfig.getPullMessageThreadPoolNums(),
  55.             1000 * 60,
  56.             TimeUnit.MILLISECONDS,
  57.             this.pullThreadPoolQueue,
  58.             new ThreadFactoryImpl("PullMessageThread_"));
  60.         // 回复消息的线程池
  61.         this.replyMessageExecutor = new BrokerFixedThreadPoolExecutor(
  62.             this.brokerConfig.getProcessReplyMessageThreadPoolNums(),
  63.             this.brokerConfig.getProcessReplyMessageThreadPoolNums(),
  64.             1000 * 60,
  65.             TimeUnit.MILLISECONDS,
  66.             this.replyThreadPoolQueue,
  67.             new ThreadFactoryImpl("ProcessReplyMessageThread_"));
  69.         // 查询消息的线程池
  70.         this.queryMessageExecutor = new BrokerFixedThreadPoolExecutor(
  71.             this.brokerConfig.getQueryMessageThreadPoolNums(),
  72.             this.brokerConfig.getQueryMessageThreadPoolNums(),
  73.             1000 * 60,
  74.             TimeUnit.MILLISECONDS,
  75.             this.queryThreadPoolQueue,
  76.             new ThreadFactoryImpl("QueryMessageThread_"));
  78.         // 管理broker一些命令执行的线程池
  79.         this.adminBrokerExecutor =
  80.             Executors.newFixedThreadPool(this.brokerConfig.getAdminBrokerThreadPoolNums(), new ThreadFactoryImpl(
  81.                 "AdminBrokerThread_"));
  83.         // 管理客户端的线程池
  84.         this.clientManageExecutor = new ThreadPoolExecutor(
  85.             this.brokerConfig.getClientManageThreadPoolNums(),
  86.             this.brokerConfig.getClientManageThreadPoolNums(),
  87.             1000 * 60,
  88.             TimeUnit.MILLISECONDS,
  89.             this.clientManagerThreadPoolQueue,
  90.             new ThreadFactoryImpl("ClientManageThread_"));
  92.         // 发送心跳的线程池
  93.         this.heartbeatExecutor = new BrokerFixedThreadPoolExecutor(
  94.             this.brokerConfig.getHeartbeatThreadPoolNums(),
  95.             this.brokerConfig.getHeartbeatThreadPoolNums(),
  96.             1000 * 60,
  97.             TimeUnit.MILLISECONDS,
  98.             this.heartbeatThreadPoolQueue,
  99.             new ThreadFactoryImpl("HeartbeatThread_", true));
  101.         // 结束事务的线程池
  102.         this.endTransactionExecutor = new BrokerFixedThreadPoolExecutor(
  103.             this.brokerConfig.getEndTransactionThreadPoolNums(),
  104.             this.brokerConfig.getEndTransactionThreadPoolNums(),
  105.             1000 * 60,
  106.             TimeUnit.MILLISECONDS,
  107.             this.endTransactionThreadPoolQueue,
  108.             new ThreadFactoryImpl("EndTransactionThread_"));
  110.         // 管理consumer的线程池
  111.         this.consumerManageExecutor =
  112.             Executors.newFixedThreadPool(this.brokerConfig.getConsumerManageThreadPoolNums(), new ThreadFactoryImpl(
  113.                 "ConsumerManageThread_"));
  115.         // 注册请求处理器和请求线程池
  116.         this.registerProcessor();
  118.         // 下面就是开启一些定时任务
  119.         final long initialDelay = UtilAll.computeNextMorningTimeMillis() - System.currentTimeMillis();
  120.         final long period = 1000 * 60 * 60 * 24;
  121.         // 定时进行broker统计的任务
  122.         this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
  123.             @Override
  124.             public void run() { 
  125.                 try { 
  126.                     BrokerController.this.getBrokerStats().record();
  127.                 } catch (Throwable e) { 
  128.                     log.error("schedule record error.", e);
  129.                 }
  130.             }
  131.         }, initialDelay, period, TimeUnit.MILLISECONDS);
  133.         // 定时持久化消费进度
  134.         this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
  135.             @Override
  136.             public void run() { 
  137.                 try { 
  138.                     BrokerController.this.consumerOffsetManager.persist();
  139.                 } catch (Throwable e) { 
  140.                     log.error("schedule persist consumerOffset error.", e);
  141.                 }
  142.             }
  143.         }, 1000 * 10, this.brokerConfig.getFlushConsumerOffsetInterval(), TimeUnit.MILLISECONDS);
  145.         this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
  146.             @Override
  147.             public void run() { 
  148.                 try { 
  149.                     BrokerController.this.consumerFilterManager.persist();
  150.                 } catch (Throwable e) { 
  151.                     log.error("schedule persist consumer filter error.", e);
  152.                 }
  153.             }
  154.         }, 1000 * 10, 1000 * 10, TimeUnit.MILLISECONDS);
  156.         this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
  157.             @Override
  158.             public void run() { 
  159.                 try { 
  160.                     BrokerController.this.protectBroker();
  161.                 } catch (Throwable e) { 
  162.                     log.error("protectBroker error.", e);
  163.                 }
  164.             }
  165.         }, 3, 3, TimeUnit.MINUTES);
  167.         this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
  168.             @Override
  169.             public void run() { 
  170.                 try { 
  171.                     BrokerController.this.printWaterMark();
  172.                 } catch (Throwable e) { 
  173.                     log.error("printWaterMark error.", e);
  174.                 }
  175.             }
  176.         }, 10, 1, TimeUnit.SECONDS);
  178.         this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
  180.             @Override
  181.             public void run() { 
  182.                 try { 
  183.                     log.info("dispatch behind commit log {} bytes", BrokerController.this.getMessageStore().dispatchBehindBytes());
  184.                 } catch (Throwable e) { 
  185.                     log.error("schedule dispatchBehindBytes error.", e);
  186.                 }
  187.             }
  188.         }, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);
  190.         if (this.brokerConfig.getNamesrvAddr() != null) { 
  191.             this.brokerOuterAPI.updateNameServerAddressList(this.brokerConfig.getNamesrvAddr());
  192.             log.info("Set user specified name server address: {}", this.brokerConfig.getNamesrvAddr());
  193.         } else if (this.brokerConfig.isFetchNamesrvAddrByAddressServer()) { 
  194.             this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
  196.                 @Override
  197.                 public void run() { 
  198.                     try { 
  199.                         BrokerController.this.brokerOuterAPI.fetchNameServerAddr();
  200.                     } catch (Throwable e) { 
  201.                         log.error("ScheduledTask fetchNameServerAddr exception", e);
  202.                     }
  203.                 }
  204.             }, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS);
  205.         }
  207.         if (!messageStoreConfig.isEnableDLegerCommitLog()) { 
  208.             if (BrokerRole.SLAVE == this.messageStoreConfig.getBrokerRole()) { 
  209.                 if (this.messageStoreConfig.getHaMasterAddress() != null && this.messageStoreConfig.getHaMasterAddress().length() >= 6) { 
  210.                     this.messageStore.updateHaMasterAddress(this.messageStoreConfig.getHaMasterAddress());
  211.                     this.updateMasterHAServerAddrPeriodically = false;
  212.                 } else { 
  213.                     this.updateMasterHAServerAddrPeriodically = true;
  214.                 }
  215.             } else { 
  216.                 this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
  217.                     @Override
  218.                     public void run() { 
  219.                         try { 
  220.                             BrokerController.this.printMasterAndSlaveDiff();
  221.                         } catch (Throwable e) { 
  222.                             log.error("schedule printMasterAndSlaveDiff error.", e);
  223.                         }
  224.                     }
  225.                 }, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);
  226.             }
  227.         }
  229.         if (TlsSystemConfig.tlsMode != TlsMode.DISABLED) { 
  230.             // Register a listener to reload SslContext
  231.             try { 
  232.                 fileWatchService = new FileWatchService(
  233.                     new String[] { 
  234.                         TlsSystemConfig.tlsServerCertPath,
  235.                         TlsSystemConfig.tlsServerKeyPath,
  236.                         TlsSystemConfig.tlsServerTrustCertPath
  237.                     },
  238.                     new FileWatchService.Listener() { 
  239.                         boolean certChanged, keyChanged = false;
  241.                         @Override
  242.                         public void onChanged(String path) { 
  243.                             if (path.equals(TlsSystemConfig.tlsServerTrustCertPath)) { 
  244.                                 log.info("The trust certificate changed, reload the ssl context");
  245.                                 reloadServerSslContext();
  246.                             }
  247.                             if (path.equals(TlsSystemConfig.tlsServerCertPath)) { 
  248.                                 certChanged = true;
  249.                             }
  250.                             if (path.equals(TlsSystemConfig.tlsServerKeyPath)) { 
  251.                                 keyChanged = true;
  252.                             }
  253.                             if (certChanged && keyChanged) { 
  254.                                 log.info("The certificate and private key changed, reload the ssl context");
  255.                                 certChanged = keyChanged = false;
  256.                                 reloadServerSslContext();
  257.                             }
  258.                         }
  260.                         private void reloadServerSslContext() { 
  261.                             ((NettyRemotingServer) remotingServer).loadSslContext();
  262.                             ((NettyRemotingServer) fastRemotingServer).loadSslContext();
  263.                         }
  264.                     });
  265.             } catch (Exception e) { 
  266.                 log.warn("FileWatchService created error, can't load the certificate dynamically");
  267.             }
  268.         }
  269.         initialTransaction();
  270.         initialAcl();
  271.         initialRpcHooks();
  272.     }
  273.     return result;
  274. }

代码比较长，相关的流程都写在注释里了，简单概括就是加载配置和消费进度，将消费进度从磁盘读到内存，创建消息存储相关组件及服务以及加载该服务，创建一个netty服务器，毕竟也要和生产者消费者通信，创建一些处理不同请求的线程池和处理器（这里就会有专门的线程池和处理器，不像namesrv就只有一个默认的，毕竟broker要处理的东西比较多），以及开启一些定时任务。
最后一步就是开启BrokerController：

  1. controller.start();
  3. /** * 启动各个组件 */
  4. public void start() throws Exception { 
  5.     if (this.messageStore != null) { 
  6.         // 启动存储组件里的各个后台功能，比如定时刷盘，indexFile服务等
  7.         this.messageStore.start();
  8.     }
  10.     if (this.remotingServer != null) { 
  11.         this.remotingServer.start();
  12.     }
  14.     if (this.fastRemotingServer != null) { 
  15.         this.fastRemotingServer.start();
  16.     }
  18.     if (this.fileWatchService != null) { 
  19.         this.fileWatchService.start();
  20.     }
  22.     // 这个组件就是让Broker通过Netty客户端去发送请求出去给别人
  23.     if (this.brokerOuterAPI != null) { 
  24.         this.brokerOuterAPI.start();
  25.     }
  27.     if (this.pullRequestHoldService != null) { 
  28.         this.pullRequestHoldService.start();
  29.     }
  31.     if (this.clientHousekeepingService != null) { 
  32.         this.clientHousekeepingService.start();
  33.     }
  35.     if (this.filterServerManager != null) { 
  36.         this.filterServerManager.start();
  37.     }
  39.     if (!messageStoreConfig.isEnableDLegerCommitLog()) { 
  40.         startProcessorByHa(messageStoreConfig.getBrokerRole());
  41.         handleSlaveSynchronize(messageStoreConfig.getBrokerRole());
  42.         this.registerBrokerAll(true, false, true);
  43.     }
  45.     // 开一个任务去向namesrv注册，第一次是注册，后续就相当于发送心跳了
  46.     this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
  48.         // 可以看到发送心跳间隔只能在10s-60s之间
  49.         @Override
  50.         public void run() { 
  51.             try { 
  52.                 BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister());
  53.             } catch (Throwable e) { 
  54.                 log.error("registerBrokerAll Exception", e);
  55.             }
  56.         }
  57.     }, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), TimeUnit.MILLISECONDS);
  59.     if (this.brokerStatsManager != null) { 
  60.         this.brokerStatsManager.start();
  61.     }
  63.     if (this.brokerFastFailure != null) { 
  64.         this.brokerFastFailure.start();
  65.     }
  66. }

可以看到是启动一些组件，但这些组件里又会启动一些后台任务，这些等后面解析具体功能时再说明。最后还会显式开启一个定时任务，用于broker向namesrv发送心跳。

好了，最后再来总结下broker启动流程：先解析并创建4个核心配置，根据这4个配置创建BrokerController组件，然后对该组件进行初始化，在初始化的过程中会创建一些必要的功能组件，netty客户端，netty服务端，请求处理器和线程池以及开启一些定时任务，最后再启动各个功能组件并创建一个发送心跳的定时任务。
在这里插入图片描述

路由注册和心跳流程

这节来研究broker向namesrv注册以及发送心跳的流程（同一个定时器）

Broker启动时向集群中所有的NameServer发送心跳，每隔30s发送一次，NameServer收到Broker心跳包时会更新brokerLiveTable的lastUpdateTimestamp，然后NameServer每隔10s扫描brokerLiveTable，如果连续120s没有收到心跳包，NameServer将移除该Broker的路由信息同时关闭Socket连接。

broker发送心跳包

入口代码在启动BrokerController里

  1. // 开一个任务去向namesrv注册，第一次是注册，后续就相当于发送心跳了
  2. this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
  4.     // 可以看到发送心跳间隔只能在10s-60s之间
  5.     @Override
  6.     public void run() { 
  7.         try { 
  8.             BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister());
  9.         } catch (Throwable e) { 
  10.             log.error("registerBrokerAll Exception", e);
  11.         }
  12.     }
  13. }, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), TimeUnit.MILLISECONDS);

定时器中调用BrokerController的registerBrokerAll方法：

  1. /** * 去向namesrv注册 */
  2. public synchronized void registerBrokerAll(final boolean checkOrderConfig, boolean oneway, boolean forceRegister) { 
  3.     TopicConfigSerializeWrapper topicConfigWrapper = this.getTopicConfigManager().buildTopicConfigSerializeWrapper();
  4.     // 不懂，跳过
  5.     if (!PermName.isWriteable(this.getBrokerConfig().getBrokerPermission())
  6.         || !PermName.isReadable(this.getBrokerConfig().getBrokerPermission())) { 
  7.         ConcurrentHashMap<String, TopicConfig> topicConfigTable = new ConcurrentHashMap<String, TopicConfig>();
  8.         for (TopicConfig topicConfig : topicConfigWrapper.getTopicConfigTable().values()) { 
  9.             TopicConfig tmp =
  10.                 new TopicConfig(topicConfig.getTopicName(), topicConfig.getReadQueueNums(), topicConfig.getWriteQueueNums(),
  11.                     this.brokerConfig.getBrokerPermission());
  12.             topicConfigTable.put(topicConfig.getTopicName(), tmp);
  13.         }
  14.         topicConfigWrapper.setTopicConfigTable(topicConfigTable);
  15.     }
  17.     // 判断是否需要注册
  18.     if (forceRegister || needRegister(this.brokerConfig.getBrokerClusterName(),
  19.         this.getBrokerAddr(),
  20.         this.brokerConfig.getBrokerName(),
  21.         this.brokerConfig.getBrokerId(),
  22.         this.brokerConfig.getRegisterBrokerTimeoutMills())) { 
  23.         // 注册
  24.         doRegisterBrokerAll(checkOrderConfig, oneway, topicConfigWrapper);
  25.     }
  26. }

其中，forceRegister默认为true，所以到下面注册的逻辑里

  1. private void doRegisterBrokerAll(boolean checkOrderConfig, boolean oneway,
  2.     TopicConfigSerializeWrapper topicConfigWrapper) { 
  3.     // 通过brokerOuterAPI组件发送网络请求到每个NameServer去注册
  4.     List<RegisterBrokerResult> registerBrokerResultList = this.brokerOuterAPI.registerBrokerAll(
  5.         this.brokerConfig.getBrokerClusterName(),
  6.         this.getBrokerAddr(),
  7.         this.brokerConfig.getBrokerName(),
  8.         this.brokerConfig.getBrokerId(),
  9.         this.getHAServerAddr(),
  10.         topicConfigWrapper,
  11.         this.filterServerManager.buildNewFilterServerList(),
  12.         oneway,
  13.         this.brokerConfig.getRegisterBrokerTimeoutMills(),
  14.         this.brokerConfig.isCompressedRegister());
  16.     if (registerBrokerResultList.size() > 0) { 
  17.         // 获取第一个注册结果进行处理
  18.         RegisterBrokerResult registerBrokerResult = registerBrokerResultList.get(0);
  19.         if (registerBrokerResult != null) { 
  20.             if (this.updateMasterHAServerAddrPeriodically && registerBrokerResult.getHaServerAddr() != null) { 
  21.                 this.messageStore.updateHaMasterAddress(registerBrokerResult.getHaServerAddr());
  22.             }
  24.             this.slaveSynchronize.setMasterAddr(registerBrokerResult.getMasterAddr());
  26.             if (checkOrderConfig) { 
  27.                 this.getTopicConfigManager().updateOrderTopicConfig(registerBrokerResult.getKvTable());
  28.             }
  29.         }
  30.     }
  31. }

通过brokerOuterAPI组件去发送心跳

  1. /** * 构建请求去注册Broker */
  2. public List<RegisterBrokerResult> registerBrokerAll(
  3.     final String clusterName,
  4.     final String brokerAddr,
  5.     final String brokerName,
  6.     final long brokerId,
  7.     final String haServerAddr,
  8.     final TopicConfigSerializeWrapper topicConfigWrapper,
  9.     final List<String> filterServerList,
  10.     final boolean oneway,
  11.     final int timeoutMills,
  12.     final boolean compressed) { 
  14.     // 初始化注册结果列表
  15.     final List<RegisterBrokerResult> registerBrokerResultList = Lists.newArrayList();
  16.     // nameServer地址列表
  17.     List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
  18.     if (nameServerAddressList != null && nameServerAddressList.size() > 0) { 
  20.         // 构建请求数据
  21.         final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();
  22.         requestHeader.setBrokerAddr(brokerAddr);
  23.         requestHeader.setBrokerId(brokerId);
  24.         requestHeader.setBrokerName(brokerName);
  25.         requestHeader.setClusterName(clusterName);
  26.         requestHeader.setHaServerAddr(haServerAddr);
  27.         requestHeader.setCompressed(compressed);
  29.         // 主题配置信息
  30.         RegisterBrokerBody requestBody = new RegisterBrokerBody();
  31.         requestBody.setTopicConfigSerializeWrapper(topicConfigWrapper);
  32.         requestBody.setFilterServerList(filterServerList);
  33.         final byte[] body = requestBody.encode(compressed);
  34.         final int bodyCrc32 = UtilAll.crc32(body);
  35.         requestHeader.setBodyCrc32(bodyCrc32);
  36.         // 所有nameserver都注册完后再继续执行
  37.         final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());
  38.         // 循环向每个nameserver注册
  39.         for (final String namesrvAddr : nameServerAddressList) { 
  40.             // 提交任务给线程池同时向每个nameserver注册
  41.             brokerOuterExecutor.execute(new Runnable() { 
  42.                 @Override
  43.                 public void run() { 
  44.                     try { 
  45.                         RegisterBrokerResult result = registerBroker(namesrvAddr,oneway, timeoutMills,requestHeader,body);
  46.                         if (result != null) { 
  47.                             // 将注册结果放到列表中
  48.                             registerBrokerResultList.add(result);
  49.                         }
  51.                         log.info("register broker[{}]to name server {} OK", brokerId, namesrvAddr);
  52.                     } catch (Exception e) { 
  53.                         log.warn("registerBroker Exception, {}", namesrvAddr, e);
  54.                     } finally { 
  55.                         countDownLatch.countDown();
  56.                     }
  57.                 }
  58.             });
  59.         }
  61.         try { 
  62.             // 都向每个nameserver注册完才继续往下，才算注册完成
  63.             countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS);
  64.         } catch (InterruptedException e) { 
  65.         }
  66.     }
  68.     return registerBrokerResultList;
  69. }

通过将数据封装在RegisterBrokerRequestHeader中发送请求。
通过CountDownLatch来保证等待向所有nameserver发送完心跳后返回。

  1. /** * 根据nameserver地址、构建的请求头和请求体去注册broker */
  2. private RegisterBrokerResult registerBroker(
  3.     final String namesrvAddr,
  4.     final boolean oneway,
  5.     final int timeoutMills,
  6.     final RegisterBrokerRequestHeader requestHeader,
  7.     final byte[] body
  8. ) throws RemotingCommandException, MQBrokerException, RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException,
  9.     InterruptedException { 
  10.     // 构建一个注册的请求数据，其实请求数据和响应数据都是RemotingCommand
  11.     RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.REGISTER_BROKER, requestHeader);
  12.     request.setBody(body);
  14.     // 单向请求，表示不用等待注册结果，返回null
  15.     if (oneway) { 
  16.         try { 
  17.             this.remotingClient.invokeOneway(namesrvAddr, request, timeoutMills);
  18.         } catch (RemotingTooMuchRequestException e) { 
  19.             // Ignore
  20.         }
  21.         return null;
  22.     }
  24.     // 同步调用
  25.     RemotingCommand response = this.remotingClient.invokeSync(namesrvAddr, request, timeoutMills);
  26.     assert response != null;
  27.     // 根据响应码判断注册是否成功
  28.     switch (response.getCode()) { 
  29.         case ResponseCode.SUCCESS: { 
  30.             RegisterBrokerResponseHeader responseHeader =
  31.                 (RegisterBrokerResponseHeader) response.decodeCommandCustomHeader(RegisterBrokerResponseHeader.class);
  32.             RegisterBrokerResult result = new RegisterBrokerResult();
  33.             result.setMasterAddr(responseHeader.getMasterAddr());
  34.             result.setHaServerAddr(responseHeader.getHaServerAddr());
  35.             if (response.getBody() != null) { 
  36.                 result.setKvTable(KVTable.decode(response.getBody(), KVTable.class));
  37.             }
  38.             return result;
  39.         }
  40.         default:
  41.             break;
  42.     }
  44.     // 失败则抛异常
  45.     throw new MQBrokerException(response.getCode(), response.getRemark());
  46. }

发送心跳是同步请求，通过NettyRemotingClient发送请求，NettyRemotingClient可以看作是netty客户端，专门用来发送请求，与具体业务无关。
NettyRemotingClient：

  1. /** * netty客户端同步调用 */
  2. @Override
  3. public RemotingCommand invokeSync(String addr, final RemotingCommand request, long timeoutMillis)
  4.     throws InterruptedException, RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException { 
  5.     long beginStartTime = System.currentTimeMillis();
  6.     // 从缓存获取或创建channel
  7.     final Channel channel = this.getAndCreateChannel(addr);
  8.     if (channel != null && channel.isActive()) { 
  9.         try { 
  10.             doBeforeRpcHooks(addr, request);
  11.             long costTime = System.currentTimeMillis() - beginStartTime;
  12.             // 已经超时了
  13.             if (timeoutMillis < costTime) { 
  14.                 throw new RemotingTimeoutException("invokeSync call timeout");
  15.             }
  16.             // 同步调用
  17.             RemotingCommand response = this.invokeSyncImpl(channel, request, timeoutMillis - costTime);
  18.             doAfterRpcHooks(RemotingHelper.parseChannelRemoteAddr(channel), request, response);
  19.             return response;
  20.         } catch (RemotingSendRequestException e) { 
  21.             log.warn("invokeSync: send request exception, so close the channel[{}]", addr);
  22.             this.closeChannel(addr, channel);
  23.             throw e;
  24.         } catch (RemotingTimeoutException e) { 
  25.             if (nettyClientConfig.isClientCloseSocketIfTimeout()) { 
  26.                 this.closeChannel(addr, channel);
  27.                 log.warn("invokeSync: close socket because of timeout, {}ms, {}", timeoutMillis, addr);
  28.             }
  29.             log.warn("invokeSync: wait response timeout exception, the channel[{}]", addr);
  30.             throw e;
  31.         }
  32.     } else { 
  33.         this.closeChannel(addr, channel);
  34.         throw new RemotingConnectException(addr);
  35.     }
  36. }
  38. public RemotingCommand invokeSyncImpl(final Channel channel, final RemotingCommand request,
  39.     final long timeoutMillis)
  40.     throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException { 
  41.     // 可以看作request id
  42.     final int opaque = request.getOpaque();
  44.     try { 
  45.         // 创建一个该请求的ResponseFuture暂存在响应表中
  46.         final ResponseFuture responseFuture = new ResponseFuture(channel, opaque, timeoutMillis, null, null);
  47.         this.responseTable.put(opaque, responseFuture);
  48.         final SocketAddress addr = channel.remoteAddress();
  49.         // 发送请求
  50.         channel.writeAndFlush(request).addListener(new ChannelFutureListener() { 
  51.             // 请求发送成功的回调函数
  52.             @Override
  53.             public void operationComplete(ChannelFuture f) throws Exception { 
  54.                 if (f.isSuccess()) { 
  55.                     responseFuture.setSendRequestOK(true);
  56.                     return;
  57.                 } else { 
  58.                     responseFuture.setSendRequestOK(false);
  59.                 }
  61.                 // 请求发送失败，移除缓存，响应设置为null
  62.                 responseTable.remove(opaque);
  63.                 responseFuture.setCause(f.cause());
  64.                 responseFuture.putResponse(null);
  65.                 log.warn("send a request command to channel <" + addr + "> failed.");
  66.             }
  67.         });
  68.         // 同步等待响应结果，由响应线程唤醒
  69.         RemotingCommand responseCommand = responseFuture.waitResponse(timeoutMillis);
  70.         if (null == responseCommand) { 
  71.             // 响应为空但请求发送成功表示响应超时
  72.             if (responseFuture.isSendRequestOK()) { 
  73.                 throw new RemotingTimeoutException(RemotingHelper.parseSocketAddressAddr(addr), timeoutMillis,
  74.                     responseFuture.getCause());
  75.             } else { 
  76.                 // 请求发送失败
  77.                 throw new RemotingSendRequestException(RemotingHelper.parseSocketAddressAddr(addr), responseFuture.getCause());
  78.             }
  79.         }
  81.         return responseCommand;
  82.     } finally { 
  83.         this.responseTable.remove(opaque);
  84.     }
  85. }

channel.writeAndFlush(request)本质上也是异步请求，发送请求成功后就返回了，所以如果需要实现同步请求，就需要阻塞等待请求结果。
实现过程是创建一个该请求的ResponseFuture暂存在响应表中，用于客户端发送数据后，服务端返回响应结果的时候，和客户端的请求正确匹配起来。具体过程是客户端发送请求时将<请求id, responseFuture>缓存起来，这时用CountDownLatch阻塞请求线程，等待响应结果。当netty客户端接收到服务端的响应后，响应线程根据响应数据中的请求id从缓存中取出responseFuture， 将响应结果设置到responseFuture中，然后通过CountDownLatch操作通知请求线程，将请求线程唤醒，请求线程通过responseFuture拿到响应结果进行处理。所以说到底还是异步请求，只是通过CountDownLatch将线程阻塞住等待结果。

这里分析了netty请求过程，后面涉及netty请求就不再分析了。

下面看看nameserver接收到心跳请求后如何处理。

nameserver处理心跳包

netty收到请求的处理入口在NettyRemotingServer中，刚好与发送请求的NettyRemotingClient对应。
NettyRemotingServer

  1. public class NettyRemotingServer extends NettyRemotingAbstract implements RemotingServer {     
  3.     @ChannelHandler.Sharable
  4.     class NettyServerHandler extends SimpleChannelInboundHandler<RemotingCommand> { 
  6.         @Override
  7.         protected void channelRead0(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception { 
  8.             // netty服务端处理请求入口
  9.             processMessageReceived(ctx, msg);
  10.         }
  11.     }
  12. }

调用父类方法
NettyRemotingAbstract

  1. public void processMessageReceived(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception { 
  2.     final RemotingCommand cmd = msg;
  3.     if (cmd != null) { 
  4.         switch (cmd.getType()) { 
  5.             case REQUEST_COMMAND:
  6.                 // 处理请求
  7.                 processRequestCommand(ctx, cmd);
  8.                 break;
  9.             case RESPONSE_COMMAND:
  10.                 // 处理响应
  11.                 processResponseCommand(ctx, cmd);
  12.                 break;
  13.             default:
  14.                 break;
  15.         }
  16.     }
  17. }

可以看到这有两种类型，处理请求可以理解，但为什么会有处理响应的？上面说到了，由于netty发送同步请求本质上还是异步请求，所以netty处理完请求后需要将响应结果重新发送给客户端。看源码发现NettyRemotingClient也继承了NettyRemotingAbstract

  1. public class NettyRemotingClient extends NettyRemotingAbstract implements RemotingClient { 
  3.     class NettyClientHandler extends SimpleChannelInboundHandler<RemotingCommand> { 
  5.         @Override
  6.         protected void channelRead0(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception { 
  7.             processMessageReceived(ctx, msg);
  8.         }
  9.     }
  11. }

所以NettyRemotingClient这边的该方法是用来处理响应的。
好了，接着回到server端处理接收到的请求

  1. /** * 处理接收到的请求 * * @param ctx channel handler context. * @param cmd request command. */
  2. public void processRequestCommand(final ChannelHandlerContext ctx, final RemotingCommand cmd) { 
  3.     // 获得属于该请求类型的Processor和Executor，没有则用默认的
  4.     // 这里的注册请求没有专门的处理器和线程池，使用默认的
  5.     final Pair<NettyRequestProcessor, ExecutorService> matched = this.processorTable.get(cmd.getCode());
  6.     final Pair<NettyRequestProcessor, ExecutorService> pair = null == matched ? this.defaultRequestProcessor : matched;
  7.     // 相当于请求id
  8.     final int opaque = cmd.getOpaque();
  10.     if (pair != null) { 
  11.         Runnable run = new Runnable() { 
  12.             @Override
  13.             public void run() { 
  14.                 try { 
  15.                     doBeforeRpcHooks(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), cmd);
  16.                     // 对应处理器处理请求
  17.                     final RemotingCommand response = pair.getObject1().processRequest(ctx, cmd);
  18.                     doAfterRpcHooks(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), cmd, response);
  20.                     // 如果不是单向请求，需要将结果返回去，异步也会，只是不对响应进行处理
  21.                     if (!cmd.isOnewayRPC()) { 
  22.                         if (response != null) { 
  23.                             // 将请求id设到响应中，用于客户端根据请求id找到对应的请求
  24.                             response.setOpaque(opaque);
  25.                             response.markResponseType();
  26.                             try { 
  27.                                 ctx.writeAndFlush(response);
  28.                             } catch (Throwable e) { 
  29.                                 log.error("process request over, but response failed", e);
  30.                                 log.error(cmd.toString());
  31.                                 log.error(response.toString());
  32.                             }
  33.                         } else { 
  35.                         }
  36.                     }
  37.                 } catch (Throwable e) { 
  38.                     log.error("process request exception", e);
  39.                     log.error(cmd.toString());
  41.                     if (!cmd.isOnewayRPC()) { 
  42.                         final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_ERROR,
  43.                             RemotingHelper.exceptionSimpleDesc(e));
  44.                         response.setOpaque(opaque);
  45.                         ctx.writeAndFlush(response);
  46.                     }
  47.                 }
  48.             }
  49.         };
  51.         if (pair.getObject1().rejectRequest()) { 
  52.             final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_BUSY,
  53.                 "[REJECTREQUEST]system busy, start flow control for a while");
  54.             response.setOpaque(opaque);
  55.             ctx.writeAndFlush(response);
  56.             return;
  57.         }
  59.         try { 
  60.             // 外面再封装一层任务，任务中直接调用Runnable对象的run方法执行
  61.             final RequestTask requestTask = new RequestTask(run, ctx.channel(), cmd);
  62.             // 将任务提交到对应线程池中
  63.             // 这里注册请求也是使用默认的线程池
  64.             pair.getObject2().submit(requestTask);
  65.         } catch (RejectedExecutionException e) { 
  66.             if ((System.currentTimeMillis() % 10000) == 0) { 
  67.                 log.warn(RemotingHelper.parseChannelRemoteAddr(ctx.channel())
  68.                     + ", too many requests and system thread pool busy, RejectedExecutionException "
  69.                     + pair.getObject2().toString()
  70.                     + " request code: " + cmd.getCode());
  71.             }
  73.             if (!cmd.isOnewayRPC()) { 
  74.                 final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_BUSY,
  75.                     "[OVERLOAD]system busy, start flow control for a while");
  76.                 response.setOpaque(opaque);
  77.                 ctx.writeAndFlush(response);
  78.             }
  79.         }
  80.     } else { 
  81.         String error = " request type " + cmd.getCode() + " not supported";
  82.         final RemotingCommand response =
  83.             RemotingCommand.createResponseCommand(RemotingSysResponseCode.REQUEST_CODE_NOT_SUPPORTED, error);
  84.         response.setOpaque(opaque);
  85.         ctx.writeAndFlush(response);
  86.         log.error(RemotingHelper.parseChannelRemoteAddr(ctx.channel()) + error);
  87.     }
  88. }

接下来就需要使用对应的处理器创建一个线程提交到线程池中，由于注册请求没有专门的处理器和线程池，所以都是用默认的。
DefaultRequestProcessor

  1. /** * 处理netty接收到的请求 */
  2. @Override
  3. public RemotingCommand processRequest(ChannelHandlerContext ctx,
  4.     RemotingCommand request) throws RemotingCommandException { 
  6.     if (ctx != null) { 
  7.         log.debug("receive request, {} {} {}",
  8.             request.getCode(),
  9.             RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
  10.             request);
  11.     }
  13.     // 根据不同请求类型进行不同处理
  14.     switch (request.getCode()) { 
  15.         case RequestCode.PUT_KV_CONFIG:
  16.             return this.putKVConfig(ctx, request);
  17.         case RequestCode.GET_KV_CONFIG:
  18.             return this.getKVConfig(ctx, request);
  19.         case RequestCode.DELETE_KV_CONFIG:
  20.             return this.deleteKVConfig(ctx, request);
  21.         case RequestCode.QUERY_DATA_VERSION:
  22.             return queryBrokerTopicConfig(ctx, request);
  23.         // 注册broker和broke发送心跳的请求
  24.         case RequestCode.REGISTER_BROKER:
  25.             Version brokerVersion = MQVersion.value2Version(request.getVersion());
  26.             if (brokerVersion.ordinal() >= MQVersion.Version.V3_0_11.ordinal()) { 
  27.                 return this.registerBrokerWithFilterServer(ctx, request);
  28.             } else { 
  29.                 // 注册broker
  30.                 return this.registerBroker(ctx, request);
  31.             }
  32.         // broker正常下线，移除broker的注册信息
  33.         case RequestCode.UNREGISTER_BROKER:
  34.             return this.unregisterBroker(ctx, request);
  35.         // 根据topic获取路由信息
  36.         case RequestCode.GET_ROUTEINTO_BY_TOPIC:
  37.             return this.getRouteInfoByTopic(ctx, request);
  38.         case RequestCode.GET_BROKER_CLUSTER_INFO:
  39.             return this.getBrokerClusterInfo(ctx, request);
  40.         case RequestCode.WIPE_WRITE_PERM_OF_BROKER:
  41.             return this.wipeWritePermOfBroker(ctx, request);
  42.         case RequestCode.GET_ALL_TOPIC_LIST_FROM_NAMESERVER:
  43.             return getAllTopicListFromNameserver(ctx, request);
  44.         case RequestCode.DELETE_TOPIC_IN_NAMESRV:
  45.             return deleteTopicInNamesrv(ctx, request);
  46.         case RequestCode.GET_KVLIST_BY_NAMESPACE:
  47.             return this.getKVListByNamespace(ctx, request);
  48.         case RequestCode.GET_TOPICS_BY_CLUSTER:
  49.             return this.getTopicsByCluster(ctx, request);
  50.         case RequestCode.GET_SYSTEM_TOPIC_LIST_FROM_NS:
  51.             return this.getSystemTopicListFromNs(ctx, request);
  52.         case RequestCode.GET_UNIT_TOPIC_LIST:
  53.             return this.getUnitTopicList(ctx, request);
  54.         case RequestCode.GET_HAS_UNIT_SUB_TOPIC_LIST:
  55.             return this.getHasUnitSubTopicList(ctx, request);
  56.         case RequestCode.GET_HAS_UNIT_SUB_UNUNIT_TOPIC_LIST:
  57.             return this.getHasUnitSubUnUnitTopicList(ctx, request);
  58.         case RequestCode.UPDATE_NAMESRV_CONFIG:
  59.             return this.updateConfig(ctx, request);
  60.         case RequestCode.GET_NAMESRV_CONFIG:
  61.             return this.getConfig(ctx, request);
  62.         default:
  63.             break;
  64.     }
  65.     return null;
  66. }

可以看到该默认的请求处理器也可以处理多种类型的请求，我们就看处理注册broker和broke发送心跳的请求

  1. /** * 处理注册的请求 */
  2. public RemotingCommand registerBroker(ChannelHandlerContext ctx,
  3.     RemotingCommand request) throws RemotingCommandException { 
  4.     // 构建响应的RemotingCommand
  5.     final RemotingCommand response = RemotingCommand.createResponseCommand(RegisterBrokerResponseHeader.class);
  6.     final RegisterBrokerResponseHeader responseHeader = (RegisterBrokerResponseHeader) response.readCustomHeader();
  7.     final RegisterBrokerRequestHeader requestHeader =
  8.         (RegisterBrokerRequestHeader) request.decodeCommandCustomHeader(RegisterBrokerRequestHeader.class);
  10.     if (!checksum(ctx, request, requestHeader)) { 
  11.         response.setCode(ResponseCode.SYSTEM_ERROR);
  12.         response.setRemark("crc32 not match");
  13.         return response;
  14.     }
  16.     // topic路由配置信息
  17.     TopicConfigSerializeWrapper topicConfigWrapper;
  18.     if (request.getBody() != null) { 
  19.         topicConfigWrapper = TopicConfigSerializeWrapper.decode(request.getBody(), TopicConfigSerializeWrapper.class);
  20.     } else { 
  21.         topicConfigWrapper = new TopicConfigSerializeWrapper();
  22.         topicConfigWrapper.getDataVersion().setCounter(new AtomicLong(0));
  23.         topicConfigWrapper.getDataVersion().setTimestamp(0);
  24.     }
  25.     // 向路由信息管理器实际注册broker信息
  26.     RegisterBrokerResult result = this.namesrvController.getRouteInfoManager().registerBroker(
  27.         requestHeader.getClusterName(),
  28.         requestHeader.getBrokerAddr(),
  29.         requestHeader.getBrokerName(),
  30.         requestHeader.getBrokerId(),
  31.         requestHeader.getHaServerAddr(),
  32.         topicConfigWrapper,
  33.         null,
  34.         ctx.channel()
  35.     );
  37.     responseHeader.setHaServerAddr(result.getHaServerAddr());
  38.     responseHeader.setMasterAddr(result.getMasterAddr());
  40.     byte[] jsonValue = this.namesrvController.getKvConfigManager().getKVListByNamespace(NamesrvUtil.NAMESPACE_ORDER_TOPIC_CONFIG);
  41.     response.setBody(jsonValue);
  42.     response.setCode(ResponseCode.SUCCESS);
  43.     response.setRemark(null);
  44.     return response;
  45. }

从NamesrvController中获取RouteInfoManager路由信息管理器来真正注册broker
在真正注册之前，先来看看RouteInfoManager中即将用到的属性吧

  1. private final static long BROKER_CHANNEL_EXPIRED_TIME = 1000 * 60 * 2;
  3. private final ReadWriteLock lock = new ReentrantReadWriteLock();
  4. /** * 每个topic对应的队列 */
  5. private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
  6. /** * 维护每个主从架构中的broker信息 */
  7. private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
  8. /** * 维护每个broker集群中的每个broker名称，一般只会有一个集群 */
  9. private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
  10. /** * 维护心跳信息 */
  11. private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
  13. private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;

现在看真正执行注册的逻辑

  1. /** * 向该路由信息管理器注册broker信息 */
  2. public RegisterBrokerResult registerBroker(
  3.     final String clusterName,
  4.     final String brokerAddr,
  5.     final String brokerName,
  6.     final long brokerId,
  7.     final String haServerAddr,
  8.     final TopicConfigSerializeWrapper topicConfigWrapper,
  9.     final List<String> filterServerList,
  10.     final Channel channel) { 
  11.     RegisterBrokerResult result = new RegisterBrokerResult();
  12.     try { 
  13.         try { 
  14.             // 获得写锁，防止并发修改路由表
  15.             this.lock.writeLock().lockInterruptibly();
  17.             // 取出该集群名下的broker名称集合，判断该broker所属集群是否存在，不存在则先创建一个
  18.             Set<String> brokerNames = this.clusterAddrTable.get(clusterName);
  19.             if (null == brokerNames) { 
  20.                 brokerNames = new HashSet<String>();
  21.                 this.clusterAddrTable.put(clusterName, brokerNames);
  22.             }
  23.             // 将broker名称添加到clusterName下
  24.             brokerNames.add(brokerName);
  26.             // broker是否是第一次注册
  27.             boolean registerFirst = false;
  29.             BrokerData brokerData = this.brokerAddrTable.get(brokerName);
  30.             if (null == brokerData) { 
  31.                 // 没有broker主从架构信息，说明broker是第一次注册
  32.                 registerFirst = true;
  33.                 // 构建broker数据
  34.                 brokerData = new BrokerData(clusterName, brokerName, new HashMap<Long, String>());
  35.                 this.brokerAddrTable.put(brokerName, brokerData);
  36.             }
  37.             Map<Long, String> brokerAddrsMap = brokerData.getBrokerAddrs();
  38.             //Switch slave to master: first remove <1, IP:PORT> in namesrv, then add <0, IP:PORT>
  39.             //The same IP:PORT must only have one record in brokerAddrTable
  40.             Iterator<Entry<Long, String>> it = brokerAddrsMap.entrySet().iterator();
  41.             while (it.hasNext()) { 
  42.                 Entry<Long, String> item = it.next();
  43.                 if (null != brokerAddr && brokerAddr.equals(item.getValue()) && brokerId != item.getKey()) { 
  44.                     it.remove();
  45.                 }
  46.             }
  48.             // 设置对应brokerId的地址，若存在，则更新，并且registerFirst设为true
  49.             String oldAddr = brokerData.getBrokerAddrs().put(brokerId, brokerAddr);
  50.             registerFirst = registerFirst || (null == oldAddr);
  52.             // broker是master
  53.             if (null != topicConfigWrapper
  54.                 && MixAll.MASTER_ID == brokerId) { 
  55.                 // broker的topic路由信息改变了或首次注册，则需要创建或更新topic路由元数据，填充topicQueueTable
  56.                 // 其实就是为默认主题自动注册路由信息，其中包含MixAll.DEFAULT_TOPIC的路由信息。当消息生产者发送
  57.                 // 主题时，如果该主题为创建并且BrokerConfig的autoCreateTopicEnable为true时，将返回
  58.                 // MixAll.DEFAULT_TOPIC的路由信息
  59.                 if (this.isBrokerTopicConfigChanged(brokerAddr, topicConfigWrapper.getDataVersion())
  60.                     || registerFirst) { 
  61.                     ConcurrentMap<String, TopicConfig> tcTable =
  62.                         topicConfigWrapper.getTopicConfigTable();
  63.                     if (tcTable != null) { 
  64.                         for (Map.Entry<String, TopicConfig> entry : tcTable.entrySet()) { 
  65.                             // 要创建或更新topic路由元数据，填充topicQueueTable
  66.                             this.createAndUpdateQueueData(brokerName, entry.getValue());
  67.                         }
  68.                     }
  69.                 }
  70.             }
  72.             // 心跳逻辑
  73.             // 默认每隔30s收到心跳请求，生成一个新的BrokerLiveInfo，覆盖掉原来的BrokerLiveInfo，
  74.             // BrokerLiveInfo里有一个lastUpdateTimestamp属性，表示最近收到心跳的时间
  75.             BrokerLiveInfo prevBrokerLiveInfo = this.brokerLiveTable.put(brokerAddr,
  76.                 new BrokerLiveInfo(
  77.                     System.currentTimeMillis(),
  78.                     topicConfigWrapper.getDataVersion(),
  79.                     channel,
  80.                     haServerAddr));
  81.             if (null == prevBrokerLiveInfo) { 
  82.                 log.info("new broker registered, {} HAServer: {}", brokerAddr, haServerAddr);
  83.             }
  85.             if (filterServerList != null) { 
  86.                 if (filterServerList.isEmpty()) { 
  87.                     this.filterServerTable.remove(brokerAddr);
  88.                 } else { 
  89.                     this.filterServerTable.put(brokerAddr, filterServerList);
  90.                 }
  91.             }
  93.             // 如果是从节点，并且master信息已经注册过了，则将master地址设置到RegisterBrokerResult
  94.             if (MixAll.MASTER_ID != brokerId) { 
  95.                 String masterAddr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
  96.                 if (masterAddr != null) { 
  97.                     BrokerLiveInfo brokerLiveInfo = this.brokerLiveTable.get(masterAddr);
  98.                     if (brokerLiveInfo != null) { 
  99.                         result.setHaServerAddr(brokerLiveInfo.getHaServerAddr());
  100.                         result.setMasterAddr(masterAddr);
  101.                     }
  102.                 }
  103.             }
  104.         } finally { 
  105.             this.lock.writeLock().unlock();
  106.         }
  107.     } catch (Exception e) { 
  108.         log.error("registerBroker Exception", e);
  109.     }
  111.     return result;
  112. }

具体的流程都写在注释里了

这里使用了读写锁来控制对路由表的并发读写，多个生产者可以同时来读取路由表，但同时只能一个broker来更改路由表。

broker处理响应

nameserver处理完注册请求后，会将结果响应回去，broker这边由NettyRemotingClient来处理响应，唤醒阻塞的请求线程，这时就会用到上面讲到的client来处理响应。

  1. public void processMessageReceived(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception { 
  2.     final RemotingCommand cmd = msg;
  3.     if (cmd != null) { 
  4.         switch (cmd.getType()) { 
  5.             case REQUEST_COMMAND:
  6.                 // 处理请求
  7.                 processRequestCommand(ctx, cmd);
  8.                 break;
  9.             case RESPONSE_COMMAND:
  10.                 // 处理响应
  11.                 processResponseCommand(ctx, cmd);
  12.                 break;
  13.             default:
  14.                 break;
  15.         }
  16.     }
  17. }

此时参数中的RemotingCommand就不是请求数据了，而是响应数据。

  1. /** * 处理响应 */
  2. public void processResponseCommand(ChannelHandlerContext ctx, RemotingCommand cmd) { 
  3.     final int opaque = cmd.getOpaque();
  4.     // 根据id从正在等待响应的缓存中获取请求
  5.     final ResponseFuture responseFuture = responseTable.get(opaque);
  6.     if (responseFuture != null) { 
  7.         // 设置响应数据
  8.         responseFuture.setResponseCommand(cmd);
  10.         responseTable.remove(opaque);
  12.         // 异步会有InvokeCallback
  13.         if (responseFuture.getInvokeCallback() != null) { 
  14.             executeInvokeCallback(responseFuture);
  15.         } else { 
  16.             // 同步执行这里
  17.             // 会使用countDownLatch.countDown()唤醒
  18.             responseFuture.putResponse(cmd);
  19.             responseFuture.release();
  20.         }
  21.     } else { 
  22.         log.warn("receive response, but not matched any request, " + RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
  23.         log.warn(cmd.toString());
  24.     }
  25. }

重新贴一下发送同步请求的代码：

  1. public RemotingCommand invokeSyncImpl(final Channel channel, final RemotingCommand request,
  2.     final long timeoutMillis)
  3.     throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException { 
  4.     // 可以看作request id
  5.     final int opaque = request.getOpaque();
  7.     try { 
  8.         final ResponseFuture responseFuture = new ResponseFuture(channel, opaque, timeoutMillis, null, null);
  9.         this.responseTable.put(opaque, responseFuture);
  10.         final SocketAddress addr = channel.remoteAddress();
  11.         // 发送请求
  12.         channel.writeAndFlush(request).addListener(new ChannelFutureListener() { 
  13.             // 请求发送成功的回调函数
  14.             @Override
  15.             public void operationComplete(ChannelFuture f) throws Exception { 
  16.                 if (f.isSuccess()) { 
  17.                     responseFuture.setSendRequestOK(true);
  18.                     return;
  19.                 } else { 
  20.                     responseFuture.setSendRequestOK(false);
  21.                 }
  23.                 // 请求发送失败，移除缓存，响应设置为null
  24.                 responseTable.remove(opaque);
  25.                 responseFuture.setCause(f.cause());
  26.                 responseFuture.putResponse(null);
  27.                 log.warn("send a request command to channel <" + addr + "> failed.");
  28.             }
  29.         });
  30.         // 使用countDownLatch.await同步等待响应结果，由响应线程唤醒
  31.         RemotingCommand responseCommand = responseFuture.waitResponse(timeoutMillis);
  32.         if (null == responseCommand) { 
  33.             // 响应为空但请求发送成功表示响应超时
  34.             if (responseFuture.isSendRequestOK()) { 
  35.                 throw new RemotingTimeoutException(RemotingHelper.parseSocketAddressAddr(addr), timeoutMillis,
  36.                     responseFuture.getCause());
  37.             } else { 
  38.                 // 请求发送失败
  39.                 throw new RemotingSendRequestException(RemotingHelper.parseSocketAddressAddr(addr), responseFuture.getCause());
  40.             }
  41.         }
  43.         return responseCommand;
  44.     } finally { 
  45.         this.responseTable.remove(opaque);
  46.     }
  47. }

此时请求线程就会在31行代码处唤醒，去处理响应结果。
在这里插入图片描述

参考资料

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

发表评论

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

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

相关阅读