Flume之ChannelProcessor源码分析

甩祸

接着上篇：Flume之ChannelSelector源码分析

ChannelSelector主要是根据Event选择将其发送到哪些Channel。ChannelProcessor是通过ChannelSelector获取到Channels后，如何发送Event到Channel。

　　ChannelProcessor提供了将Source接收到的Events放入到Channels的一些方法，如果写入Channels发生错误，统一抛出ChannelException异常。每个ChannelProcessor实例都有一个ChannelSelector实例（ChannelSelector实例分别维护了可选的Channels列表和必选的Channels列表）和一个拦截链。
　　

　　一，configure(Context context)加载配置文件，该方法主要是用来加载拦截器：

  private void configureInterceptors(Context context) {
List<Interceptor> interceptors = Lists.newLinkedList();
//获取拦截器
String interceptorListStr = context.getString("interceptors", "");
if (interceptorListStr.isEmpty()) {
return;
}
//解析成拦截器名的数组
String[] interceptorNames = interceptorListStr.split("\\s+");
//获取interceptors的Context
Context interceptorContexts =
new Context(context.getSubProperties("interceptors."));
// run through and instantiate all the interceptors specified in the Context
InterceptorBuilderFactory factory = new InterceptorBuilderFactory();
for (String interceptorName : interceptorNames) {
Context interceptorContext = new Context(
interceptorContexts.getSubProperties(interceptorName + "."));
//得到拦截器的类型，Flume支持TIMESTAMP， HOST， STATIC， REGEX_FILTER， REGEX_EXTRACTOR， SEARCH_REPLACE
//定义在org.apache.flume.interceptor.InterceptorType类中。
String type = interceptorContext.getString("type");
if (type == null) {
LOG.error("Type not specified for interceptor " + interceptorName);
throw new FlumeException("Interceptor.Type not specified for " +
interceptorName);
}
try {
//实例化拦截器，并存放到List中
Interceptor.Builder builder = factory.newInstance(type);
builder.configure(interceptorContext);
interceptors.add(builder.build());
} catch (ClassNotFoundException e) {
LOG.error("Builder class not found. Exception follows.", e);
throw new FlumeException("Interceptor.Builder not found.", e);
} catch (InstantiationException e) {
LOG.error("Could not instantiate Builder. Exception follows.", e);
throw new FlumeException("Interceptor.Builder not constructable.", e);
} catch (IllegalAccessException e) {
LOG.error("Unable to access Builder. Exception follows.", e);
throw new FlumeException("Unable to access Interceptor.Builder.", e);
}
}
//将拦截器List设置到拦截链中。
interceptorChain.setInterceptors(interceptors);
}二，configure方法已经初始化拦截链，接下来org.apache.flume.source.EventDrivenSourceRunner的start方法调用initialize()初始化拦截链并启动Source：
  @Override
public void start() {
Source source = getSource();
ChannelProcessor cp = source.getChannelProcessor();
//初始化ChannelProcessor，实际上就是初始化拦截链的所有的拦截器
cp.initialize();
//启动Source
source.start();
lifecycleState = LifecycleState.START;
}


三，提交Event　　ChannelProcessor提交Event有两个方法processEvent（提交单个Event）和processEventBatch（批量提交Event），下面以processEventBatch为例：


  public void processEventBatch(List<Event> events) {
Preconditions.checkNotNull(events, "Event list must not be null");
//根据拦截链挨个处理Event
events = interceptorChain.intercept(events);
Map<Channel, List<Event>> reqChannelQueue =
new LinkedHashMap<Channel, List<Event>>();
Map<Channel, List<Event>> optChannelQueue =
new LinkedHashMap<Channel, List<Event>>();
//将Event分类，放入reqChannelQueue 和 optChannelQueue中
for (Event event : events) {
List<Channel> reqChannels = selector.getRequiredChannels(event);
for (Channel ch : reqChannels) {
List<Event> eventQueue = reqChannelQueue.get(ch);
if (eventQueue == null) {
eventQueue = new ArrayList<Event>();
reqChannelQueue.put(ch, eventQueue);
}
eventQueue.add(event);
}
List<Channel> optChannels = selector.getOptionalChannels(event);
for (Channel ch: optChannels) {
List<Event> eventQueue = optChannelQueue.get(ch);
if (eventQueue == null) {
eventQueue = new ArrayList<Event>();
optChannelQueue.put(ch, eventQueue);
}
eventQueue.add(event);
}
}
//提交Event到必选的Channel
// Process required channels
for (Channel reqChannel : reqChannelQueue.keySet()) {
Transaction tx = reqChannel.getTransaction();
Preconditions.checkNotNull(tx, "Transaction object must not be null");
try {
tx.begin();
List<Event> batch = reqChannelQueue.get(reqChannel);
for (Event event : batch) {
reqChannel.put(event);
}
tx.commit();
} catch (Throwable t) {
tx.rollback();
if (t instanceof Error) {
LOG.error("Error while writing to required channel: " +
reqChannel, t);
throw (Error) t;
} else {
throw new ChannelException("Unable to put batch on required " +
"channel: " + reqChannel, t);
}
} finally {
if (tx != null) {
tx.close();
}
}
}
//提交Event到可选的Channel
// Process optional channels
for (Channel optChannel : optChannelQueue.keySet()) {
Transaction tx = optChannel.getTransaction();
Preconditions.checkNotNull(tx, "Transaction object must not be null");
try {
tx.begin();
List<Event> batch = optChannelQueue.get(optChannel);
for (Event event : batch ) {
optChannel.put(event);
}
tx.commit();
} catch (Throwable t) {
tx.rollback();
LOG.error("Unable to put batch on optional channel: " + optChannel, t);
if (t instanceof Error) {
throw (Error) t;
}
} finally {
if (tx != null) {
tx.close();
}
}
}
}以上批量批量提交Event的流程：　　1，根据配置的拦截器组成的拦截链挨个处理Event
　　2，根据Event获取必选和可选的Channel，然后分别根据可选和必选的Channel，将Event和Channel存放在以Channel为Key，Event列表为Value的Map中。
　　3，遍历可选和必选的Map，通过Channel的Transaction将Event commit到Channel中。
　　看下可选和必选map提交Channel异常处理有什么区别：
　　（1）必选Map的异常处理

        if (t instanceof Error) {
LOG.error("Error while writing to required channel: " +
reqChannel, t);
throw (Error) t;
} else {
//抛出ChannelException异常，Source可以通过该异常保证数据不丢失的
throw new ChannelException("Unable to put batch on required " +
"channel: " + reqChannel, t);
}（2）可选Map的异常处理
        tx.rollback();
LOG.error("Unable to put batch on optional channel: " + optChannel, t);
if (t instanceof Error) {
//只有在异常是Error才会抛出异常
throw (Error) t;
}也就是说，只有异常为java.lang.Error类实例，可选和必选Channel才会回滚，不然可选提交到Channel后失败是不会回滚的。