29 友盟大数据--flume源码查看分析ExecSource--UmengExecSource 改造exec源 ：监控目录、收集新文件---增加个守护线程不断监控目录

flume自定义源防丢失---解决flume还未收集完日志便已经滚动，数据丢失问题

防重、防丢失

改造exec源 ：监控目录、收集新文件---增加个守护线程不断监控目录--- 一次收集完一个文件

redis 3号库维护key  --- 防止重复

UmengExecSource

package com.oldboy.umeng.flume;

import java.io.*;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import org.apache.flume.Channel;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.Source;
import org.apache.flume.SystemClock;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.instrumentation.SourceCounter;
import org.apache.flume.source.AbstractSource;
import org.apache.flume.source.ExecSourceConfigurationConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import redis.clients.jedis.Jedis;

import java.nio.charset.Charset;

public class UmengExecSource extends AbstractSource implements EventDrivenSource, Configurable {

    private static final Logger logger = LoggerFactory.getLogger(org.apache.flume.source.ExecSource.class);

    private String shell;
    private String command;
    private SourceCounter sourceCounter;
    private ExecutorService executor;
    private Future<?> runnerFuture;
    private long restartThrottle;
    private boolean restart;
    private boolean logStderr;
    private Integer bufferCount;
    private long batchTimeout;
    private UmengExecSource.ExecRunnable runner;
    private Charset charset;
    private String spooldir ;
    private String prefix ;
    private String suffix ;
    private String redisHost ;
    private int redisPort ;

    public void start() {
        logger.info("Exec source starting with command:{}", command);

        executor = Executors.newSingleThreadExecutor();

        runner = new UmengExecSource.ExecRunnable(shell, command, getChannelProcessor(),
                                                                            sourceCounter, restart, restartThrottle,
                                                                            logStderr, bufferCount, batchTimeout,
                                                                            charset);

        runnerFuture = executor.submit(runner);
        sourceCounter.start();

        new SpoolDirThread(spooldir, prefix, suffix, redisHost, redisPort, this).start();
        //启动守护线程
        super.start();
        logger.debug("Exec source started");
    }

    @Override
    public void stop() {
        logger.info("Stopping exec source with command:{}", command);
        if (runner != null) {
            runner.setRestart(false);
            runner.kill();
        }

        if (runnerFuture != null) {
            logger.debug("Stopping exec runner");
            runnerFuture.cancel(true);
            logger.debug("Exec runner stopped");
        }
        executor.shutdown();

        while (!executor.isTerminated()) {
            logger.debug("Waiting for exec executor service to stop");
            try {
                executor.awaitTermination(500, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
                logger.debug("Interrupted while waiting for exec executor service " + "to stop. Just exiting.");
                Thread.currentThread().interrupt();
            }
        }

        sourceCounter.stop();
        super.stop();

        logger.debug("Exec source with command:{} stopped. Metrics:{}", command, sourceCounter);
    }

    /**
     * 获得配置属性
     */
    public void configure(Context context) {
        command = context.getString("command");

        Preconditions.checkState(command != null, "The parameter command must be specified");

        restartThrottle = context.getLong(ExecSourceConfigurationConstants.CONFIG_RESTART_THROTTLE,
                ExecSourceConfigurationConstants.DEFAULT_RESTART_THROTTLE);

        restart = context.getBoolean(ExecSourceConfigurationConstants.CONFIG_RESTART,
                ExecSourceConfigurationConstants.DEFAULT_RESTART);

        logStderr = context.getBoolean(ExecSourceConfigurationConstants.CONFIG_LOG_STDERR,
                ExecSourceConfigurationConstants.DEFAULT_LOG_STDERR);

        bufferCount = context.getInteger(ExecSourceConfigurationConstants.CONFIG_BATCH_SIZE,
                ExecSourceConfigurationConstants.DEFAULT_BATCH_SIZE);

        batchTimeout = context.getLong(ExecSourceConfigurationConstants.CONFIG_BATCH_TIME_OUT,
                ExecSourceConfigurationConstants.DEFAULT_BATCH_TIME_OUT);

        charset = Charset.forName(context.getString(ExecSourceConfigurationConstants.CHARSET,
                ExecSourceConfigurationConstants.DEFAULT_CHARSET));

        shell = context.getString(ExecSourceConfigurationConstants.CONFIG_SHELL, null);

134         spooldir = context.getString("spooldir") ;
135         prefix = context.getString("prefix" , "access.log.") ;//判断前缀
136         suffix = context.getString("suffix" , "COMPLETED") ;//加后缀
137         redisHost = context.getString("redisHost" , "s101") ;//配置redis
138         redisPort = context.getInteger("redisPort" , 6379) ;
        if(spooldir == null){
            System.out.println("spooldir没有配置!");
            System.exit(-1);
        }

        if (sourceCounter == null) {
            sourceCounter = new SourceCounter(getName());
146         }
    }

    private static class ExecRunnable implements Runnable {

        public ExecRunnable(String shell, String command, ChannelProcessor channelProcessor, SourceCounter sourceCounter, boolean restart, long restartThrottle, boolean logStderr, int bufferCount, long batchTimeout, Charset charset) {
            this.command = command;
            this.channelProcessor = channelProcessor;
            this.sourceCounter = sourceCounter;
            this.restartThrottle = restartThrottle;
            this.bufferCount = bufferCount;
            this.batchTimeout = batchTimeout;
            this.restart = restart;
            this.logStderr = logStderr;
            this.charset = charset;
            this.shell = shell;
        }

        private final String shell;
        private final String command;
        private final ChannelProcessor channelProcessor;
        private final SourceCounter sourceCounter;
        private volatile boolean restart;
        private final long restartThrottle;
        private final int bufferCount;
        private long batchTimeout;
        private final boolean logStderr;
        private final Charset charset;
        private Process process = null;
        private SystemClock systemClock = new SystemClock();
        private Long lastPushToChannel = systemClock.currentTimeMillis();
        ScheduledExecutorService timedFlushService;
        ScheduledFuture<?> future;

        public void run() {
            do {
                String exitCode = "unknown";
                BufferedReader reader = null;
                String line = null;
                final List<Event> eventList = new ArrayList<Event>();

                timedFlushService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryBuilder().setNameFormat(
                        "timedFlushExecService" + Thread.currentThread().getId() + "-%d").build());
                try {
                    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));

                    // StderrLogger dies as soon as the input stream is invalid
                    UmengExecSource.StderrReader stderrReader = new UmengExecSource.StderrReader(new BufferedReader(new InputStreamReader(process.getErrorStream(),
                                                                                                                                                                                       charset)),
                                                                                                                                              logStderr);
                    stderrReader.setName("StderrReader-[" + command + "]");
                    stderrReader.setDaemon(true);
                    stderrReader.start();

                    future = timedFlushService.scheduleWithFixedDelay(new Runnable() {
                        public void run() {
                            try {
                                synchronized (eventList) {
                                    if (!eventList.isEmpty() && timeout()) {
                                        flushEventBatch(eventList);
                                    }
                                }
                            } catch (Exception e) {
                                logger.error("Exception occured when processing event batch", e);
                                if (e instanceof InterruptedException) {
                                    Thread.currentThread().interrupt();
                                }
                            }
                        }
                    }, batchTimeout, batchTimeout, TimeUnit.MILLISECONDS);

                    while ((line = reader.readLine()) != null) {
                        synchronized (eventList) {
                            sourceCounter.incrementEventReceivedCount();
                            eventList.add(EventBuilder.withBody(line.getBytes(charset)));
                            if (eventList.size() >= bufferCount || timeout()) {
                                flushEventBatch(eventList);
                            }
                        }
                    }

                    synchronized (eventList) {
                        if (!eventList.isEmpty()) {
                            flushEventBatch(eventList);
                        }
                    }
                } catch (Exception e) {
                    logger.error("Failed while running command: " + command, e);
                    if (e instanceof InterruptedException) {
                        Thread.currentThread().interrupt();
                    }
                } finally {
                    if (reader != null) {
                        try {
                            reader.close();
                        } catch (IOException ex) {
                            logger.error("Failed to close reader for exec source", ex);
                        }
                    }
                    exitCode = String.valueOf(kill());
                }
                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);
        }

        private void flushEventBatch(List<Event> eventList) {
            channelProcessor.processEventBatch(eventList);
            sourceCounter.addToEventAcceptedCount(eventList.size());
            eventList.clear();
            lastPushToChannel = systemClock.currentTimeMillis();
        }

        private boolean timeout() {
            return (systemClock.currentTimeMillis() - lastPushToChannel) >= batchTimeout;
        }

        private static String[] formulateShellCommand(String shell, String command) {
            String[] shellArgs = shell.split("\s+");
            String[] result = new String[shellArgs.length + 1];
            System.arraycopy(shellArgs, 0, result, 0, shellArgs.length);
            result[shellArgs.length] = command;
            return result;
        }

        public int kill() {
            if (process != null) {
                synchronized (process) {
                    process.destroy();

                    try {
                        int exitValue = process.waitFor();

                        // Stop the Thread that flushes periodically
                        if (future != null) {
                            future.cancel(true);
                        }

                        if (timedFlushService != null) {
                            timedFlushService.shutdown();
                            while (!timedFlushService.isTerminated()) {
                                try {
                                    timedFlushService.awaitTermination(500, TimeUnit.MILLISECONDS);
                                } catch (InterruptedException e) {
                                    logger.debug("Interrupted while waiting for exec executor service " + "to stop. Just exiting.");
                                    Thread.currentThread().interrupt();
                                }
                            }
                        }
                        return exitValue;
                    } catch (InterruptedException ex) {
                        Thread.currentThread().interrupt();
                    }
                }
                return Integer.MIN_VALUE;
            }
            return Integer.MIN_VALUE / 2;
        }

        public void setRestart(boolean restart) {
            this.restart = restart;
        }
    }

    private static class StderrReader extends Thread {
        private BufferedReader input;
        private boolean logStderr;

        protected StderrReader(BufferedReader input, boolean logStderr) {
            this.input = input;
            this.logStderr = logStderr;
        }

        public void run() {
            try {
                int i = 0;
                String line = null;
                while ((line = input.readLine()) != null) {
                    if (logStderr) {
                        // There is no need to read 'line' with a charset
                        // as we do not to propagate it.
                        // It is in UTF-16 and would be printed in UTF-8 format.
                        logger.info("StderrLogger[{}] = '{}'", ++i, line);
                    }
                }
            } catch (IOException e) {
                logger.info("StderrLogger exiting", e);
            } finally {
                try {
                    if (input != null) {
                        input.close();
                    }
                } catch (IOException ex) {
                    logger.error("Failed to close stderr reader for exec source", ex);
                }
            }
        }
    }

360     public static class SpoolDirThread extends Thread{
        private String spoolDir ;
        private String prefix ;
        private String suffix ;
        private Jedis redis ;
        private UmengExecSource source ;

        public SpoolDirThread(String spooldir, String prefix , String suffix ,String host , int port , UmengExecSource source){
            //守护线程
            this.setDaemon(true);
            this.spoolDir = spooldir ;
            this.prefix = prefix ;
            this.suffix = suffix ;
            this.source = source ;
            redis = new Jedis(host , port ) ;//配置redis
            redis.select(3) ;

        }
        public void run() {
            while(true){
                File dir = new File(spoolDir) ;
                if(dir.isDirectory()){
                    File[] files = dir.listFiles() ;
                    for(File f :files){
                        String fname = f.getName() ;
                        //处理滚动的文件 有前缀 没后缀  进行处理
                        if(fname.startsWith(prefix) && !fname.endsWith(suffix)){
                            doProcessLog(f) ;
                        }
                    }
                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }

        /**
         * 处理一个滚动的日志文件
         */
        private void doProcessLog(File f) {
            try {
                BufferedReader br =new BufferedReader(new InputStreamReader(new FileInputStream(f))) ;
                String line = null ;
                while((line = br.readLine()) != null){
                    String key = line.substring(0,line.lastIndexOf("#")) ;
                    //已经收集过了
                    System.out.println(" : ke防丢失源y = " + key);
                    if(!redis.exists(key)){
                        System.out.println(key + " : 未处理！");
                        //交给通道处理器
                        source.getChannelProcessor().processEvent(EventBuilder.withBody(line.getBytes()));
                    }
                }
                //重命名文件  处理完加后缀COMPLETED
                f.renameTo(new File(f.getParentFile() , f.getName() + "." + suffix)) ;
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
422     }
}