Flume 1.7 源码分析
本帖最后由 desehawk 于 2017-2-27 17:34 编辑问题导读
1.如何配置maven获取源码？
2.本文认为flume启动包含哪两个步骤？
3.源码如何实现获取启动配置的？
static/image/hrline/4.gif
1源码编译
1 说明
Flume是Cloudera提供的一个高可用的，高可靠的，分布式的海量日志采集、聚合和传输的系统，目前已经是Apache的一个子项目。Flume是一个专用工具被设计为旨在往HDFS、HBase发送数据。它对HDFS有特殊的优化，并且集成了Hadoop的安全特性。本文将详细分析Flume核心模块的源码实现。
2 下载、编译
2.1 源码检出
Flume源码的Git地址是：https://git-wip-us.apache.org/repos/asf/flume.git。本文采用的分支是flume-1.7。
2.2 源码编译
Flume采用Maven进行构建。本文采用IDEA作为编译、调试工具。
2.2.1 Maven依赖下载
配置阿里云的Maven服务器，速度较快，并可完成绝大多数包的下载。
<mirror>
<id>nexus-aliyun</id>
<mirrorOf>*</mirrorOf>
<name>Nexus aliyun</name>
<url>http://maven.aliyun.com/nexus/content/groups/public</url>
</mirror>
使用阿里云Maven服务器，仍然有一些包无法解决，例如FlumeAuth模块的hadoop-minikdc和SolrSink的kite-morphlines-all等。由于这部分缺失的包，基本都是属于flume的插件模块之中，可直接关闭这些插件模块，不会影响主程序的执行和源码阅读。同时，还可移除其他的你不需要使用的source和sink和channel插件模块，可加快编译速度。导入好的项目如下图所示：
其中比较重要的就是上图圈起来的模块了，其中flume-ng-core存放了最核心部分的代码，包含基础的Source、Channel、Sink等；flume-ng-node则是存放了程序启动的代码（入口函数）。其他可能会用到的模块就是flume-ng-sources、flume-ng-channels、flume-ng-sinks，这3个模块存放了非必须的flume组件（flume-ng-core中未包含的），里面有些组件也是很常用的。2.2.2 编译打包使用mvn clean install -Dmaven.test.skip=true进行打包，看到BUILD SUCCESS则代表编译成功，可进行下一步。本步骤一定要有，不然debug的时候会报某些类找不到，这些类大多数是由avro定义的文件编译后才会生成的class文件。2.2.3 简单测试在TestMemoryChannel类右键选择Run ‘TestMemoryChannel’运行JUnit单元测试。检查是否有报错。
2整体架构
Flume有三大组件：Source、Channel、Sink。
[*]Source就是数据来源，例如Web Server产生日志后，可使用ExecSource执行tail -F命令后不断监听日志文件新生成的数据，然后传给Channel。
[*]Channel就是一个缓存队列，由于读取数据和写入数据的速度可能不匹配，假如用同步完成的方式可能效率低下，所以Source把数据写到Channel这个队列里面，Sink再用另外的线程去读取。
[*]Sink就是最终的存储，例如可以是HDFS或LOG文件输出等，Sink负责去Channel里面读取数据，并存储。
在程序启动时，会启动所有的SourceRunner、Channel、SinkRunner。其中Channel的启动，没做什么特别的事情，就是初始化一下状态、创建一下计数器，算做一个被动的角色。比较重要的是SourceRunner和SinkRunner。
[*]SourceRunner会调用Source的start方法。以ExecSource为例，其start方法就是启动一个线程，去不断获取标准输出流写入一个列表（eventList），同时再启动一个线程去定期批量地把列表中的数据往Channel发，如下图所示。
[*]SinkRunner则是不断循环调用SinkProcess的process的方法，SinkProcess有几种类型，用于决定选择哪个Sink进行存储（Sink可以有多个），选择了Sink后，调用其process方法。Sink的process方法，主要做的就是去Channel中读取数据，并写入对应的存储，如下图所示。
3 程序入口
启动Flume的过程可以简单分为2个步骤：
1. 获取相关配置文件（一般来说就是flume-conf.properties）。
2. 启动各组件。不特别说明，本文中的组件是指实现了LifecycleAware接口的类的对象，一般就是Source、Channel、Sink这3种对象。
3.1 获取启动配置
3.1.1 Main函数
启动Flume的Main函数在flume-ng-node模块的org.apache.flume.node.Application。该函数的功能可以简单划分为以下三个步骤：
1. 使用commons.cli类获取命令行参数（就是启动时传入的参数）
2. 根据启动参数确定的读取配置的方式。读取配置的方式总共有4种，分别根据配置是保存在zookeeper上还是本地properties文件、以及是否reload（自动重载配置文件）分为4种方式。
3. 根据相应的配置启动程序，并注册关闭钩子。
接下来以properties文件、不重载的方式为例，主要的代码如下：
PropertiesFileConfigurationProvider configurationProvider =
new PropertiesFileConfigurationProvider(agentName, configurationFile);
//创建Application对象，包含初始化组件列表（components），初始化LifecycleSupervisor。
application = new Application();
application.handleConfigurationEvent(configurationProvider.getConfiguration());
//start方法用于检查所有组件是否是启动状态，如果不是则启动该组件。
application.start();
//监听程序关闭事件，用于当程序被kill后能够执行一些清理工作。
final Application appReference = application;
Runtime.getRuntime().addShutdownHook(new Thread("agent-shutdown-hook") {
public void run() {
appReference.stop();
}
});
上面的代码，有两处比较关键：
configurationProvider.getConfiguration()会返回一个MaterializedConfiguration类型的对象，用于从文件形式的配置转为物化的配置，即包含实际的channel、sinkRunner等对象的实例，在“物化配置”一节分析。
handleConfigurationEvent用于停止所有components，并使用新的配置进行启动，在“使用新配置重启”一节分析。
4.1.2 物化配置
configurationProvider.getConfiguration()方法主要做了以下两件事：
1. 读取配置文件（flume-conf.properties），保存在AgentConfiguration对象中。
public static class AgentConfiguration {
private final String agentName;
private String sources;
private String sinks;
private String channels;
private String sinkgroups;
private final Map<String, ComponentConfiguration> sourceConfigMap;
private final Map<String, ComponentConfiguration> sinkConfigMap;
private final Map<String, ComponentConfiguration> channelConfigMap;
private final Map<String, ComponentConfiguration> sinkgroupConfigMap;
private Map<String, Context> sourceContextMap;
private Map<String, Context> sinkContextMap;
private Map<String, Context> channelContextMap;
private Map<String, Context> sinkGroupContextMap;
private Set<String> sinkSet;
private Set<String> sourceSet;
private Set<String> channelSet;
private Set<String> sinkgroupSet;
}
到这个步骤还仅仅是做好了分类的文本形式的配置项。
2. 创建出配置中的各组件实例，并添加到MaterializedConfiguration实例中。
public interface MaterializedConfiguration {
public void addSourceRunner(String name, SourceRunner sourceRunner);
public void addSinkRunner(String name, SinkRunner sinkRunner);
public void addChannel(String name, Channel channel);
public ImmutableMap<String, SourceRunner> getSourceRunners();
public ImmutableMap<String, SinkRunner> getSinkRunners();
public ImmutableMap<String, Channel> getChannels();
}
在这个实例中，可以获取配置文件中配置的所有的source、channel、sink，并且是“物化”的，即可以直接取得相关组件的实例。
4.2 启动所有组件
4.2.1 使用新配置重启
有了上面的MaterializedConfiguration实例，我们就可以启动组件了。
在handleConfigurationEvent方法中，首先会停止所有组件，然后再启动所有组件。
stopAllComponents();
startAllComponents(conf); //这里的conf就是上节的MaterializedConfiguration。
在startAllComponents方法中，会遍历组件列表（SourceRunners、SinkRunners、Channels），分别调用supervise方法。以Channel为例：
for (Entry<String, Channel> entry :
materializedConfiguration.getChannels().entrySet()) {
try {
logger.info("Starting Channel " + entry.getKey());
supervisor.supervise(entry.getValue(),
new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START);
} catch (Exception e) {
logger.error("Error while starting {}", entry.getValue(), e);
}
}
这个supervise方法简单来说，就是将相应组件的状态转化为期望的状态。例如上面代码中的LifecycleState.START就是期望的状态。
4.2.2 LifecycleSupervisor
上节的supervisor是一个LifecycleSupervisor对象。前面有说到，在创建Application的时候初始化了一个LifecycleSupervisor对象，就是这里的supervisor。这个对象，我理解为各组件生命周期的管理者，用于实时监控所有组件的状态，如果不是期望的状态（desiredState），则进行状态转换。
上节的代码中调用了supervisor.supervise方法，接下来分析一下supervise这个方法：
public synchronized void supervise(LifecycleAware lifecycleAware,
SupervisorPolicy policy, LifecycleState desiredState) {
//省略状态检查的代码
Supervisoree process = new Supervisoree();
process.status = new Status();
process.policy = policy;
process.status.desiredState = desiredState;
process.status.error = false;
MonitorRunnable monitorRunnable = new MonitorRunnable();
monitorRunnable.lifecycleAware = lifecycleAware;
monitorRunnable.supervisoree = process;
monitorRunnable.monitorService = monitorService;
supervisedProcesses.put(lifecycleAware, process);
ScheduledFuture<?> future = monitorService.scheduleWithFixedDelay(
monitorRunnable, 0, 3, TimeUnit.SECONDS);
monitorFutures.put(lifecycleAware, future);
}
由于所有的组件都实现了LifecycleAware接口，所以这里的supervise方法传入的是LifecycleAware接口的对象。
可以看到创建了一个Supervisoree对象，顾名思义，就是被监控的的对象，该对象有以下几种状态：IDLE, START, STOP, ERROR。
scheduleWithFixedDelay每隔3秒触发一次监控任务（monitorRunnable）。
4.2.3 MonitorRunnable
在MonitorRunnable中主要是检查组件的状态，并实现从lifecycleState到desiredState的转变。
switch (supervisoree.status.desiredState) {
case START:
try {
lifecycleAware.start();
} catch (Throwable e) {省略}
break;
case STOP:
try {
lifecycleAware.stop();
} catch (Throwable e) {省略}
break;
default:
logger.warn("I refuse to acknowledge {} as a desired state", supervisoree.status.desiredState);
}
到这里为止，可以看到监控的进程，调用了组件自己的start和stop方法来启动、停止。前面有提到有3种类型的组件，SourceRunner、Channel、SinkRunner，而Channel的start只做了初始化计数器，没什么实质内容，所以接下来从SourceRunner的启动（从Source写数据到Channel）和SinkRunner的启动（从Channel获取数据写入Sink）来展开说明。
作者：Lnho
来自：csdn
谢谢分享 感谢分享 这个好，写的很不错，需要学习
页:
[1]