以下是代码实现，仅做参考，生产上直接用TAILDIR读取文件内容即可，若要读取一个目录下的子目录，可使用github上以实现的这个项目包：https://github.com/qwurey/flume-source-taildir-recursive
package com.fwmagic.flume.source;import org.apache.commons.io.FileUtils;import org.apache.commons.lang.StringUtils;import org.apache.flume.Context;import org.apache.flume.Event;import org.apache.flume.EventDrivenSource;import org.apache.flume.channel.ChannelProcessor;import org.apache.flume.conf.Configurable;import org.apache.flume.event.EventBuilder;import org.apache.flume.source.AbstractSource;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import java.io.File;import java.io.IOException;import java.io.RandomAccessFile;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.TimeUnit;/** * @Description:自定义Source 1、读取指定目录下的文件，如nginx的access.log * 2、读取文件前先判断offset文件是否存在，不存在则创建它 * 3、每次读取完都写一个offset文件记录读取到文件的什么位置，防止重启flume时发生重复消费的情况 * 4、如何自定义？参考ExecSource * <p> * (1):获取自定义配置文件属性 * (2):创建线程池，用channelProcessor发送数据给channel * (3):线程池提交(启动任务) * 任务内容： * (1):读取偏移量文件，没有则创建，有则获取偏移量，将读取的指针重置到指定偏移量 * (2):读取指定的日志文件，将读取的一行内容打包成Event,用Channel发送Event * (3):获取读取内容后的偏移量，重置偏移量 * (4):stop方法调用，关闭线程池，调用super.stop方法。 * @Date:Create in 2018/8/19 */public class TailFileSource extends AbstractSource implements EventDrivenSource, Configurable { ???/*监听的文件*/ ???private String filePath; ???/*记录读取偏移量的文件*/ ???private String posiFile; ???/*若读取文件暂无内容，则等待数秒*/ ???private Long interval; ???/*读写文件的字符集*/ ???private String charset; ???/*读取文件内容的线程*/ ???private FileRunner fileRunner; ???/*线程池*/ ???private ExecutorService executor; ???private static final Logger logger = LoggerFactory.getLogger(TailFileSource.class); ???/** ????* 初始化配置文件内容 ????* ????* @param context ????*/ ???@Override ???public void configure(Context context) { ???????filePath = context.getString("filePath"); ???????posiFile = context.getString("posiFile"); ???????interval = context.getLong("interval", 2000L); ???????charset = context.getString("charset", "UTF-8"); ???} ???@Override ???public synchronized void start() { ???????//启动一个线程，用于监听对应的日志文件 ???????//创建一个线程池 ???????executor = Executors.newSingleThreadExecutor(); ???????//用channelProcessor发送数据给channel ???????ChannelProcessor channelProcessor = super.getChannelProcessor(); ???????fileRunner = new FileRunner(filePath, posiFile, interval, charset, channelProcessor); ???????executor.submit(fileRunner); ???????super.start(); ???} ???@Override ???public synchronized void stop() { ???????fileRunner.setFlag(Boolean.FALSE); ???????while (!executor.isTerminated()) { ???????????logger.debug("waiting for exec executor service to stop"); ???????????try { ???????????????executor.awaitTermination(500, TimeUnit.MILLISECONDS); ???????????} catch (InterruptedException e) { ???????????????e.printStackTrace(); ???????????????logger.debug("Interrupted while waiting for executor service to stop,Just exiting."); ???????????????Thread.currentThread().interrupt(); ???????????} ???????} ???????super.stop(); ???} ???public static class FileRunner implements Runnable { ???????private Long interval; ???????private String charset; ???????private Long offset = 0L; ???????private File pFile; ???????private RandomAccessFile raf; ???????private ChannelProcessor channelProcessor; ???????private Boolean flag = Boolean.TRUE; ???????public void setFlag(Boolean flag) { ???????????this.flag = flag; ???????} ???????public FileRunner(String filePath, String posiFile, Long interval, String charset, ChannelProcessor channelProcessor) { ???????????this.interval = interval; ???????????this.charset = charset; ???????????this.channelProcessor = channelProcessor; ???????????//1、判断是否有偏移量文件，有则读取偏移量,没有则创建 ???????????pFile = new File(posiFile); ???????????if (!pFile.exists()) { ???????????????try { ???????????????????pFile.createNewFile(); ???????????????} catch (IOException e) { ???????????????????e.printStackTrace(); ???????????????????logger.error("create position file error!", e); ???????????????} ???????????} ???????????//2、判断偏移量中的文件内容是否大于0 ???????????try { ???????????????String offsetStr = FileUtils.readFileToString(pFile, this.charset);// ?????????3、如果偏移量文件中有记录，则将内容转换为Long ???????????????if (StringUtils.isNotBlank(offsetStr)) { ???????????????????offset = Long.parseLong(offsetStr); ???????????????}// ??????????4、如果有偏移量，则直接跳到文件的偏移量位置 ???????????????raf = new RandomAccessFile(filePath, "r");// ?????????????跳到指定的位置 ???????????????raf.seek(offset); ???????????} catch (IOException e) { ???????????????e.printStackTrace(); ???????????????logger.error("read position file error!", e); ???????????} ???????} ???????@Override ???????public void run() { ???????????//监听文件 ???????????while (flag) {// ???????????读取文件中的内容 ???????????????String line = null; ???????????????try { ???????????????????line = raf.readLine(); ???????????????????if (StringUtils.isNotBlank(line)) {// ?????????????????????把数据打包成Event,发送到Channel ???????????????????????line = new String(line.getBytes("ISO-8859-1"), "UTF-8"); ???????????????????????Event event = EventBuilder.withBody(line.getBytes()); ???????????????????????channelProcessor.processEvent(event); ???????????????????????//更新偏移量文件，把偏移量写入文件 ???????????????????????offset = raf.getFilePointer(); ???????????????????????FileUtils.writeStringToFile(pFile, offset.toString()); ???????????????????} else { ???????????????????????try { ???????????????????????????Thread.sleep(interval); ???????????????????????} catch (InterruptedException e) { ???????????????????????????e.printStackTrace(); ???????????????????????????logger.error("thread sleep error", e); ???????????????????????} ???????????????????} ???????????????} catch (IOException e) { ???????????????????e.printStackTrace(); ???????????????} ???????????} ???????} ???}}Flume自定义Source
原文地址：http://blog.51cto.com/simplelife/2307943