28 友盟大数据--flume源码查看分析- ExecSource-参照主机名拦截器HostInterceptor ---写限速拦截器

//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by Fernflower decompiler)
//

package org.apache.flume.source;

import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.charset.Charset;
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.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDrivenSource;
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.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ExecSource extends AbstractSource implements EventDrivenSource, Configurable {
    private static final Logger logger = LoggerFactory.getLogger(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 ExecSource.ExecRunnable runner;
    private Charset charset;

    public ExecSource() {
    }

    public void start() {
        logger.info("Exec source starting with command:{}", this.command);
        this.executor = Executors.newSingleThreadExecutor();
        this.runner = new ExecSource.ExecRunnable(this.shell, this.command, this.getChannelProcessor(), this.sourceCounter, this.restart, this.restartThrottle, this.logStderr, this.bufferCount, this.batchTimeout, this.charset);
        this.runnerFuture = this.executor.submit(this.runner);
        this.sourceCounter.start();
        super.start();
        logger.debug("Exec source started");
    }

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

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

        this.executor.shutdown();

        while(!this.executor.isTerminated()) {
            logger.debug("Waiting for exec executor service to stop");

            try {
                this.executor.awaitTermination(500L, TimeUnit.MILLISECONDS);
            } catch (InterruptedException var2) {
                logger.debug("Interrupted while waiting for exec executor service to stop. Just exiting.");
                Thread.currentThread().interrupt();
            }
        }

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

    public void configure(Context context) {
        this.command = context.getString("command");
        Preconditions.checkState(this.command != null, "The parameter command must be specified");
        this.restartThrottle = context.getLong("restartThrottle", 10000L);
        this.restart = context.getBoolean("restart", false);
        this.logStderr = context.getBoolean("logStdErr", false);
        this.bufferCount = context.getInteger("batchSize", 20);
        this.batchTimeout = context.getLong("batchTimeout", 3000L);
        this.charset = Charset.forName(context.getString("charset", "UTF-8"));
        this.shell = context.getString("shell", (String)null);
        if (this.sourceCounter == null) {
            this.sourceCounter = new SourceCounter(this.getName());
        }

    }

    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 = this.input.readLine()) != null) {
                    if (this.logStderr) {
                        Logger var10000 = ExecSource.logger;
                        ++i;
                        var10000.info("StderrLogger[{}] = '{}'", i, line);
                    }
                }
            } catch (IOException var11) {
                ExecSource.logger.info("StderrLogger exiting", var11);
            } finally {
                try {
                    if (this.input != null) {
                        this.input.close();
                    }
                } catch (IOException var10) {
                    ExecSource.logger.error("Failed to close stderr reader for exec source", var10);
                }

            }

        }
    }

    private static class ExecRunnable implements Runnable {
        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;
        ScheduledExecutorService timedFlushService;
        ScheduledFuture<?> future;

        public ExecRunnable(String shell, String command, ChannelProcessor channelProcessor, SourceCounter sourceCounter, boolean restart, long restartThrottle, boolean logStderr, int bufferCount, long batchTimeout, Charset charset) {
            this.lastPushToChannel = this.systemClock.currentTimeMillis();
            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;
        }

        public void run() {
            do {
                String exitCode = "unknown";
                BufferedReader reader = null;
                String line = null;
                final List<Event> eventList = new ArrayList();
                this.timedFlushService = Executors.newSingleThreadScheduledExecutor((new ThreadFactoryBuilder()).setNameFormat("timedFlushExecService" + Thread.currentThread().getId() + "-%d").build());

                try {
                    String[] commandArgs;
                    if (this.shell != null) {
                        commandArgs = formulateShellCommand(this.shell, this.command);
                        this.process = Runtime.getRuntime().exec(commandArgs);
                    } else {
                        commandArgs = this.command.split("\s+");
                        this.process = (new ProcessBuilder(commandArgs)).start();
                    }

                    reader = new BufferedReader(new InputStreamReader(this.process.getInputStream(), this.charset));
                    ExecSource.StderrReader stderrReader = new ExecSource.StderrReader(new BufferedReader(new InputStreamReader(this.process.getErrorStream(), this.charset)), this.logStderr);
                    stderrReader.setName("StderrReader-[" + this.command + "]");
                    stderrReader.setDaemon(true);
                    stderrReader.start();
                    this.future = this.timedFlushService.scheduleWithFixedDelay(new Runnable() {
                        public void run() {
                            try {
                                List var1 = eventList;
                                synchronized(eventList) {
                                    if (!eventList.isEmpty() && ExecRunnable.this.timeout()) {
                                        ExecRunnable.this.flushEventBatch(eventList);
                                    }
                                }
                            } catch (Exception var4) {
                                ExecSource.logger.error("Exception occured when processing event batch", var4);
                                if (var4 instanceof InterruptedException) {
                                    Thread.currentThread().interrupt();
                                }
                            }

                        }
                    }, this.batchTimeout, this.batchTimeout, TimeUnit.MILLISECONDS);

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

                    synchronized(eventList) {
                        if (!eventList.isEmpty()) {
                            this.flushEventBatch(eventList);
                        }
                    }
                } catch (Exception var23) {
                    ExecSource.logger.error("Failed while running command: " + this.command, var23);
                    if (var23 instanceof InterruptedException) {
                        Thread.currentThread().interrupt();
                    }
                } finally {
                    if (reader != null) {
                        try {
                            reader.close();
                        } catch (IOException var19) {
                            ExecSource.logger.error("Failed to close reader for exec source", var19);
                        }
                    }

                    exitCode = String.valueOf(this.kill());
                }

                if (this.restart) {
                    ExecSource.logger.info("Restarting in {}ms, exit code {}", this.restartThrottle, exitCode);

                    try {
                        Thread.sleep(this.restartThrottle);
                    } catch (InterruptedException var20) {
                        Thread.currentThread().interrupt();
                    }
                } else {
                    ExecSource.logger.info("Command [" + this.command + "] exited with " + exitCode);
                }
            } while(this.restart);

        }

        private void flushEventBatch(List<Event> eventList) {
            this.channelProcessor.processEventBatch(eventList);//通道处理器  详细见下面代码
            this.sourceCounter.addToEventAcceptedCount((long)eventList.size());
            eventList.clear();
            this.lastPushToChannel = this.systemClock.currentTimeMillis();
        }

        private boolean timeout() {
            return this.systemClock.currentTimeMillis() - this.lastPushToChannel >= this.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 (this.process != null) {
                Process var1 = this.process;
                synchronized(this.process) {
                    this.process.destroy();

                    int var10000;
                    try {
                        int exitValue = this.process.waitFor();
                        if (this.future != null) {
                            this.future.cancel(true);
                        }

                        if (this.timedFlushService != null) {
                            this.timedFlushService.shutdown();

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

                        var10000 = exitValue;
                    } catch (InterruptedException var6) {
                        Thread.currentThread().interrupt();
                        return -2147483648;
                    }

                    return var10000;
                }
            } else {
                return -1073741824;
            }
        }

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

ChannelProcessor  processEventBatch()

 public void processEventBatch(List<Event> events) {
        Preconditions.checkNotNull(events, "Event list must not be null");
        events = this.interceptorChain.intercept(events);//拦截器链---拦截事件
        Map<Channel, List<Event>> reqChannelQueue = new LinkedHashMap();
        Map<Channel, List<Event>> optChannelQueue = new LinkedHashMap();
        Iterator i$ = events.iterator();

        List batch;
        Iterator i$;
        while(i$.hasNext()) {
            Event event = (Event)i$.next();
            List<Channel> reqChannels = this.selector.getRequiredChannels(event);

            Object eventQueue;
            for(Iterator i$ = reqChannels.iterator(); i$.hasNext(); ((List)eventQueue).add(event)) {
                Channel ch = (Channel)i$.next();
                eventQueue = (List)reqChannelQueue.get(ch);
                if (eventQueue == null) {
                    eventQueue = new ArrayList();
                    reqChannelQueue.put(ch, eventQueue);
                }
            }

            batch = this.selector.getOptionalChannels(event);

            Object eventQueue;
            for(i$ = batch.iterator(); i$.hasNext(); ((List)eventQueue).add(event)) {
                Channel ch = (Channel)i$.next();
                eventQueue = (List)optChannelQueue.get(ch);
                if (eventQueue == null) {
                    eventQueue = new ArrayList();
                    optChannelQueue.put(ch, eventQueue);
                }
            }
        }

        i$ = reqChannelQueue.keySet().iterator();

        Channel optChannel;
        Transaction tx;
        Event event;
        while(i$.hasNext()) {
            optChannel = (Channel)i$.next();
            tx = optChannel.getTransaction();
            Preconditions.checkNotNull(tx, "Transaction object must not be null");

            try {
                tx.begin();
                batch = (List)reqChannelQueue.get(optChannel);
                i$ = batch.iterator();

                while(i$.hasNext()) {
                    event = (Event)i$.next();
                    optChannel.put(event);
                }

                tx.commit();
            } catch (Throwable var23) {
                tx.rollback();
                if (var23 instanceof Error) {
                    LOG.error("Error while writing to required channel: " + optChannel, var23);
                    throw (Error)var23;
                }

                if (var23 instanceof ChannelException) {
                    throw (ChannelException)var23;
                }

                throw new ChannelException("Unable to put batch on required channel: " + optChannel, var23);
            } finally {
                if (tx != null) {
                    tx.close();
                }

            }
        }

        i$ = optChannelQueue.keySet().iterator();

        while(i$.hasNext()) {
            optChannel = (Channel)i$.next();
            tx = optChannel.getTransaction();
            Preconditions.checkNotNull(tx, "Transaction object must not be null");

            try {
                tx.begin();
                batch = (List)optChannelQueue.get(optChannel);
                i$ = batch.iterator();

                while(i$.hasNext()) {
                    event = (Event)i$.next();
                    optChannel.put(event);
                }

                tx.commit();
            } catch (Throwable var21) {
                tx.rollback();
                LOG.error("Unable to put batch on optional channel: " + optChannel, var21);
                if (var21 instanceof Error) {
                    throw (Error)var21;
                }
            } finally {
                if (tx != null) {
                    tx.close();
                }

            }
        }

    }

参照主机名拦截器HostInterceptor ---写限速拦截器  实现  Interceptor 

//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by Fernflower decompiler)
//

package org.apache.flume.interceptor;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class HostInterceptor implements Interceptor {
    private static final Logger logger = LoggerFactory.getLogger(HostInterceptor.class);
    private final boolean preserveExisting;
    private final String header;
    private String host;

    private HostInterceptor(boolean preserveExisting, boolean useIP, String header) {
        this.host = null;
        this.preserveExisting = preserveExisting;
        this.header = header;

        try {
            InetAddress addr = InetAddress.getLocalHost();
            if (useIP) {
                this.host = addr.getHostAddress();
            } else {
                this.host = addr.getCanonicalHostName();
            }
        } catch (UnknownHostException var6) {
            logger.warn("Could not get local host address. Exception follows.", var6);
        }

    }

    public void initialize() {
    }

    public Event intercept(Event event) {
        Map<String, String> headers = event.getHeaders();
        if (this.preserveExisting && headers.containsKey(this.header)) {
            return event;
        } else {
            if (this.host != null) {
                headers.put(this.header, this.host);
            }

            return event;
        }
    }

    public List<Event> intercept(List<Event> events) {
        Iterator i$ = events.iterator();

        while(i$.hasNext()) {
            Event event = (Event)i$.next();
            this.intercept(event);
        }

        return events;
    }

    public void close() {
    }

    public static class Constants {
        public static String HOST = "host";
        public static String PRESERVE = "preserveExisting";
        public static boolean PRESERVE_DFLT = false;
        public static String USE_IP = "useIP";
        public static boolean USE_IP_DFLT = true;
        public static String HOST_HEADER = "hostHeader";

        public Constants() {
        }
    }

    public static class Builder implements org.apache.flume.interceptor.Interceptor.Builder {
        private boolean preserveExisting;
        private boolean useIP;
        private String header;

        public Builder() {
            this.preserveExisting = HostInterceptor.Constants.PRESERVE_DFLT;
            this.useIP = HostInterceptor.Constants.USE_IP_DFLT;
            this.header = HostInterceptor.Constants.HOST;
        }

        public Interceptor build() {
            return new HostInterceptor(this.preserveExisting, this.useIP, this.header);
        }

        public void configure(Context context) {
            this.preserveExisting = context.getBoolean(HostInterceptor.Constants.PRESERVE, HostInterceptor.Constants.PRESERVE_DFLT);
            this.useIP = context.getBoolean(HostInterceptor.Constants.USE_IP, HostInterceptor.Constants.USE_IP_DFLT);
            this.header = context.getString(HostInterceptor.Constants.HOST_HEADER, HostInterceptor.Constants.HOST);
        }
    }
}