Flume-ng 1.6启动过程源码分析(一)

 启动部分主要分为四块

 

1. 命令行参数的载入，这部分用的common cli实现

2. 对于可选的zk配置的加载

3. 对于flume-ng配置的加载，这里利用EventBus实现配置文件动态加载

4. 组件的启动

简单介绍下EventBus，它是Guava提供的消息发布-订阅类库，机制与观察者模式类似。

 

下面主要介绍上面提到的第3和4部分。

首先flume-ng的启动类是org.apache.flume.node.Application 位于flume-ng-node的项目里

首先介绍ConfigurationProvider这个接口，它仅提供一个方法

public MaterializedConfiguration getConfiguration();

 

 

下面就从加载flume-ng配置开始，介绍启动流程

总体来说三大部分:

1. 初始化Source Channel Sink的工厂对象

2. 创建具体的Source Channel Sink实例

3. 组件启动

详细介绍:

1. 创建Source Channel Sink的工厂对象

进入启动类org.apache.flume.node.Application的main方法

 

已else流程为例，看一下PropertiesFileConfigurationProvider初始化时做了哪些

 

调用了父类的构造，从之前的继承关系图上可以看到，其父类AbstractConfigurationProvider，

继续看一下父类的构造，初始化工厂类及channelCache

 

 

2. 创建具体的Source Channel Sink实例

在创建完PropertiesFileConfigurationProvider的实例configurationProvider后，代码中调用了configurationProvider.getConfiguration()，由于继承的关系，实际上调用的AbstractConfigurationProvider的getConfiguration，看一下它做了什么

 

读取了Flume的配置，加载组件，加载好的组件会分别放到channelComponentMap，sourceRunnerMap，sinkRunnerMap三个Map中。

已loadSources为例，调查下如何加载Source

loadSources主要做了四件事情，

 

1. 利用之前实例化的DefaultSourceFactory对象实例化source，其中create方法会先根据type判断type是内置还是自定制，然后通过反射的方式创建实例

2. 创建ChannelProcessor对象，加载interceptor配置

3. 返回SourceRunner对象，它分为两类PollableSourceRunner和EventDrivenSourceRunner，这里的判断依据是source实现了PollableSource接口还是EventDrivenSource接口，当然在代码中是通过 instanceof来判断
   

这里拓展一下，Source分为PollableSource和EventDrivenSource，EventDrivenSource可以理解时间驱动的，这样的source有HttpSource，ExecSource等，PollableSource是需要外部的驱动去轮询的Source，例如KafkaSource。