Overview
-
source
-
Source和Sink之间是异步的，sink只需要监听自己关系的Channel的变化即可。
-
sink存在写失败的情况，flume提供了如下策略：
-
默认是一个sink，若写入失败，则该事务失败，稍后重试。
-
故障转移策略：给多个sink定义优先级，失败时会路由到下一个优先级的sink。sink只要抛出一次异常就会被认为是失败了，则从存活Sink中移除，然后指数级时间等待重试，默认是等待1s开始重试，最大等待重试时间是30s.
-
-
flume还提供了负载均衡策略：默认提供轮训和随机两种算法。通过抽象一个类似ChannelSelector的SinkSelector进行选择。
-
以上，对于Source和sink如何异步、channel如何实现事务机制，详见后面的具体源码分析。
The whole process
-
首先是flume的启动， 提供了两种启动方式：使用EmbeddedAgent内嵌在Java应用中或使用Application单独启动一个进程。 一般使用Application起一个进程比较多，我们这里也主要分析这种方式。
-
程序入口：org.apache.flume.node.Application的main方法。
-
注：因为暂时还没有了解到Zookeeper原理，所以这里关于ZK的部分就跳过了。
-
flume启动流程大致如下：
-
设置默认值启动参数，参数是否是必须的
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); ...... -
解析命令行参数
if (commandLine.hasOption('h')) { new HelpFormatter().printHelp("flume-ng agent", options, true); return; } String agentName = commandLine.getOptionValue('n'); boolean reload = !commandLine.hasOption("no-reload-conf"); // 是否reload配置文件 if (commandLine.hasOption('z') || commandLine.hasOption("zkConnString")) { isZkConfigured = true; } -
Zookepper相关：暂时略
-
打开配置文件
if (isZkConfigured) { ... // 若配置了zk，则使用zk参数启动 } else { // 打开配置文件，如果不存在则快速失败 File configurationFile = new File(commandLine.getOptionValue('f')); // 确保没有配置文件的时候agent会启动失败 if (!configurationFile.exists()) { ...// If command line invocation, then need to fail fast } List<LifecycleAware> components = Lists.newArrayList(); // 若需要定期reload配置文件 if (reload) { // 使用EventBus事件总线, to allow publish-subscribe-style communication EventBus eventBus = new EventBus(agentName + "-event-bus"); // 读取配置文件，使用定期轮训拉起策略，默认30s拉取一次 PollingPropertiesFileConfigurationProvider configurationProvider = new PollingPropertiesFileConfigurationProvider( agentName, configurationFile, eventBus, 30); components.add(configurationProvider); // 向Application注册组件 application = new Application(components); // 向EventBus注册本应用，EB会自动注册Application中使用@Subscribe声明的方法 // TODO: EventBus, and why reload configuration eventBus.register(application); } else { // 若配置文件不支持定期reload PropertiesFileConfigurationProvider configurationProvider = new PropertiesFileConfigurationProvider(agentName, configurationFile); application = new Application(); // 直接使用配置文件初始化Flume组件 application.handleConfigurationEvent(configurationProvider.getConfiguration()); } } -
reload conf：若需要reload，则使用事件总线EventBus实现，Application的handleConfigurationEvent是事件订阅者，PollingPropertiesFileConfigurationProvider是事件发布者，其会定期轮训检查文件是否变更，如果变更则重新读取配置文件，发布配置文件事件变更，而handleConfigurationEvent会收到该配置变更重新进行初始化。
-
handleConfigurationEvent：
@Subscribe public synchronized void handleConfigurationEvent(MaterializedConfiguration conf) { // MaterializedConfiguration存储Flume运行时需要的组件：Source、Channel、Sink、SourceRunner、SinkRunner等。其是通过ConfigurationProvider进行初始化获取，比如PollingPropertiesFileConfigurationProvider会读取配置文件然后进行组件的初始化。 stopAllComponents(); // 停止所有组件 startAllComponents(conf);// 使用配置文件初始化所有组件 } -
startAllComponents
-
要首先启动channels，等待所有channels启动才能继续。然后启动SinkRunner，准备好消费者。最后启动SourceRunner开始进行采集日志。
-
LifecycleSupervisor是组件守护哨兵，对这些组件进行守护，出问题时默认策略是自动重启。
-
这里的启动都是supervisor.supervise(entry.getValue(),new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START); 这是如何启动的，我们后面再介绍。
private void startAllComponents(MaterializedConfiguration materializedConfiguration) { logger.info("Starting new configuration:{}", materializedConfiguration); this.materializedConfiguration = materializedConfiguration; // 启动channels。 for (Entry<String, Channel> entry : materializedConfiguration.getChannels().entrySet()) { try { logger.info("Starting Channel " + entry.getKey()); // TODO: LifecycleSupervisor启动 // new SupervisorPolicy.AlwaysRestartPolicy():使用失败时总是重启的策略 // LifecycleState.START: 初始化组件默认状态为START supervisor.supervise(entry.getValue(), new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START); } catch (Exception e) { logger.error("Error while starting {}", entry.getValue(), e); } } /* * Wait for all channels to start. */ for (Channel ch : materializedConfiguration.getChannels().values()) { while (ch.getLifecycleState() != LifecycleState.START && !supervisor.isComponentInErrorState(ch)) { try { logger.info("Waiting for channel: " + ch.getName() + " to start. Sleeping for 500 ms"); Thread.sleep(500); } catch (InterruptedException e) { logger.error("Interrupted while waiting for channel to start.", e); Throwables.propagate(e); } } } // 启动sinkRunner for (Entry<String, SinkRunner> entry : materializedConfiguration.getSinkRunners().entrySet()) { try { logger.info("Starting Sink " + entry.getKey()); supervisor.supervise(entry.getValue(), new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START); } catch (Exception e) { logger.error("Error while starting {}", entry.getValue(), e); } } // 启动SourceRunner TODO: SourceRunner & SinkRunner for (Entry<String, SourceRunner> entry : materializedConfiguration.getSourceRunners().entrySet()) { try { logger.info("Starting Source " + entry.getKey()); supervisor.supervise(entry.getValue(), new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START); } catch (Exception e) { logger.error("Error while starting {}", entry.getValue(), e); } } this.loadMonitoring(); } -
-
之后main函数调用了
application.start();/** 其循环Application注册的组件，然后守护哨兵对它进行守护，默认策略是出现问题会自动重启组件，假设我们支持reload配置文件，则之前启动Application时注册过PollingPropertiesFileConfigurationProvider组件，即该组件会被守护哨兵守护着，出现问题默认策略自动重启 **/ public synchronized void start() { // private final List<LifecycleAware> components; for (LifecycleAware component : components) { // private final LifecycleSupervisor supervisor; supervisor.supervise(component, new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START); } }相应的stop函数。首先是main函数中：
final Application appReference = application; // Runtinme.getRuntime(): Returns the runtime object associated with the current Java application. /** addShutdownHook: 注册一个新的虚拟机关闭钩子。 虚拟机shutdown有两种情况：1）当最后一个非守护进行户外那个退出或调用system.exit时，程序正常退出；2）JVM通过ctrl-c等被用户中断。 **/ Runtime.getRuntime().addShutdownHook(new Thread("agent-shutdown-hook") { @Override public void run() { appReference.stop(); } }); public synchronized void stop() { // 关闭守护哨兵和监控服务。 supervisor.stop(); if (monitorServer != null) { monitorServer.stop(); } } -
至此，Application整个流程就分析完了。
-
-
整体流程可以总结为：
-
首先初始化命令行配置；
-
接着读取配置文件；
-
根据是否需要reload初始化配置文件中的组件；如果需要reload会使用EventBus进行发布订阅变化；
-
接着创建Application，创建守护哨兵LifecycleSupervisor，并先停止所有组件，接着启动所有组件；启动顺序：Channel、SinkRunner、SourceRunner，并把这些组件注册给守护哨兵、初始化监控服务MonitorService；停止顺序：SourceRunner、SinkRunner、Channel；
-
如果配置文件需要定期reload，则需要注册PollingPropertiesFileConfigurationProvider到守护哨兵；
-
最后注册虚拟机关闭钩子，停止守护哨兵和监控服务。
-
LifecycleSupervisor
-
守护哨兵，负责监控和重启组件
-
My: 所有需要被监控和重启的组件都应implements LifecycleAware
public class LifecycleSupervisor implements LifecycleAware { public LifecycleSupervisor() { lifecycleState = LifecycleState.IDLE; // 存放被监控的组件 supervisedProcesses = new HashMap<LifecycleAware, Supervisoree>(); // 存放正在被监控的组件 monitorFutures = new HashMap<LifecycleAware, ScheduledFuture<?>>(); // 创建监控服务线程池 monitorService = new ScheduledThreadPoolExecutor(10, new ThreadFactoryBuilder().setNameFormat( "lifecycleSupervisor-" + Thread.currentThread().getId() + "-%d") .build()); monitorService.setMaximumPoolSize(20); monitorService.setKeepAliveTime(30, TimeUnit.SECONDS); // 定期清理被取消的组件 purger = new Purger(); // 默认不进行清理 needToPurge = false; } ... // start() & stop()... // 进行组件守护 public synchronized void supervise(LifecycleAware lifecycleAware, SupervisorPolicy policy, LifecycleState desiredState) { if (this.monitorService.isShutdown() || this.monitorService.isTerminated() || this.monitorService.isTerminating()) { ...// 如果哨兵已停止则抛出异常 } // 初始化守护组件 Supervisoree process = new Supervisoree(); process.status = new Status(); // 默认策略是失败重启 process.policy = policy; process.status.desiredState = desiredState; // 初始化组件默认状态，一般为START process.status.error = false; // 组件监控器，用于定时获取组件的最新状态，或重启组件。后面会介绍MonitorRunnable具体做什么。 MonitorRunnable monitorRunnable = new MonitorRunnable(); monitorRunnable.lifecycleAware = lifecycleAware; monitorRunnable.supervisoree = process; monitorRunnable.monitorService = monitorService; supervisedProcesses.put(lifecycleAware, process); // 以固定时间间隔执行monitorRunnable线程 // scheduleWithFixedDelay: Creates and executes a periodic action. If any execution of the task encounters an exception, subsequent executions are suppressed. Otherwise, the task will only terminate via cancellation or termination of the executor. // 所以需要把所有异常捕获，才能保证定时任务继续执行。 ScheduledFuture<?> future = monitorService.scheduleWithFixedDelay( monitorRunnable, 0, 3, TimeUnit.SECONDS); monitorFutures.put(lifecycleAware, future); } -
MonitorRunnable：负责进行组件状态迁移或组件故障恢复
public static class MonitorRunnable implements Runnable { public ScheduledExecutorService monitorService; public LifecycleAware lifecycleAware; public Supervisoree supervisoree; @Override public void run() { long now = System.currentTimeMillis(); try { if (supervisoree.status.firstSeen == null) { logger.debug("first time seeing {}", lifecycleAware); supervisoree.status.firstSeen = now; // 记录第一次状态查看时间 } supervisoree.status.lastSeen = now; // 记录最后一次状态查看时间 synchronized (lifecycleAware) { // 如果守护组件被丢弃或出错了，则直接返回 if (supervisoree.status.discard) { // 也就是此时已经调用了unsupervise logger.info("Component has already been stopped {}", lifecycleAware); return; } else if (supervisoree.status.error) { logger.info("Component {} is in error state, and Flume will not" + "attempt to change its state", lifecycleAware); return; } // 更新最后一次查看到的状态 supervisoree.status.lastSeenState = lifecycleAware.getLifecycleState(); // 如果组件的状态和守护组件看到的状态不一致，则以守护组件的状态为准，然后进行初始化 if (!lifecycleAware.getLifecycleState().equals( supervisoree.status.desiredState)) { switch (supervisoree.status.desiredState) { // 如果是启动状态，则启动组件。# 最开始的时候组件应该就是这么启动的 case START: try { lifecycleAware.start(); } catch (Throwable e) { logger.error("Unable to start " + lifecycleAware + " - Exception follows.", e); if (e instanceof Error) { // This component can never recover, shut it down. supervisoree.status.desiredState = LifecycleState.STOP; try { lifecycleAware.stop(); logger.warn("Component {} stopped, since it could not be" + "successfully started due to missing dependencies", lifecycleAware); } catch (Throwable e1) { logger.error("Unsuccessful attempt to " + "shutdown component: {} due to missing dependencies." + " Please shutdown the agent" + "or disable this component, or the agent will be" + "in an undefined state.", e1); supervisoree.status.error = true; if (e1 instanceof Error) { throw (Error) e1; } // Set the state to stop, so that the conf poller can // proceed. } } supervisoree.status.failures++; } break; case STOP: try { lifecycleAware.stop(); } catch (Throwable e) { logger.error("Unable to stop " + lifecycleAware + " - Exception follows.", e); if (e instanceof Error) { throw (Error) e; } supervisoree.status.failures++; } break; default: logger.warn("I refuse to acknowledge {} as a desired state", supervisoree.status.desiredState); } if (!supervisoree.policy.isValid(lifecycleAware, supervisoree.status)) { logger.error( "Policy {} of {} has been violated - supervisor should exit!", supervisoree.policy, lifecycleAware); } } } } catch (Throwable t) { logger.error("Unexpected error", t); } logger.debug("Status check complete"); } }
Source
SourceRunner
-
首先是SourceRunner，它控制how a source is driven。
-
它是一个用来实例化derived classes(派生类)的抽象类。 根据指定的source，来通过其内的static factory method 来实例化runner。
// 根据指定source的类型来实例化一个source runner的静态工厂方法 // 输入是要运行的source，返回可以运行指定source的runner public static SourceRunner forSource(Source source) { SourceRunner runner = null; if (source instanceof PollableSource) { runner = new PollableSourceRunner(); ((PollableSourceRunner) runner).setSource((PollableSource) source); } else if (source instanceof EventDrivenSource) { runner = new EventDrivenSourceRunner(); ((EventDrivenSourceRunner) runner).setSource((EventDrivenSource) source); } else { throw new IllegalArgumentException("No known runner type for source " + source); } return runner; }
EventDrivenSourceRunner
-
starts、stops and manages EventDrivenSource event-driven sources
-
其内有如下几个方法：
-
构造方法
public EventDrivenSourceRunner() { lifecycleState = LifecycleState.IDLE; } -
start()
@Override public void start() { Source source = getSource(); //获取Source ChannelProcessor cp = source.getChannelProcessor(); //Channel处理器 cp.initialize(); //初始化Channel处理器 source.start(); //启动Source lifecycleState = LifecycleState.START; //本组件状态改成启动状态 } -
stop()、toString()、getLifecycleState()
-
PollableSourceRunner
public class PollableSourceRunner extends SourceRunner {
@Override
public void start() {
PollableSource source = (PollableSource) getSource(); //获取Source
ChannelProcessor cp = source.getChannelProcessor(); //Channel处理器
cp.initialize(); //初始化channel处理器
source.start(); //启动source
runner = new PollingRunner(); //新建一个PollingRunner线程来拉取数据
runner.source = source;
runner.counterGroup = counterGroup;
runner.shouldStop = shouldStop;
runnerThread = new Thread(runner);
runnerThread.setName(getClass().getSimpleName() + "-" +
source.getClass().getSimpleName() + "-" + source.getName());
runnerThread.start();
lifecycleState = LifecycleState.START;
}
}
-
PollingRunner线程
@Override
public void run() {
while (!shouldStop.get()) { //如果没有停止，则一直在死循环运行
counterGroup.incrementAndGet("runner.polls"); //原子操作
try {
//调用PollableSource的process方法进行轮训拉取，然后判断是否遇到了失败补偿
if (source.process().equals(PollableSource.Status.BACKOFF)) {/
counterGroup.incrementAndGet("runner.backoffs");
//失败补偿时暂停线程处理，等待超时时间之后重试
Thread.sleep(Math.min(
counterGroup.incrementAndGet("runner.backoffs.consecutive")
* source.getBackOffSleepIncrement(), source.getMaxBackOffSleepInterval()));
} else {
counterGroup.set("runner.backoffs.consecutive", 0L);
}
} catch (InterruptedException e) {
}
}
}
}
}
-
TODO
Source
public interface Source extends LifecycleAware, NamedComponent {
public void setChannelProcessor(ChannelProcessor channelProcessor);
public ChannelProcessor getChannelProcessor();
}
-
继承了LifecycleAware接口，然后只提供了ChannelProcessor的setter和getter接口。其中：
-
它的的所有逻辑的实现应该在LifecycleAware接口的start和stop中实现；
-
ChannelProcessor用来进行日志流的过滤和Channel的选择及调度。
-
-
由上述的Runner我们知道，Source 提供了两种机制： PollableSource （轮训拉取）和 EventDrivenSource （事件驱动）
-
Source作用就是监听日志，采集，然后交给ChannelProcessor处理。
EventDrivenSource
-
事件驱动型source不需要外部driver来获取event，EventDriven是一个implement Source的空接口。
-
从这里开始～～～‘
Channel
-
通过 Channel 实现了 Source 和 Sink 的解耦，可以实现多对多的关联，和 Source 、 Sink 的异步化
-
Channel exposes a transaction interface that can be used by its clients to ensure automic put(Event) and take() semantics.
ChannelProcesoor
-
前面我们了解到Source采集日志后会交给ChannelProcessor处理，so接下来我们从ChannelProcessor入手，其依赖如下组件：
private final ChannelSelector selector; //Channel选择器,.flume.ChannelSelector private final InterceptorChain interceptorChain; //拦截器链,.flume.interceptor.InterceptorChain private ExecutorService execService; //用于实现可选Channel的ExecutorService，默认是单线程实现 [注：这个我在某个博客上看到的，但这个组件我在ChannelProcessor中没有搜到]
-
我们来看ChannelProcessor是如何处理Event的：
// Attempts to put the given event into each configured channel public void processEvent(Event event) { event = interceptorChain.intercept(event); //首先进行拦截器链过滤,TODO:intercep... // InterceptorChain实现了Interceptor接口，调用a list of other Interceptors. 实现event的过滤和加工。具体见后面 if (event == null) { return; } // Process required channels //通过Channel选择器获取必须成功处理的Channel，然后事务中执行. List<Channel> requiredChannels = selector.getRequiredChannels(event); for (Channel reqChannel : requiredChannels) { Transaction tx = reqChannel.getTransaction(); // 继承自Channel接口的类要实现getTransaction()方法，TODO:getTransaction Preconditions.checkNotNull(tx, "Transaction object must not be null"); try { tx.begin(); //开始事务 reqChannel.put(event); // 将event放到reqChannel tx.commit(); //提交事务 } catch (Throwable t) { tx.rollback(); // 如果捕捉到throwable(including Error & Exception)，则回滚事务 if (t instanceof Error) { LOG.error("Error while writing to required channel: " + reqChannel, t); throw (Error) t; } else if (t instanceof ChannelException) { throw (ChannelException) t; } else { throw new ChannelException("Unable to put event on required " + "channel: " + reqChannel, t); //TODO: Channelexception可能会被handle，不然如何保证RequiredChannel的成功处理? } } finally { if (tx != null) { tx.close(); // 最后如果事务非空，还得关闭该事务 } } } // Process optional channels //通过Channel选择器获取可选的Channel，这些Channel失败是可以忽略，不影响其他Channel的处理 List<Channel> optionalChannels = selector.getOptionalChannels(event); for (Channel optChannel : optionalChannels) { Transaction tx = null; try { tx = optChannel.getTransaction(); tx.begin(); optChannel.put(event); tx.commit(); } catch (Throwable t) { tx.rollback(); LOG.error("Unable to put event on optional channel: " + optChannel, t); if (t instanceof Error) { throw (Error) t; } } finally { if (tx != null) { tx.close(); } } } } -
看下flume内实现的channel类
Channel接口
public interface Channel extends LifecycleAware, NamedComponent {
// put() and get() must be invoked within an active Transaction boundary
public void put(Event event) throws ChannelException;
public Event take() throws ChannelException;
// @return: the transaction instance associated with this channel
public Transaction getTransaction();
}
AbstractChannel
-
abstract class AbstractChannel implements Channel, LifecycleAware, Configurable -
实现了lifecycleStatus的改变(在构造、start()和stop()方法中)，实现了空configure()方法。没有做什么具体的channel相关的处理。
BasicChannelSemantics
-
基本Channel语义的实现，包括Transaction类的thread-local语义的实现。
public abstract class BasicChannelSemantics extends AbstractChannel {
// 1. 事务使用ThreadLocal存储，保证事务线程安全
private ThreadLocal<BasicTransactionSemantics> currentTransaction
= new ThreadLocal<BasicTransactionSemantics>();
private boolean initialized = false;
protected void initialize() {} // 2. 进行一些初始化工作
// 3.提供给实现类(子类)的创建事务的回调
// 用于new Transaction对象，该对象必须继承自BasicTransactionSemantics
// 比如MemoryChannel覆盖了该方法，方法体内new了一个实例，该实例为其内私有类MemoryTransaction，该私有类继承了BasicTransactionSemantics。
// MemoryTransaction内部用两条双向并发阻塞队列LinkedBlockingDeque实现putList和takeList。具体的稍后看，会介绍到MemoryChannel TODO
protected abstract BasicTransactionSemantics createTransaction();
// 4. 往Channel中放Event，其直接委托给事务的put方法
// 确保该thread存在一个事务，然后将put方法委托给该线程的BasicTransactionSemantics实例
@Override
public void put(Event event) throws ChannelException {
// ThreadLocal<BasicTransactionSemantics>的实例currentTransaction
// 即取得当前线程的事务实例
BasicTransactionSemantics transaction = currentTransaction.get();
Preconditions.checkState(transaction != null,
"No transaction exists for this thread");
transaction.put(event);
}
// 5.从Channel获取Event，也是直接委托给事务的take方法实现
@Override
public Event take() throws ChannelException {
BasicTransactionSemantics transaction = currentTransaction.get();
Preconditions.checkState(transaction != null,
"No transaction exists for this thread");
return transaction.take();
}
@Override
public Transaction getTransaction() {
// 1. 如果channel is not ready, then 初始化该channel
if (!initialized) {
synchronized (this) {
if (!initialized) {
initialize();
initialized = true;
}
}
}
// 2. 如果当前线程没有open的事务(无事务或已关闭)，则创建一个，并绑定到currentTransaction中
BasicTransactionSemantics transaction = currentTransaction.get();
if (transaction == null || transaction.getState().equals(
BasicTransactionSemantics.State.CLOSED)) {
transaction = createTransaction();
currentTransaction.set(transaction);
}
return transaction;
}
}
MemoryChannel
-
当写入硬盘不实际或不需要数据持久化时，推荐使用。或在单元测试时使用。
-
大部分channel会把put和take委托给事务去完成。
-
纯内存的Channel实现，整个事务操作都是在内存中完成的。
-
每个事务都有一个TakeList和PutList，分别用于存储事务相关的取数据和放数据，等事务提交时才完全同步到Channel Queue，或者失败把取数据回滚到Channel Queue。 TODO：整体理解何时commit、rollback。
public class MemoryChannel extends BasicChannelSemantics { // TODO: about factory private static Logger LOGGER = LoggerFactory.getLogger(MemoryChannel.class); ...//一些常量定义：缺省值defaultCapacity、defaultTransCapacity、byteCapacitySlotSize.. // 内部类，继承自BasicTransactionSemantics。TODO: About BasicTransactionSemantics private class MemoryTransaction extends BasicTransactionSemantics { // 每个事务都有两条双向并发阻塞队列，TODO: LinkedBlockingDeque private LinkedBlockingDeque<Event> takeList; private LinkedBlockingDeque<Event> putList; private final ChannelCounter channelCounter; ...// public MemoryTransaction(int transCapacity, ChannelCounter counter) { putList = new LinkedBlockingDeque<Event>(transCapacity); takeList = new LinkedBlockingDeque<Event>(transCapacity); channelCounter = counter; } // 将event放到putList中 // 整个doPut操作相对来说比较简单，就是往事务putList队列放入Event，如果满了则直接抛异常回滚事务；否则放入putList暂存，等事务提交时转移到Channel Queue。另外需要增加放入队列的字节数计数器，以便之后做字节容量限制 @Override protected void doPut(Event event) throws InterruptedException { // channelCounter是一个计数器，记录当前队列放入Event数、取出event数、成功数等。 channelCounter.incrementEventPutAttemptCount(); // 增加放入event计数器 // estimateEventSize计算当前Event body大小,ceil():向上取整 int eventByteSize = (int) Math.ceil(estimateEventSize(event) / byteCapacitySlotSize); // 往事务队列的putList中放入Event，如果满了，则抛异常回滚事务 if (!putList.offer(event)) { throw new ChannelException( "Put queue for MemoryTransaction of capacity " + putList.size() + " full, consider committing more frequently, " + "increasing capacity or increasing thread count"); } putByteCounter += eventByteSize; // 增加放入队列字节数计数器 } // 从Channel Queue中取event放到takeList中 @Override protected Event doTake() throws InterruptedException { channelCounter.incrementEventTakeAttemptCount(); // 如果takeList队列没有剩余容量，即当前事务已经消费了最大容量的Event if (takeList.remainingCapacity() == 0) { throw new ChannelException("Take list for MemoryTransaction, capacity " + takeList.size() + " full, consider committing more frequently, " + "increasing capacity, or increasing thread count"); } // queueStored试图获取一个信号量，超时直接返回null if (!queueStored.tryAcquire(keepAlive, TimeUnit.SECONDS)) { return null; } Event event; // 从Channel Queue获取一个Event， 对Channel Queue的操作必须加queueLock synchronized (queueLock) { event = queue.poll(); } // 因为信号量的保证，Channel Queue不应该返回null，出现了就不正常了 Preconditions.checkNotNull(event, "Queue.poll returned NULL despite semaphore " + "signalling existence of entry"); // 暂存到事务的takeList队列 takeList.put(event); // 计算当前Event body大小并增加取出队列字节数计数器 int eventByteSize = (int) Math.ceil(estimateEventSize(event) / byteCapacitySlotSize); takeByteCounter += eventByteSize; return event; } // 等事务提交时，才将当前事务的put list同步到Channel Queue @Override protected void doCommit() throws InterruptedException { // /1、计算改变的Event数量，即取出数量-放入数 int remainingChange = takeList.size() - putList.size(); if (remainingChange < 0) { // bytesRemaining是字节容量信号量，超出容量则回滚事务 if (!bytesRemaining.tryAcquire(putByteCounter, keepAlive, TimeUnit.SECONDS)) { throw new ChannelException("Cannot commit transaction. Byte capacity " + "allocated to store event body " + byteCapacity * byteCapacitySlotSize + "reached. Please increase heap space/byte capacity allocated to " + "the channel as the sinks may not be keeping up with the sources"); } if (!queueRemaining.tryAcquire(-remainingChange, keepAlive, TimeUnit.SECONDS)) { bytesRemaining.release(putByteCounter); throw new ChannelFullException("Space for commit to queue couldn't be acquired." + " Sinks are likely not keeping up with sources, or the buffer size is too tight"); } } int puts = putList.size(); int takes = takeList.size(); synchronized (queueLock) { if (puts > 0) { while (!putList.isEmpty()) { if (!queue.offer(putList.removeFirst())) { // offer:添加一个元素并返回true throw new RuntimeException("Queue add failed, this shouldn't be able to happen"); } } } putList.clear(); takeList.clear(); } bytesRemaining.release(takeByteCounter); takeByteCounter = 0; putByteCounter = 0; queueStored.release(puts); if (remainingChange > 0) { queueRemaining.release(remainingChange); } if (puts > 0) { channelCounter.addToEventPutSuccessCount(puts); } if (takes > 0) { channelCounter.addToEventTakeSuccessCount(takes); } channelCounter.setChannelSize(queue.size()); } // 事务失败时，将take list数据回滚到Channel Queue // 在回滚时，需要把takeList中暂存的事件回滚到Channel Queue，并回滚queueStored信号量。 @Override protected void doRollback() { int takes = takeList.size(); synchronized (queueLock) { Preconditions.checkState(queue.remainingCapacity() >= takeList.size(), "Not enough space in memory channel " + "queue to rollback takes. This should never happen, please report"); while (!takeList.isEmpty()) { queue.addFirst(takeList.removeLast()); } putList.clear(); } bytesRemaining.release(putByteCounter); putByteCounter = 0; takeByteCounter = 0; queueStored.release(takes); channelCounter.setChannelSize(queue.size()); } } private Object queueLock = new Object(); // 在操作Channel Queue时都需要锁定，因为Channel Queue可能动态扩容(会被重新new)。用法就是synchronized(queueLock){...操作queue} @GuardedBy(value = "queueLock") // 用@GuardedBy注解告诉维护者这个变量被哪个锁保护着 private LinkedBlockingDeque<Event> queue; // 由一个Channel Queue存储整个Channel的Event数据 // Semaphore可控制某资源可被同时访问的个数，acquire()获取一个许可，若无等待，而release()释放一个许可 // queueRemaining表示可存储事件容量。在提交事务时增加或减少该信号量 // 1. 首先在configure()函数中初始化为一个capacity大小的信号量 // 2. 在resize的时候，如果要缩容则要看是否还能acquire到oldCapacity - capacity个许可，不能则不允许缩容(很合理啊，不然就丢失数据了)。若是扩容，则queueRemaining.release(capacity - oldCapacity) // 3. 提交事务时，如果takeList.size() < putList.size(),则要检查是否有足够的queueRemaining private Semaphore queueRemaining; // 表示ChannelQueue已存储事件容量 // 2. 在configure()中初始化为一个大小为0的信号量 // 3. 在doTake()时tryAcquire是否有许可 // 4. 在commit()时release(puts)增加puts个许可 // 5. 在rollback()时release(takes)个许可 private Semaphore queueStored; // maximum items in a transaction queue private volatile Integer transCapacity; private volatile int keepAlive; private volatile int byteCapacity; private volatile int lastByteCapacity; private volatile int byteCapacityBufferPercentage; private Semaphore bytesRemaining; private ChannelCounter channelCounter; public MemoryChannel() { super(); } @Override public void configure(Context context) { // Read parameters from context // capacity、transactionCapacity、byteCapacity、byteCapacityBufferPercentage... } // 因为多个事务要操作ChannelQueue，还要考虑ChannelQueue的扩容问题，因此MemoryChannel使用了锁来实现；而容量问题则使用了信号量来实现。 // 改变queue的容量，是通过新建一个LinkedBlockingDeque来实现的，并将原queue的东西加进来。 private void resizeQueue(int capacity) throws InterruptedException { int oldCapacity; // 计算原queue的capacity，注意该方法需加锁 synchronized (queueLock) { oldCapacity = queue.size() + queue.remainingCapacity(); } if (oldCapacity == capacity) { return; } else if (oldCapacity > capacity) { // tryAcquire():从该信号量中获取指定数量的许可 //首先要预占老容量-新容量的大小，以便缩容容量。如果获取失败，默认是记录日志，然后忽略 if (!queueRemaining.tryAcquire(oldCapacity - capacity, keepAlive, TimeUnit.SECONDS)) { LOGGER.warn("Couldn't acquire permits to downsize the queue, resizing has been aborted"); } else { //否则，直接缩容，然后复制老Queue的数据，缩容时需要锁定queueLock，这一系列操作要线程安全 synchronized (queueLock) { LinkedBlockingDeque<Event> newQueue = new LinkedBlockingDeque<Event>(capacity); newQueue.addAll(queue); queue = newQueue; } } }
Interceptor
-
flume内部实现了很多自定义的Interceptor，如下图：
-
同时还实现了InterceptorChain用来链式处理event。
InterceptorChain
-
Implementation of Interceptor that calls a list of other Interceptors
-
Interptor接口: 用于过滤、加工Event，然后返回一个新的Event。
-
相比之下，InterceptorChain就是对event逐个(链式)调用其内的Interceptor(接口子类)实例的各个方法。
public class InterceptorChain implements Interceptor { // list of interceptors that will be traversed, in order private List<Interceptor> interceptors; public InterceptorChain() { interceptors = Lists.newLinkedList(); // 构造方法，type LinkedList } public void setInterceptors(List<Interceptor> interceptors) { this.interceptors = interceptors; // set方法 } // Interceptor接口的intercept方法: Interception of a single Event.事件拦截 // @return: Original or modified event, or null if the Event is to be dropped. @Override public Event intercept(Event event) { for (Interceptor interceptor : interceptors) { if (event == null) { return null; } event = interceptor.intercept(event); // 注意：该类的实例会调用上面的set方法初始化intercptors，其中的intercptor是Interceptor接口子类的实例。所以这里的intercept()方法调用的是Interceptor的某个接口所覆盖的方法。[Interceptor有很多子类，下面有一个demo子类的分析，可以往下看HostInterceptor] } return event; } // Interceptor接口: Interception of a batch of events // @return: Output list of events @Override public List<Event> intercept(List<Event> events) { ... // 基本同上面的方法，不过调用的是interceptor.intercept(events); } // Interceptor: Any initialization / startup needed by the Interceptor. @Override public void initialize() { Iterator<Interceptor> iter = interceptors.iterator(); while (iter.hasNext()) { Interceptor interceptor = iter.next(); interceptor.initialize(); // 挨个对linkedlist中的interceptor实例进行initialize } } @Override public void close() { ...// 挨个对linkedlist中的interceptor实例进行close }
HostInterceptor
-
implements Interceptor
-
功能：在所有拦截的events的header中上加上本机的host name或IP
public class HostInterceptor implements Interceptor { ... // 一些private变量 /** * Only {@link HostInterceptor.Builder} can build me */ // private的构造方法，so只能通过下面的静态方法Builder实例化 private HostInterceptor(boolean preserveExisting, boolean useIP, String header) { // 用xx.conf内的值初始化这些变量 this.preserveExisting = preserveExisting; this.header = header; InetAddress addr; try { addr = InetAddress.getLocalHost(); //Returns the address of the local host. if (useIP) { //Returns the IP address string in textual presentation host = addr.getHostAddress(); } else { // Gets the fully qualified domain name for this IP address. host = addr.getCanonicalHostName(); } } catch (UnknownHostException e) { logger.warn("Could not get local host address. Exception follows.", e); } } @Override public void initialize() { // no-op } /** * Modifies events in-place. */ @Override public Event intercept(Event event) { Map<String, String> headers = event.getHeaders(); // 如果要要保存当前的'host‘值并且当前已有头部，那么就不处理直接返回。 if (preserveExisting && headers.containsKey(header)) { return event; } if (host != null) { headers.put(header, host); //将host添加到头部 } return event; } @Override public List<Event> intercept(List<Event> events) { ... // 为events中的每一个event调用intercept(Event event) } @Override public void close() { // no-op } /** * Builder which builds new instances of the HostInterceptor. */ public static class Builder implements Interceptor.Builder { private boolean preserveExisting = PRESERVE_DFLT; private boolean useIP = USE_IP_DFLT; private String header = HOST; @Override public Interceptor build() { return new HostInterceptor(preserveExisting, useIP, header); } @Override public void configure(Context context) { preserveExisting = context.getBoolean(PRESERVE, PRESERVE_DFLT); useIP = context.getBoolean(USE_IP, USE_IP_DFLT); header = context.getString(HOST_HEADER, HOST); } } public static class Constants { public static String HOST = "host"; ... // 一些配置的缺省值 } } -
demo Usage in xx.conf: more details see User Guide
agent.sources.r1.interceptors = i1 agent.sources.r1.interceptors.i1.type = host # preserveExisting: 是否保存当前已存在的'host'值，缺省是不保存 agent.sources.r1.interceptors.i1.preserveExisting = true agent.sources.r1.interceptors.i1.useIP = false agent.sources.r1.interceptors.i1.hostHeader = hostname
Selector
-
先上一张所有selector的继承关系图
可见ChannelSelector默认提供了两种实现：复制和多路复用。默认实现是ReplicatingChannelSelector。
ChannelSelector
-
interface
-
基于不同实现政策，允许在channels的集合中选取channels子集。
// NamedComponent接口：用于给component附加一个名字，包括setName()和getName()方法 public interface ChannelSelector extends NamedComponent, Configurable { // @param channels：all channels the selector could select from. public void setChannels(List<Channel> channels); /** * Returns a list of required channels. 这些channels的写入失败会传达回接收事件的source. * @param: event * @return: the list of required channels that this selector has selected for * the given event. */ public List<Channel> getRequiredChannels(Event event); /** * Returns a list of optional channels. 这些channels的写入失败会被忽略。 * @param: event * @return: the list of optional channels that this selector has selected for * the given event. */ public List<Channel> getOptionalChannels(Event event); /** * @return the list of all channels that this selector is configured to work * with. */ public List<Channel> getAllChannels(); } ## AbstractChannelSelector * abstract class ```java public abstract class AbstractChannelSelector implements ChannelSelector { private List<Channel> channels; private String name; ...// override ChannelSelctor的getAllChannels()、setChannels(List<Channel> channels)、setName(String name)、getName()方法。 //@return: A map of name to channel instance. protected Map<String, Channel> getChannelNameMap() { Map<String, Channel> channelNameMap = new HashMap<String, Channel>(); for (Channel ch : getAllChannels()) { // 对每一个Channel, 将Channel和其名字放到HashMap中 channelNameMap.put(ch.getName(), ch); } return channelNameMap; } /** * Given a list of channel names as space delimited string, * returns list of channels. * @return List of {@linkplain Channel}s represented by the names. */ // 根据(space分隔的channel名字的)字符串， 返回相应的channel，利用名字-channel的HashMap protected List<Channel> getChannelListFromNames(String channels, Map<String, Channel> channelNameMap) { List<Channel> configuredChannels = new ArrayList<Channel>(); if (channels == null || channels.isEmpty()) { // 判空 return configuredChannels; } String[] chNames = channels.split(" "); for (String name : chNames) { Channel ch = channelNameMap.get(name); if (ch != null) { configuredChannels.add(ch); } else { throw new FlumeException("Selector channel not found: " + name); } } return configuredChannels; } }
ReplicatingChannelSelector
-
ChannelSelector的一个具体实现，即把接收到的消息复制到每一个Channel。【与之对应的，MultiplexingChannelSelector会根据 Event Header 中的参数进行选择，以此来选择使用哪个 Channel】
-
Replicating channel selector. 允许event被放置到source所配置的所有channels中。
-
实际的实现方式是，默认将所有channel加入requiredChannels中，optionalChannels为空。然后根据配置的"optional"将该配置对应的channel加入optionalChannels，并从requiredChannels中移除（添加和移除是在configure方法中实现的）。 TODO：看一下这个配置如何实现
public class ReplicatingChannelSelector extends AbstractChannelSelector { // Configuration to set a subset of the channels as optional. public static final String CONFIG_OPTIONAL = "optional"; List<Channel> requiredChannels = null; // 在configure()中被设置为getAllChannels()的返回值，即所有配置的channels List<Channel> optionalChannels = new ArrayList<Channel>(); @Override public List<Channel> getRequiredChannels(Event event) { /* * Seems like there are lot of components within flume that do not call * configure method. It is conceiveable that custom component tests too * do that. So in that case, revert to old behavior. */ // 如果component没有调用configure()，那么requiredChannels为null，此时再调用一次。 // TODO: configure()方法是在哪里调用的？ 同样的问题在很多class中都存在 if (requiredChannels == null) { return getAllChannels(); } return requiredChannels; } @Override public List<Channel> getOptionalChannels(Event event) { return optionalChannels; } @Override public void configure(Context context) { String optionalList = context.getString(CONFIG_OPTIONAL); requiredChannels = new ArrayList<Channel>(getAllChannels()); Map<String, Channel> channelNameMap = getChannelNameMap(); // 根据OptionList(String, 是空格分隔的channel名字),得到相应的Channel，并将channel放到optionalChannel&& 从requiredChannels中移除。 if (optionalList != null && !optionalList.isEmpty()) { for (String optional : optionalList.split("\s+")) { Channel optionalChannel = channelNameMap.get(optional); requiredChannels.remove(optionalChannel); if (!optionalChannels.contains(optionalChannel)) { optionalChannels.add(optionalChannel); } } } } }
Sink
Sink Runner
-
A driver for sinks that polls them, attempting to process events if any are available in the Channel. All sinks are polled.
public class SinkRunner implements LifecycleAware { private PollingRunner runner; // 内部类，实现了Runnable接口 private SinkProcessor policy; // }
Sink Processor
-
分为两类：
-
DefaultSinkProcessor处理单sink，直接传送不附加任何处理。
public void start() { Preconditions.checkNotNull(sink, "DefaultSinkProcessor sink not set"); sink.start(); // start()方法直接启动single sink lifecycleState = LifecycleState.START; } // stop()方法类似，configure()方法为空 public Status process() throws EventDeliveryException { return sink.process(); // 直接调用sink的process() } public void setSinks(List<Sink> sinks) { Preconditions.checkNotNull(sinks); Preconditions.checkArgument(sinks.size() == 1, "DefaultSinkPolicy can " + "only handle one sink, " + "try using a policy that supports multiple sinks"); sink = sinks.get(0); } -
多sink处理(AbstractSinkProcessor)，其中又包括两种：
-
FailoverSinkProcessor：故障切换—>通过维持一个sinks的优先级list —> 把故障sinks降级放到一个pool中被赋予一个冷冻周期。必须先调用setSinks()再configure()
public void setSinks(List<Sink> sinks) { // needed to implement the start/stop functionality super.setSinks(sinks); this.sinks = new HashMap<String, Sink>(); for (Sink sink : sinks) { this.sinks.put(sink.getName(), sink); } } private Sink moveActiveToDeadAndGetNext() { Integer key = liveSinks.lastKey(); failedSinks.add(new FailedSink(key, activeSink, 1)); // 把当前liveSinks里的第一优先级key移除到failedSinks中 liveSinks.remove(key); if (liveSinks.isEmpty()) return null; if (liveSinks.lastKey() != null) { return liveSinks.get(liveSinks.lastKey()); } else { return null; } } ... -
LoadBalancingSinkProcessor: 提供在多个sinks之间负载均衡的能力—> 维持一个active sinks的索引序列(load需分布在这些sinks上) —> 算法包括ROUND_ROBIN(default)和RANDOM选择机制。
内部通过一个
private interface SinkSelector实现。该接口下实现了两个私有静态类RoundRobinSinkSelector和RandomOrderSinkSelector.
-