欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

MetaQ技术内幕——源码分析(五)

程序员文章站 2022-07-13 13:48:39
...

Broker接收从Producer(Client端)发送的消息,也能够返回消息到Consumer(Client),对于Broker来说,就是网络输入输出流的处理。

Broker使用淘宝内部的gecko框架作为网络传输框架,gecko是一个NIO框架,能够支持一下特性:

1、 可自定义协议,协议可扩展、紧凑、高效

2、 可自动管理重连,重连由客户端发起

3、 需进行心跳检测,及时发现连接失效

4、 请求应答模型应当支持同步和异步 

5、 连接的分组管理,并且在重连情况下能正确处理连接的分组

6、 请求的发送应当支持四种模型:  (1) 向单个连接发起请求  (2) 向分组内的某个连接发起请求,这个选择策略可定义 (3) 向分组内的所有连接发起请求  (4) 向多个分组发起请求,每个分组的请求遵循(2) 

7、 编程模型上尽量做到简单、易用、透明,向上层代码屏蔽网络处理的复杂细节。

8、 高度可配置,包括网络参数、服务层参数等 

9、 高度可靠,正确处理网络异常,对内存泄露等隐患有预防措施

10、 可扩展

如果时间允许的话,笔者也可以做一下 gecko的源码分析

由于网络模块与其他模块关联性极强,不像存储模块可以独立分析,所以此篇文章开始将从全局开始分析Broker。

先看看Broker的启动类MetamorphosisStartup:

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 Server
this.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注册到MBeanServer
MetaMBeanServer.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下创建临时节点,名称为节点Id
this.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列表变化并注册到zk
this.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-pub
this.brokerTopicsPubPath = this.metaRoot + "/brokers/topics-pub";
//前面讲的/brokers/topics-sub
this.brokerTopicsSubPath = this.metaRoot + "/brokers/topics-sub";
}

 至于更复杂的,我们将在后面具体再进行分析,主要先了解该存储结构即可。

回归正题 , registerTopicsInZk 方法完成向 zookeeper 注册 topic 和分区信息功能

private void registerTopicsInZk() throws Exception {
// 先注册配置的topic到zookeeper
for (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 Server
try {
this.remotingServer.stop();
} catch (final NotifyRemotingException e) {
log.error("Shutdown remoting server failed", e);
}
//释放输入输出流处理器
this.brokerProcessor.dispose();
//如果是独立的zk,则关闭zk
EmbedZookeeperServer.getInstance().stop();
//释放钩子	
if (!this.runShutdownHook && this.shutdownHook != null) {
Runtime.getRuntime().removeShutdownHook(this.shutdownHook);
}
log.info("Stop metamorphosis server successfully");
}

今天, Broker 的分析先到这,前面介绍的 zk 中注册的结构是与 Client 相关的,这里也向大家介绍了一下,以后分析 Client 的时候,该结构将不再介绍了。

相关标签: MetaQ