MetaQ技术内幕——源码分析(五)
public static void main(final String[] args) {final String configFilePath = getConfigFilePath(args);final MetaConfig metaConfig = getMetaConfig(configFilePath);final MetaMorphosisBroker server = new MetaMorphosisBroker(metaConfig);server.start();}
从MetamorphosisStartup可以看出其逻辑是先加载了配置文件，然后构造了MetaMorphosisBroker实例，并调用该实例的start方法，MetaMorphosisBroker才是Broker真正的启动类。
?
?
看看真正的启动类MetaMorphosisBroker, MetaMorphosisBroker实现接口MetaMorphosisBrokerMBean，可以通过 jmx 协议关闭MetaMorphosisBroker。看看在构造MetaMorphosisBroker实例的时候干了些什么事情。
?
public MetaMorphosisBroker(final MetaConfig metaConfig) { //配置信息this.metaConfig = metaConfig; //Broker对外提供的nio Serverthis.remotingServer = newRemotingServer(metaConfig); //线程池管理器，主要是提供给nio Server在并发环境下可以使用多线程处理，提高性能 this.executorsManager = new ExecutorsManager(metaConfig); //全局唯一的id生成器 this.idWorker = new IdWorker(metaConfig.getBrokerId()); //存储模块管理器 this.storeManager = new MessageStoreManager(metaConfig, this.newDeletePolicy(metaConfig)); //统计模块管理器this.statsManager = new StatsManager(this.metaConfig, this.storeManager, this.remotingServer); //zookeeper客户端，前面介绍过metaq使用zookeeper作为中间协调者，Broker会将自己注册到zookeeper上，也会从zookeeper查询相关数据this.brokerZooKeeper = new BrokerZooKeeper(metaConfig); //网络输入输出流处理器 final BrokerCommandProcessor next = new BrokerCommandProcessor(this.storeManager, this.executorsManager, this.statsManager, this.remotingServer, metaConfig, this.idWorker, this.brokerZooKeeper); //事务存储引擎JournalTransactionStore transactionStore = null;try {transactionStore = new JournalTransactionStore(metaConfig.getDataLogPath(), this.storeManager, metaConfig);} catch (final Exception e) {throw new MetamorphosisServerStartupException("Initializing transaction store failed.", e);} //带事务处理的网络输入输出流处理器，设计采用了责任链的设计模式，使用事务存储引擎存储中间结果this.brokerProcessor = new TransactionalCommandProcessor(metaConfig, this.storeManager, this.idWorker, next, transactionStore, this.statsManager); //钩子，JVM退出钩子，钩子实现在JVM退出的时候尽力正确关闭MetaMorphosisBroker this.shutdownHook = new ShutdownHook(); //注册钩子Runtime.getRuntime().addShutdownHook(this.shutdownHook); //注册MBean，因为MetaMorphosisBroker实现MetaMorphosisBrokerMBean接口，可以将自己作为MBean注册到MBeanServerMetaMBeanServer.registMBean(this, null);}?
前面我们知道在启动的时候会调用MetaMorphosisBroker的start() 方法，来看看start()方法里究竟做了些什么事情
?
public synchronized void start() {//判断是否已经启动，如果已经启动，则不在启动if (!this.shutdown) {return;}this.shutdown = false;//初始化存储模块，加载验证已有数据this.storeManager.init();//初始化线程池this.executorsManager.init();//初始化统计模块this.statsManager.init();//向nio server注册处理器this.registerProcessors();try {//NIO server启动 this.remotingServer.start();} catch (final NotifyRemotingException e) {throw new MetamorphosisServerStartupException("start remoting server failed", e);}try {//在/brokers/ids下创建临时节点，名称为节点Idthis.brokerZooKeeper.registerBrokerInZk();//如果为master节点，则创建/brokers/ids/master_config_checksum节点 this.brokerZooKeeper.registerMasterConfigFileChecksumInZk();//添加主题列表监听器，监听主题列表变化，如果主题列表发生变化，则向zookeeper重新注册主题和分区信息 this.addTopicsChangeListener();//注册主题和分区信息 this.registerTopicsInZk();//设置标志位主题和分区注册成功this.registerZkSuccess = true;} catch (final Exception e) {this.registerZkSuccess = false;throw new MetamorphosisServerStartupException("Register broker to zk failed", e);}log.info("Starting metamorphosis server...");//初始化输入输出流处理器this.brokerProcessor.init();log.info("Start metamorphosis server successfully");}?
?
下面，让我们具体来看看start()方法里调用的MetaMorphosisBroker每一个方法，首先是registerProcessors()方法：
?
private void registerProcessors() {//注册Get命令处理器this.remotingServer.registerProcessor(GetCommand.class, new GetProcessor(this.brokerProcessor, this.executorsManager.getGetExecutor()));//注册Put命令的处理器this.remotingServer.registerProcessor(PutCommand.class, new PutProcessor(this.brokerProcessor, this.executorsManager.getUnOrderedPutExecutor()));//查询最近有效的offset处理器this.remotingServer.registerProcessor(OffsetCommand.class, new OffsetProcessor(this.brokerProcessor, this.executorsManager.getGetExecutor()));//心跳检测处理器this.remotingServer.registerProcessor(HeartBeatRequestCommand.class, new VersionProcessor(this.brokerProcessor));//注册退出命令处理器this.remotingServer.registerProcessor(QuitCommand.class, new QuitProcessor(this.brokerProcessor));//注册统计信息查询处理器this.remotingServer.registerProcessor(StatsCommand.class, new StatsProcessor(this.brokerProcessor));//注册事务命令处理器this.remotingServer.registerProcessor(TransactionCommand.class, new TransactionProcessor(this.brokerProcessor, this.executorsManager.getUnOrderedPutExecutor()));}?
?
依赖于不同的处理器，可以将不同的请求进行处理并返回结果。接下来就是addTopicsChangeListener()方法。
?
//addTopicsChangeListener方法比较简单，主要简单配置的topic列表的变化，前面介绍过MetaConfig提供监听机制监听topic列表的变化，该方法向MetaConfig注册一个匿名监听器监听topic列表变化，一旦发生变化则向zookeeper进行注册private void addTopicsChangeListener() {// 监听topics列表变化并注册到zkthis.metaConfig.addPropertyChangeListener("topics", new PropertyChangeListener() {public void propertyChange(final PropertyChangeEvent evt) {try {MetaMorphosisBroker.this.registerTopicsInZk();} catch (final Exception e) {log.error("Register topic in zk failed", e);}}});}?
?
MetaMorphosisBroker在启动过程中被调用的方法还有registerTopicsInZk()方法，registerTopicsInZk完成向zookeeper注册topic和分区信息功能。在分析方法之前，有必要插入分析一下Broker在zk上注册的结构，代码在common工程的类MetaZookeeper，该结构是Broker和Client共享的。
?
Zk中有4中类型的根目录，分别是:
1) /consumers:存放消费者列表以及消费记录，消费者列表主要是以组的方式存在，结构主要如下：
? ? ? ?/consumers/xxGroup/ids/xxConsumerId：DATA(“：”后的DATA表示节点xxConsumerId对应的数据) 组内消费者Id;DATA为订阅主题列表，以”,”分隔
? ? ? ?/consumers/xxGroup/offsets/xxTopic/分区N：DATA? 组内主题分区N的消费进度；DATA为topic下分区N具体进度值
? ? ? ?/consumers/xxGroup/owners/xxTopic/分区N：DATA?组内主题分区N的的消费者；DATA为消费者ID，表示XXTopic下分区N的数据由指定的消费者进行消费
?
2) /brokers/ids:存放Broker列表，如果Broker与Zookeeper失去连接，则会自动注销在/brokers/ids下的broker记录，例子如下：
? ? /brokers/ids/xxBroker
?
3) /brokers/topics-pub：存放发布的主题列表以及对应的可发送消息的Broker列表，例子如下：
? ? /brokers/topics-pub/xxTopic/xxBroker
? ? /brokers/topics-pub下记录的是可发送消息到xxTopic的Broker列表，意味着有多少个Broker允许存储Client发送到Topic数据
?
4) /brokers/topics-sub：存放订阅的主题列表以及对应可订阅的Broker列表，例子如下：
? ? /brokers/topics-sub/xxTopic/xxBroker
? ? /brokers/topics-sub下记录的可订阅xxTopic的Broker列表，意味着有多少个Broker允许被Client订阅topic的数据
具体代码如下：
?
public MetaZookeeper(final ZkClient zkClient, final String root) {//zk客户端this.zkClient = zkClient;//根路径，默认为空this.metaRoot = this.normalize(root);//前面讲的消费者列表this.consumersPath = this.metaRoot + "/consumers";//前面讲的brokers列表this.brokerIdsPath = this.metaRoot + "/brokers/ids";//前面讲的/brokers/topics-pubthis.brokerTopicsPubPath = this.metaRoot + "/brokers/topics-pub";//前面讲的/brokers/topics-subthis.brokerTopicsSubPath = this.metaRoot + "/brokers/topics-sub";}?至于更复杂的，我们将在后面具体再进行分析，主要先了解该存储结构即可。
?
回归正题, registerTopicsInZk方法完成向zookeeper注册topic和分区信息功能
?
private void registerTopicsInZk() throws Exception {// 先注册配置的topic到zookeeperfor (final String topic : this.metaConfig.getTopics()) {this.brokerZooKeeper.registerTopicInZk(topic, true);}// 注册加载的topic到zookeeper // 从下面代码可以看出，如果当前没有配置的topic，但前面配置过的topic如果有消息存在，依然会向zk注册，在某种程度，我认为这个设计不好，为什么？答：我们前面分析过MessageStoreManager类，里面有getMessageStore()方法和getOrCreateMessageStore()方法，在调用getMessageStore()方法时没有检查参数topic是否在topicsPatSet列表中(topicsPatSet只包含了配置的topic)，而getOrCreateMessageStore()方法却检查了，这就意味着使用getOrCreateMessageStore()方法时，如果要查询获取不在topicsPatSet列表中的MessageStore实例会抛出异常，而调用getMessageStore()不会，让人产生疑惑。个人见解认为一旦配置发生更改，如果要做热加载的话则先卸载再重新加载会更合适，而且在getOrCreateMessageStore()和getMessageStore()方法都使用topicsPatSet进行判断，保持一致性for (final String topic : this.storeManager.getMessageStores().keySet()) {this.brokerZooKeeper.registerTopicInZk(topic, true);}}?
MetaMorphosisBroker还有两个方法，一个是newDeletePolicy()方法，另一个是stop()方法。newDeletePolicy()用于生产全局的存储模块的删除策略，如果没有配置删除策略，则使用该策略。
?
//全局删除策略private DeletePolicy newDeletePolicy(final MetaConfig metaConfig) {final String deletePolicy = metaConfig.getDeletePolicy();if (deletePolicy != null) {return DeletePolicyFactory.getDeletePolicy(deletePolicy);}return null;}?
?
?
而stop()方法则主要在MetaMorphosisBroker关闭的时候销毁资源，尽力保证MetaQ的正确关闭。
?
public synchronized void stop() {//如果关闭了，则不再关闭if (this.shutdown) {return;}log.info("Stopping metamorphosis server...");this.shutdown = true;//关闭与zk连接，注销与当前节点相关的配置this.brokerZooKeeper.close(this.registerZkSuccess);try {// Waiting for zookeeper to notify clients.Thread.sleep(this.brokerZooKeeper.getZkConfig().zkSyncTimeMs);} catch (InterruptedException e) {// ignore}//释放线程池this.executorsManager.dispose();//释放存储模块this.storeManager.dispose();//释放统计模块this.statsManager.dispose();//关闭NIO Servertry {this.remotingServer.stop();} catch (final NotifyRemotingException e) {log.error("Shutdown remoting server failed", e);}//释放输入输出流处理器this.brokerProcessor.dispose();//如果是独立的zk，则关闭zkEmbedZookeeperServer.getInstance().stop();//释放钩子if (!this.runShutdownHook && this.shutdownHook != null) {Runtime.getRuntime().removeShutdownHook(this.shutdownHook);}log.info("Stop metamorphosis server successfully");}?
?
今天，Broker的分析先到这，前面介绍的zk中注册的结构是与Client相关的，这里也向大家介绍了一下，以后分析Client的时候，该结构将不再介绍了。
?
?
?
?