标签：execsource flume
我们直接看该Source的start方法吧
public void start() {
logger.info("Exec source starting with command:{}", command);
executor = Executors.newSingleThreadExecutor();
runner = new ExecRunnable(shell, command, getChannelProcessor(), sourceCounter,
restart, restartThrottle, logStderr, bufferCount, batchTimeout, charset);
// FIXME: Use a callback-like executor / future to signal us upon failure.
runnerFuture = executor.submit(runner);
/*
* NB: This comes at the end rather than the beginning of the method because
* it sets our state to running. We want to make sure the executor is alive
* and well first.
*/
sourceCounter.start();
super.start();
logger.debug("Exec source started");
}启动了一个线程来运行，运行的详细过程看runner
它是一个线程，实现了Runnable接口，所以直接看它重写的run方法的逻辑，我们一块一块来看：
if(shell != null) {
String[] commandArgs = formulateShellCommand(shell, command);
process = Runtime.getRuntime().exec(commandArgs);
} else {
String[] commandArgs = command.split("\\s+");
process = new ProcessBuilder(commandArgs).start();
}
reader = new BufferedReader(
new InputStreamReader(process.getInputStream(), charset));这里就是执行shell命令，并且将shell命令的输出结果作为输入流读到reader中,InputStreamReader是字节流通向字符流的桥梁，它使用指定的charset读取字节并将其解码为字符，每次调用read方法都会从底层输入流读取一个或多个字节。
while ((line = reader.readLine()) != null) {
synchronized (eventList) {
sourceCounter.incrementEventReceivedCount();
eventList.add(EventBuilder.withBody(line.getBytes(charset)));
if(eventList.size() >= bufferCount || timeout()) {
flushEventBatch(eventList);
}
}
}如果读入的内容非空，先同步eventList，如果eventList超出一定范围未做处理就会flush
private void flushEventBatch(List<Event> eventList){
channelProcessor.processEventBatch(eventList);
sourceCounter.addToEventAcceptedCount(eventList.size());
eventList.clear();
lastPushToChannel = systemClock.currentTimeMillis();
}flush就是现将积攒下来的eventList中的event都处理掉，然后清空
1、将event都放入配置的通道中
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放到通道中的详细过程了，但是这里大家注意到有两次selector的getchannel的方法，这是因为通道的选择器模式有两种：复用和复制 if(restart) {
logger.info("Restarting in {}ms, exit code {}", restartThrottle,
exitCode);
try {
Thread.sleep(restartThrottle);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
} else {
logger.info("Command [" + command + "] exited with " + exitCode);
}
} while(restart);restart参数的含义是，当shell命令执行的时候进程死了，是否重启该命令的进程，默认是false
配置为true的话，就会将刚才的所有代码循环一遍
总结：
1、event如何产出的？
eventList.add(EventBuilder.withBody(line.getBytes(charset)));
public static Event withBody(byte[] body, Map<String, String> headers) {
Event event = new SimpleEvent();
if(body == null) {
body = new byte[0];
}
event.setBody(body);
if (headers != null) {
event.setHeaders(new HashMap<String, String>(headers));
}
return event;
}2、event如何放入通道？
private void flushEventBatch(List<Event> eventList){
channelProcessor.processEventBatch(eventList);
sourceCounter.addToEventAcceptedCount(eventList.size());
eventList.clear();
lastPushToChannel = systemClock.currentTimeMillis();
}【Flume】 flume中ExecSource源码的详细分析——执行终端命令获取数据
标签：execsource flume
原文地址：http://blog.csdn.net/simonchi/article/details/43523569
