本篇内容介绍了“Flume整体流程是怎样的”的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！
专注于为中小企业提供成都做网站、成都网站设计服务,电脑端+手机端+微信端的三站合一,更高效的管理,为中小企业鄞州免费做网站提供优质的服务。我们立足成都，凝聚了一批互联网行业人才，有力地推动了上1000家企业的稳健成长，帮助中小企业通过网站建设实现规模扩充和转变。
不管是Source还是Sink都依赖Channel，那么启动时应该先启动Channel然后再启动Source或Sink即可。
Flume有两种启动方式：使用EmbeddedAgent内嵌在Java应用中或使用Application单独启动一个进程，此处我们已Application分析为主。
首先进入org.apache.flume.node.Application的main方法启动：
//1、设置默认值启动参数、参数是否必须的
Options options = new Options();
Option option = new Option("n", "name", true, "the name of this agent");
option.setRequired(true);
options.addOption(option);
option = new Option("f", "conf-file", true,
"specify a config file (required if -z missing)");
option.setRequired(false);
options.addOption(option);
//2、接着解析命令行参数
CommandLineParser parser = new GnuParser();
CommandLine commandLine = parser.parse(options, args);
String agentName = commandLine.getOptionValue('n');
boolean reload = !commandLine.hasOption("no-reload-conf");
if (commandLine.hasOption('z') || commandLine.hasOption("zkConnString")) {
isZkConfigured = true;
}
if (isZkConfigured) {
//3、如果是通过ZooKeeper配置，则使用ZooKeeper参数启动，此处忽略，我们以配置文件讲解
} else {
//4、打开配置文件，如果不存在则快速失败
File configurationFile = new File(commandLine.getOptionValue('f'));
if (!configurationFile.exists()) {
throw new ParseException(
"The specified configuration file does not exist: " + path);
}
List components = Lists.newArrayList();
if (reload) { //5、如果需要定期reload配置文件，则走如下方式
//5.1、此处使用Guava提供的事件总线
EventBus eventBus = new EventBus(agentName + "-event-bus");
//5.2、读取配置文件，使用定期轮训拉起策略，默认30s拉取一次
PollingPropertiesFileConfigurationProvider configurationProvider =
new PollingPropertiesFileConfigurationProvider(
agentName, configurationFile, eventBus, 30);
components.add(configurationProvider);
application = new Application(components); //5.3、向Application注册组件
//5.4、向事件总线注册本应用，EventBus会自动注册Application中使用@Subscribe声明的方法
eventBus.register(application);
} else { //5、配置文件不支持定期reload
PropertiesFileConfigurationProvider configurationProvider =
new PropertiesFileConfigurationProvider(
agentName, configurationFile);
application = new Application();
//6.2、直接使用配置文件初始化Flume组件
application.handleConfigurationEvent(configurationProvider
.getConfiguration());
}
}
//7、启动Flume应用
application.start();
//8、注册虚拟机关闭钩子，当虚拟机关闭时调用Application的stop方法进行终止
final Application appReference = application;
Runtime.getRuntime().addShutdownHook(new Thread("agent-shutdown-hook") {
@Override
public void run() {
appReference.stop();
}
}); 以上流程只提取了核心代码中的一部分，比如ZK的实现直接忽略了，而Flume启动大体流程如下：
1、读取命令行参数；
2、读取配置文件；
3、根据是否需要reload使用不同的策略初始化Flume；如果需要reload，则使用Guava的事件总线实现，Application的handleConfigurationEvent是事件订阅者，PollingPropertiesFileConfigurationProvider是事件发布者，其会定期轮训检查文件是否变更，如果变更则重新读取配置文件，发布配置文件事件变更，而handleConfigurationEvent会收到该配置变更重新进行初始化；
4、启动Application，并注册虚拟机关闭钩子。
handleConfigurationEvent方法比较简单，首先调用了stopAllComponents停止所有组件，接着调用startAllComponents使用配置文件初始化所有组件：
@Subscribe
public synchronized void handleConfigurationEvent(MaterializedConfiguration conf) {
stopAllComponents();
startAllComponents(conf);
}MaterializedConfiguration存储Flume运行时需要的组件：Source、Channel、Sink、SourceRunner、SinkRunner等，其是通过ConfigurationProvider进行初始化获取，比如PollingPropertiesFileConfigurationProvider会读取配置文件然后进行组件的初始化。
对于startAllComponents实现大体如下：
//1、首先启动Channel
supervisor.supervise(Channels,
new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START);
//2、确保所有Channel是否都已启动
for(Channel ch: materializedConfiguration.getChannels().values()){
while(ch.getLifecycleState() != LifecycleState.START
&& !supervisor.isComponentInErrorState(ch)){
try {
Thread.sleep(500);
} catch (InterruptedException e) {
Throwables.propagate(e);
}
}
}
//3、启动SinkRunner
supervisor.supervise(SinkRunners,
new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START);
//4、启动SourceRunner
supervisor.supervise(SourceRunner,
new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START);
//5、初始化监控服务
this.loadMonitoring();从如下代码中可以看到，首先要准备好Channel，因为Source和Sink会操作它，对于Channel如果初始化失败则整个流程是失败的；然后启动SinkRunner，先准备好消费者；接着启动SourceRunner开始进行采集日志。此处我们发现有两个单独的组件LifecycleSupervisor和MonitorService，一个是组件守护哨兵，一个是监控服务。守护哨兵对这些组件进行守护，假设出问题了默认策略是自动重启这些组件。
对于stopAllComponents实现大体如下：
//1、首先停止SourceRunner supervisor.unsupervise(SourceRunners); //2、接着停止SinkRunner supervisor.unsupervise(SinkRunners); //3、然后停止Channel supervisor.unsupervise(Channels); //4、最后停止MonitorService monitorServer.stop();
此处可以看出，停止的顺序是Source、Sink、Channel，即先停止生产，再停止消费，最后停止管道。
Application中的start方法代码实现如下：
public synchronized void start() {
for(LifecycleAware component : components) {
supervisor.supervise(component,
new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START);
}
}其循环Application注册的组件，然后守护哨兵对它进行守护，默认策略是出现问题会自动重启组件，假设我们支持reload配置文件，则之前启动Application时注册过PollingPropertiesFileConfigurationProvider组件，即该组件会被守护哨兵守护着，出现问题默认策略自动重启。
而Application关闭执行了如下动作：
public synchronized void stop() {
supervisor.stop();
if(monitorServer != null) {
monitorServer.stop();
}
}即关闭守护哨兵和监控服务。
到此基本的Application分析结束了，我们还有很多疑问，守护哨兵怎么实现的。
整体流程可以总结为：
1、首先初始化命令行配置；
2、接着读取配置文件；
3、根据是否需要reload初始化配置文件中的组件；如果需要reload会使用Guava事件总线进行发布订阅变化；
4、接着创建Application，创建守护哨兵，并先停止所有组件，接着启动所有组件；启动顺序：Channel、SinkRunner、SourceRunner，并把这些组件注册给守护哨兵、初始化监控服务；停止顺序：SourceRunner、SinkRunner、Channel；
5、如果配置文件需要定期reload，则需要注册Polling***ConfigurationProvider到守护哨兵；
6、最后注册虚拟机关闭钩子，停止守护哨兵和监控服务。
轮训实现的SourceRunner 和SinkRunner会创建一个线程进行工作，之前已经介绍了其工作方式。接下来我们看下守护哨兵的实现。
首先创建LifecycleSupervisor：
