flume源码

public static void main(String[] args) {

    try {
      SSLUtil.initGlobalSSLParameters();
    
      // 用了apache commons-cli的jar包实现的命令行提示

      Options options = new Options();

      Option option = new Option("n", "name", true, "the name of this agent");
      option.setRequired(true);
      options.addOption(option);

      option = new Option("f", "conf-file", true,
          "specify a config file (required if -z missing)");
      option.setRequired(false);
      options.addOption(option);

      option = new Option(null, "no-reload-conf", false,
          "do not reload config file if changed");
      options.addOption(option);

      // Options for Zookeeper
      option = new Option("z", "zkConnString", true,
          "specify the ZooKeeper connection to use (required if -f missing)");
      option.setRequired(false);
      options.addOption(option);

      option = new Option("p", "zkBasePath", true,
          "specify the base path in ZooKeeper for agent configs");
      option.setRequired(false);
      options.addOption(option);

      option = new Option("h", "help", false, "display help text");
      options.addOption(option);

      CommandLineParser parser = new GnuParser();
      CommandLine commandLine = parser.parse(options, args);

      if (commandLine.hasOption(‘h‘)) {
        new HelpFormatter().printHelp("flume-ng agent", options, true);
        return;
      }

      String agentName = commandLine.getOptionValue(‘n‘);
      boolean reload = !commandLine.hasOption("no-reload-conf");

     
// 代码待续

// 代码接上面
 boolean isZkConfigured = false;
      if (commandLine.hasOption(‘z‘) || commandLine.hasOption("zkConnString")) {
        isZkConfigured = true;
      }

      Application application;
      if (isZkConfigured) {
        // get options
        String zkConnectionStr = commandLine.getOptionValue(‘z‘);
        String baseZkPath = commandLine.getOptionValue(‘p‘);

        if (reload) {
          EventBus eventBus = new EventBus(agentName + "-event-bus");
          List<LifecycleAware> components = Lists.newArrayList();
          PollingZooKeeperConfigurationProvider zookeeperConfigurationProvider =
              new PollingZooKeeperConfigurationProvider(
                  agentName, zkConnectionStr, baseZkPath, eventBus);
          components.add(zookeeperConfigurationProvider);
          application = new Application(components);
          eventBus.register(application);
        } else {
          StaticZooKeeperConfigurationProvider zookeeperConfigurationProvider =
              new StaticZooKeeperConfigurationProvider(
                  agentName, zkConnectionStr, baseZkPath);
          application = new Application();
          application.handleConfigurationEvent(zookeeperConfigurationProvider.getConfiguration());
        }
      } else {
       
// 代码待续

// 代码接上面
 
// 使用本地配置文件进行配置
File configurationFile = new File(commandLine.getOptionValue(‘f‘));

        /*
         * The following is to ensure that by default the agent will fail on
         * startup if the file does not exist.
         */
        if (!configurationFile.exists()) {
          // If command line invocation, then need to fail fast
          if (System.getProperty(Constants.SYSPROP_CALLED_FROM_SERVICE) ==
              null) {
            String path = configurationFile.getPath();
            try {
              path = configurationFile.getCanonicalPath();
            } catch (IOException ex) {
              logger.error("Failed to read canonical path for file: " + path,
                  ex);
            }
            throw new ParseException(
                "The specified configuration file does not exist: " + path);
          }
        }
        
// 代码待续

// 代码接上方       
        List<LifecycleAware> components = Lists.newArrayList();

        if (reload) {
          EventBus eventBus = new EventBus(agentName + "-event-bus");
          PollingPropertiesFileConfigurationProvider configurationProvider =
              new PollingPropertiesFileConfigurationProvider(
                  agentName, configurationFile, eventBus, 30);
          components.add(configurationProvider);
          application = new Application(components);
          eventBus.register(application);
        } else {
          ......
         // else块中代码在下方
        }
      }
      application.start();

     
// 代码待续

public PollingPropertiesFileConfigurationProvider(String agentName,
      File file, EventBus eventBus, int interval) {
    super(agentName, file);
    this.eventBus = eventBus;
    this.file = file;
    this.interval = interval;
    counterGroup = new CounterGroup();
    lifecycleState = LifecycleState.IDLE;
  }

  @Override
  public void start() {
    LOGGER.info("Configuration provider starting");

    Preconditions.checkState(file != null,
        "The parameter file must not be null");

    executorService = Executors.newSingleThreadScheduledExecutor(
            new ThreadFactoryBuilder().setNameFormat("conf-file-poller-%d")
                .build());

    FileWatcherRunnable fileWatcherRunnable =
        new FileWatcherRunnable(file, counterGroup);

    executorService.scheduleWithFixedDelay(fileWatcherRunnable, 0, interval,
        TimeUnit.SECONDS);

    lifecycleState = LifecycleState.START;

    LOGGER.debug("Configuration provider started");
  }

  @Override
  public void stop() {
    LOGGER.info("Configuration provider stopping");

    executorService.shutdown();
    try {
      if (!executorService.awaitTermination(500, TimeUnit.MILLISECONDS)) {
        LOGGER.debug("File watcher has not terminated. Forcing shutdown of executor.");
        executorService.shutdownNow();
        while (!executorService.awaitTermination(500, TimeUnit.MILLISECONDS)) {
          LOGGER.debug("Waiting for file watcher to terminate");
        }
      }
    } catch (InterruptedException e) {
      LOGGER.debug("Interrupted while waiting for file watcher to terminate");
      Thread.currentThread().interrupt();
    }
    lifecycleState = LifecycleState.STOP;
    LOGGER.debug("Configuration provider stopped");
  }

  @Override
  public synchronized  LifecycleState getLifecycleState() {
    return lifecycleState;
  }


  @Override
  public String toString() {
    return "{ file:" + file + " counterGroup:" + counterGroup + "  provider:"
        + getClass().getCanonicalName() + " agentName:" + getAgentName() + " }";
  }

  public class FileWatcherRunnable implements Runnable {

    private final File file;
    private final CounterGroup counterGroup;

    private long lastChange;

    public FileWatcherRunnable(File file, CounterGroup counterGroup) {
      super();
      this.file = file;
      this.counterGroup = counterGroup;
      this.lastChange = 0L;
    }

    @Override
    public void run() {
      LOGGER.debug("Checking file:{} for changes", file);

      counterGroup.incrementAndGet("file.checks");

      long lastModified = file.lastModified();

      if (lastModified > lastChange) {
        LOGGER.info("Reloading configuration file:{}", file);

        counterGroup.incrementAndGet("file.loads");

        lastChange = lastModified;

        try {
          eventBus.post(getConfiguration());
        } catch (Exception e) {
          LOGGER.error("Failed to load configuration data. Exception follows.",
              e);
        } catch (NoClassDefFoundError e) {
          LOGGER.error("Failed to start agent because dependencies were not " +
              "found in classpath. Error follows.", e);
        } catch (Throwable t) {
          // caught because the caller does not handle or log Throwables
          LOGGER.error("Unhandled error", t);
        }
      }
    }
  }

}

@Subscribe
  public void handleConfigurationEvent(MaterializedConfiguration conf) {
    try {
      lifecycleLock.lockInterruptibly();
      stopAllComponents();
      startAllComponents(conf);
    } catch (InterruptedException e) {
      logger.info("Interrupted while trying to handle configuration event");
      return;
    } finally {
      // If interrupted while trying to lock, we don‘t own the lock, so must not attempt to unlock
      if (lifecycleLock.isHeldByCurrentThread()) {
        lifecycleLock.unlock();
      }
    }
  }

 public void start() {
    lifecycleLock.lock();
    try {
      for (LifecycleAware component : components) {
        supervisor.supervise(component,
            new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START);
      }
    } finally {
      lifecycleLock.unlock();
    }
  }

public synchronized void supervise(LifecycleAware lifecycleAware,
      SupervisorPolicy policy, LifecycleState desiredState) {
    if (this.monitorService.isShutdown()
        || this.monitorService.isTerminated()
        || this.monitorService.isTerminating()) {
      throw new FlumeException("Supervise called on " + lifecycleAware + " " +
          "after shutdown has been initiated. " + lifecycleAware + " will not" +
          " be started");
    }

    Preconditions.checkState(!supervisedProcesses.containsKey(lifecycleAware),
        "Refusing to supervise " + lifecycleAware + " more than once");

    if (logger.isDebugEnabled()) {
      logger.debug("Supervising service:{} policy:{} desiredState:{}",
          new Object[] { lifecycleAware, policy, desiredState });
    }

    Supervisoree process = new Supervisoree();
    process.status = new Status();

    process.policy = policy;
    process.status.desiredState = desiredState;
    process.status.error = false;

    MonitorRunnable monitorRunnable = new MonitorRunnable();
    monitorRunnable.lifecycleAware = lifecycleAware;
    monitorRunnable.supervisoree = process;
    monitorRunnable.monitorService = monitorService;

    supervisedProcesses.put(lifecycleAware, process);

    ScheduledFuture<?> future = monitorService.scheduleWithFixedDelay(
        monitorRunnable, 0, 3, TimeUnit.SECONDS);
    monitorFutures.put(lifecycleAware, future);
  }

@Override
    public void run() {
      logger.debug("checking process:{} supervisoree:{}", lifecycleAware,
          supervisoree);

      long now = System.currentTimeMillis();

      try {
        if (supervisoree.status.firstSeen == null) {
          logger.debug("first time seeing {}", lifecycleAware);

          supervisoree.status.firstSeen = now;
        }

        supervisoree.status.lastSeen = now;
        synchronized (lifecycleAware) {
          if (supervisoree.status.discard) {
            // Unsupervise has already been called on this.
            logger.info("Component has already been stopped {}", lifecycleAware);
            return;
          } else if (supervisoree.status.error) {
            logger.info("Component {} is in error state, and Flume will not"
                + "attempt to change its state", lifecycleAware);
            return;
          }

          supervisoree.status.lastSeenState = lifecycleAware.getLifecycleState();

          if (!lifecycleAware.getLifecycleState().equals(
              supervisoree.status.desiredState)) {

            logger.debug("Want to transition {} from {} to {} (failures:{})",
                new Object[] { lifecycleAware, supervisoree.status.lastSeenState,
                    supervisoree.status.desiredState,
                    supervisoree.status.failures });

            switch (supervisoree.status.desiredState) {
              case START:
                try {
                  lifecycleAware.start();
                } catch (Throwable e) {
                  logger.error("Unable to start " + lifecycleAware
                      + " - Exception follows.", e);
                  if (e instanceof Error) {
                    // This component can never recover, shut it down.
                    supervisoree.status.desiredState = LifecycleState.STOP;
                    try {
                      lifecycleAware.stop();
                      logger.warn("Component {} stopped, since it could not be"
                          + "successfully started due to missing dependencies",
                          lifecycleAware);
                    } catch (Throwable e1) {
                      logger.error("Unsuccessful attempt to "
                          + "shutdown component: {} due to missing dependencies."
                          + " Please shutdown the agent"
                          + "or disable this component, or the agent will be"
                          + "in an undefined state.", e1);
                      supervisoree.status.error = true;
                      if (e1 instanceof Error) {
                        throw (Error) e1;
                      }
                      // Set the state to stop, so that the conf poller can
                      // proceed.
                    }
                  }
                  supervisoree.status.failures++;
                }
                break;
              case STOP:
                try {
                  lifecycleAware.stop();
                } catch (Throwable e) {
                  logger.error("Unable to stop " + lifecycleAware
                      + " - Exception follows.", e);
                  if (e instanceof Error) {
                    throw (Error) e;
                  }
                  supervisoree.status.failures++;
                }
                break;
              default:
                logger.warn("I refuse to acknowledge {} as a desired state",
                    supervisoree.status.desiredState);
            }

            if (!supervisoree.policy.isValid(lifecycleAware, supervisoree.status)) {
              logger.error(
                  "Policy {} of {} has been violated - supervisor should exit!",
                  supervisoree.policy, lifecycleAware);
            }
          }
        }
      } catch (Throwable t) {
        logger.error("Unexpected error", t);
      }
      logger.debug("Status check complete");
    }

  public MaterializedConfiguration getConfiguration() {
    MaterializedConfiguration conf = new SimpleMaterializedConfiguration();
    FlumeConfiguration fconfig = getFlumeConfiguration();
    AgentConfiguration agentConf = fconfig.getConfigurationFor(getAgentName());
    if (agentConf != null) {
      Map<String, ChannelComponent> channelComponentMap = Maps.newHashMap();
      Map<String, SourceRunner> sourceRunnerMap = Maps.newHashMap();
      Map<String, SinkRunner> sinkRunnerMap = Maps.newHashMap();
      try {
        loadChannels(agentConf, channelComponentMap);
        loadSources(agentConf, channelComponentMap, sourceRunnerMap);
        loadSinks(agentConf, channelComponentMap, sinkRunnerMap);
        Set<String> channelNames = new HashSet<String>(channelComponentMap.keySet());
        for (String channelName : channelNames) {
          ChannelComponent channelComponent = channelComponentMap.get(channelName);
          if (channelComponent.components.isEmpty()) {
            LOGGER.warn(String.format("Channel %s has no components connected" +
                " and has been removed.", channelName));
            channelComponentMap.remove(channelName);
            Map<String, Channel> nameChannelMap =
                channelCache.get(channelComponent.channel.getClass());
            if (nameChannelMap != null) {
              nameChannelMap.remove(channelName);
            }
          } else {
            LOGGER.info(String.format("Channel %s connected to %s",
                channelName, channelComponent.components.toString()));
            conf.addChannel(channelName, channelComponent.channel);
          }
        }
        for (Map.Entry<String, SourceRunner> entry : sourceRunnerMap.entrySet()) {
          conf.addSourceRunner(entry.getKey(), entry.getValue());
        }
        for (Map.Entry<String, SinkRunner> entry : sinkRunnerMap.entrySet()) {
          conf.addSinkRunner(entry.getKey(), entry.getValue());
        }
      } catch (InstantiationException ex) {
        LOGGER.error("Failed to instantiate component", ex);
      } finally {
        channelComponentMap.clear();
        sourceRunnerMap.clear();
        sinkRunnerMap.clear();
      }
    } else {
      LOGGER.warn("No configuration found for this host:{}", getAgentName());
    }
    return conf;
  }

private void loadChannels(AgentConfiguration agentConf,
    Map<String, ChannelComponent> channelComponentMap)
        throws InstantiationException {
  LOGGER.info("Creating channels");
 
  //todo 缓存中的 channel
  ListMultimap<Class<? extends Channel>, String> channelsNotReused =
      ArrayListMultimap.create();
  // assume all channels will not be re-used
  for (Map.Entry<Class<? extends Channel>, Map<String, Channel>> entry :
       channelCache.entrySet()) {
    Class<? extends Channel> channelKlass = entry.getKey();
    Set<String> channelNames = entry.getValue().keySet();
    channelsNotReused.get(channelKlass).addAll(channelNames);
  }
 
  Set<String> channelNames = agentConf.getChannelSet();
  Map<String, ComponentConfiguration> compMap = agentConf.getChannelConfigMap();
 
 
  for (String chName : channelNames) {
    ComponentConfiguration comp = compMap.get(chName);
    if (comp != null) {
 
      // todo 使用工厂类创建Channel
      Channel channel = getOrCreateChannel(channelsNotReused,
          comp.getComponentName(), comp.getType());
 
      try {
 
        //todo 更新配置 , 因为 channelComponentMap 刚开始传进来的时候是空值
        Configurables.configure(channel, comp);
        channelComponentMap.put(comp.getComponentName(),
            new ChannelComponent(channel));
        LOGGER.info("Created channel " + chName);
 
 
      } catch (Exception e) {
        String msg = String.format("Channel %s has been removed due to an " +
            "error during configuration", chName);
        LOGGER.error(msg, e);
      }
    }
  }
 
 
 
  //todo 组合没有 ComponentConfiguration配置, 仅仅使用Context的对象.
  for (String chName : channelNames) {
    Context context = agentConf.getChannelContext().get(chName);
    if (context != null) {
      // todo 使用工厂类创建Channel
      Channel channel = getOrCreateChannel(channelsNotReused, chName,
          context.getString(BasicConfigurationConstants.CONFIG_TYPE));
 
      try {
 
        // todo 更新配置 , 因为 channelComponentMap 刚开始传进来的时候是空值
        Configurables.configure(channel, context);
        channelComponentMap.put(chName, new ChannelComponent(channel));
 
        LOGGER.info("Created channel " + chName);
      } catch (Exception e) {
        String msg = String.format("Channel %s has been removed due to an " +
            "error during configuration", chName);
        LOGGER.error(msg, e);
      }
    }
  }
 
  for (Class<? extends Channel> channelKlass : channelsNotReused.keySet()) {
    Map<String, Channel> channelMap = channelCache.get(channelKlass);
    if (channelMap != null) {
      for (String channelName : channelsNotReused.get(channelKlass)) {
        if (channelMap.remove(channelName) != null) {
          LOGGER.info("Removed {} of type {}", channelName, channelKlass);
        }
      }
      if (channelMap.isEmpty()) {
        //todo 有一些 channel 在配置中没有重新使用, 将会将其从缓存中移除.
        channelCache.remove(channelKlass);
 
      }
    }
  }
}

private Channel getOrCreateChannel(
    ListMultimap<Class<? extends Channel>, String> channelsNotReused,
    String name, String type)
    throws FlumeException {
 
  // todo 根据传入的类型, 获取对应的类
  Class<? extends Channel> channelClass = channelFactory.getClass(type);
 
  /*
   * Channel has requested a new instance on each re-configuration
   * todo 根据新的配置, 实例化对象.
   */
  //todo 如何类的注解 Disposable 存在, 则直接进行实例化,并返回  只有 jdbc 和 file 模式用到了
  if (channelClass.isAnnotationPresent(Disposable.class)) {
    Channel channel = channelFactory.create(name, type);
    channel.setName(name);
    return channel;
  }
 
  Map<String, Channel> channelMap = channelCache.get(channelClass);
 
  //todo 如果缓存中不存在 channel 的话, 那么直接加入缓存.
  if (channelMap == null) {
    channelMap = new HashMap<String, Channel>();
    channelCache.put(channelClass, channelMap);
  }
  //todo 如果channelMap 中的 channel 为 null ,使用工厂类创建.
  Channel channel = channelMap.get(name);
  if (channel == null) {
    channel = channelFactory.create(name, type);
    channel.setName(name);
    channelMap.put(name, channel);
  }
  //todo 如果缓存中已经存在对应的 channel 的话,那么移除它, 后续的方法会更新它 .
  channelsNotReused.get(channelClass).remove(name);
  return channel;
}

  private void loadSources(AgentConfiguration agentConf,
      Map<String, ChannelComponent> channelComponentMap,
      Map<String, SourceRunner> sourceRunnerMap)
      throws InstantiationException {

    Set<String> sourceNames = agentConf.getSourceSet();
    Map<String, ComponentConfiguration> compMap =
        agentConf.getSourceConfigMap();
    /*
     * Components which have a ComponentConfiguration object
     */
    for (String sourceName : sourceNames) {
      ComponentConfiguration comp = compMap.get(sourceName);
      if (comp != null) {
        SourceConfiguration config = (SourceConfiguration) comp;

        Source source = sourceFactory.create(comp.getComponentName(),
            comp.getType());
        try {
          Configurables.configure(source, config); //配置source
          Set<String> channelNames = config.getChannels();
          List<Channel> sourceChannels =
                  getSourceChannels(channelComponentMap, source, channelNames);
          if (sourceChannels.isEmpty()) {
            String msg = String.format("Source %s is not connected to a " +
                "channel",  sourceName);
            throw new IllegalStateException(msg);
          }
          ChannelSelectorConfiguration selectorConfig =
              config.getSelectorConfiguration();

          ChannelSelector selector = ChannelSelectorFactory.create(
              sourceChannels, selectorConfig); //创建channel 选择器

          ChannelProcessor channelProcessor = new ChannelProcessor(selector);
          Configurables.configure(channelProcessor, config); // 这里实际是调用ChannelProcessor的configure()方法，设置拦截器链

          source.setChannelProcessor(channelProcessor);
          sourceRunnerMap.put(comp.getComponentName(),
              SourceRunner.forSource(source));
          for (Channel channel : sourceChannels) {
            ChannelComponent channelComponent =
                Preconditions.checkNotNull(channelComponentMap.get(channel.getName()),
                                           String.format("Channel %s", channel.getName()));
            channelComponent.components.add(sourceName);
          }
        } catch (Exception e) {
          String msg = String.format("Source %s has been removed due to an " +
              "error during configuration", sourceName);
          LOGGER.error(msg, e);
        }
      }
    }
    /*
     * Components which DO NOT have a ComponentConfiguration object
     * and use only Context
     */
    Map<String, Context> sourceContexts = agentConf.getSourceContext();
    for (String sourceName : sourceNames) {
      Context context = sourceContexts.get(sourceName);
      if (context != null) {
        Source source =
            sourceFactory.create(sourceName,
                                 context.getString(BasicConfigurationConstants.CONFIG_TYPE));
        try {
          Configurables.configure(source, context);
          String[] channelNames = context.getString(
              BasicConfigurationConstants.CONFIG_CHANNELS).split("\\s+");
          List<Channel> sourceChannels =
                  getSourceChannels(channelComponentMap, source, Arrays.asList(channelNames));
          if (sourceChannels.isEmpty()) {
            String msg = String.format("Source %s is not connected to a " +
                "channel",  sourceName);
            throw new IllegalStateException(msg);
          }
          Map<String, String> selectorConfig = context.getSubProperties(
              BasicConfigurationConstants.CONFIG_SOURCE_CHANNELSELECTOR_PREFIX);

          ChannelSelector selector = ChannelSelectorFactory.create(
              sourceChannels, selectorConfig);

          ChannelProcessor channelProcessor = new ChannelProcessor(selector);
          Configurables.configure(channelProcessor, context);
          source.setChannelProcessor(channelProcessor);
          sourceRunnerMap.put(sourceName,
              SourceRunner.forSource(source));
          for (Channel channel : sourceChannels) {
            ChannelComponent channelComponent =
                Preconditions.checkNotNull(channelComponentMap.get(channel.getName()),
                                           String.format("Channel %s", channel.getName()));
            channelComponent.components.add(sourceName);
          }
        } catch (Exception e) {
          String msg = String.format("Source %s has been removed due to an " +
              "error during configuration", sourceName);
          LOGGER.error(msg, e);
        }
      }
    }
  }

 public static SourceRunner forSource(Source source) {
    SourceRunner runner = null;

    if (source instanceof PollableSource) {
      runner = new PollableSourceRunner();
      ((PollableSourceRunner) runner).setSource((PollableSource) source);
    } else if (source instanceof EventDrivenSource) {
      runner = new EventDrivenSourceRunner();
      ((EventDrivenSourceRunner) runner).setSource((EventDrivenSource) source);
    } else {
      throw new IllegalArgumentException("No known runner type for source "
          + source);
    }

    return runner;
  }

private void loadSinks(AgentConfiguration agentConf,
    Map<String, ChannelComponent> channelComponentMap, Map<String, SinkRunner> sinkRunnerMap)
    throws InstantiationException {
 
 
  Set<String> sinkNames = agentConf.getSinkSet();
  Map<String, ComponentConfiguration> compMap =
      agentConf.getSinkConfigMap();
  Map<String, Sink> sinks = new HashMap<String, Sink>();
  /*
   * Components which have a ComponentConfiguration object
   * todo 组合配置ComponentConfiguration 的对象
   */
  for (String sinkName : sinkNames) {
    ComponentConfiguration comp = compMap.get(sinkName);
    if (comp != null) {
 
      //todo 使用SinkFactory 直接采用根据类型,采用反射方式  实例化 Sink
      SinkConfiguration config = (SinkConfiguration) comp;
      Sink sink = sinkFactory.create(comp.getComponentName(), comp.getType());
 
 
      try {
 
        //todo 为 Sink 匹配对应的 channel
        Configurables.configure(sink, config);
        ChannelComponent channelComponent = channelComponentMap.get(config.getChannel());
        if (channelComponent == null) {
          String msg = String.format("Sink %s is not connected to a " +
              "channel",  sinkName);
          throw new IllegalStateException(msg);
        }
 
        //todo 检查 channel 是否可用 :  sink 的 batch size 要小于 channel 的 transaction capacity
        checkSinkChannelCompatibility(sink, channelComponent.channel);
 
        sink.setChannel(channelComponent.channel);
 
        sinks.put(comp.getComponentName(), sink);
 
        //todo Sink 向 channel 反向注册 SinkName
        channelComponent.components.add(sinkName);
      } catch (Exception e) {
        String msg = String.format("Sink %s has been removed due to an " +
            "error during configuration", sinkName);
        LOGGER.error(msg, e);
      }
    }
  }
  /*
   * Components which DO NOT have a ComponentConfiguration object
   * and use only Context
   * todo 组合没有配置 ComponentConfiguration 但是使用 context  的对象
   */
  Map<String, Context> sinkContexts = agentConf.getSinkContext();
  for (String sinkName : sinkNames) {
    Context context = sinkContexts.get(sinkName);
    if (context != null) {
 
      //todo 直接采用根据类型,采用反射方式  实例化 Sink
      Sink sink = sinkFactory.create(sinkName, context.getString(
          BasicConfigurationConstants.CONFIG_TYPE));
 
 
      try {
 
        //todo 为 Sink 匹配对应的 channel
        Configurables.configure(sink, context);
        ChannelComponent channelComponent =
            channelComponentMap.get(
                context.getString(BasicConfigurationConstants.CONFIG_CHANNEL));
        if (channelComponent == null) {
          String msg = String.format("Sink %s is not connected to a " +
              "channel",  sinkName);
          throw new IllegalStateException(msg);
        }
        //todo 检查 channel 是否可用 :  sink 的 batch size 要大于 channel 的 transaction capacity
        checkSinkChannelCompatibility(sink, channelComponent.channel);
 
 
        sink.setChannel(channelComponent.channel);
        sinks.put(sinkName, sink);
        channelComponent.components.add(sinkName);
 
      } catch (Exception e) {
        String msg = String.format("Sink %s has been removed due to an " +
            "error during configuration", sinkName);
        LOGGER.error(msg, e);
      }
    }
  }
 
  //todo 对 sink 进行分组
  loadSinkGroups(agentConf, sinks, sinkRunnerMap);
}

private void loadSinkGroups(AgentConfiguration agentConf,
    Map<String, Sink> sinks, Map<String, SinkRunner> sinkRunnerMap)
        throws InstantiationException {
 
 
  // todo 获取配置中的 group 分组
  Set<String> sinkGroupNames = agentConf.getSinkgroupSet();
 
 
  Map<String, ComponentConfiguration> compMap =
      agentConf.getSinkGroupConfigMap();
 
  Map<String, String> usedSinks = new HashMap<String, String>();
 
  for (String groupName: sinkGroupNames) {
 
    ComponentConfiguration comp = compMap.get(groupName);
 
    if (comp != null) {
      SinkGroupConfiguration groupConf = (SinkGroupConfiguration) comp;
      List<Sink> groupSinks = new ArrayList<Sink>();
      for (String sink : groupConf.getSinks()) {
        Sink s = sinks.remove(sink);
        if (s == null) {
          String sinkUser = usedSinks.get(sink);
          if (sinkUser != null) {
            throw new InstantiationException(String.format(
                "Sink %s of group %s already " +
                    "in use by group %s", sink, groupName, sinkUser));
          } else {
            throw new InstantiationException(String.format(
                "Sink %s of group %s does "
                    + "not exist or is not properly configured", sink,
                    groupName));
          }
        }
        groupSinks.add(s);
        usedSinks.put(sink, groupName);
      }
      try {
 
        SinkGroup group = new SinkGroup(groupSinks);
        Configurables.configure(group, groupConf);
 
        //todo 创建 sinkRunner
        sinkRunnerMap.put(comp.getComponentName(),
            new SinkRunner(group.getProcessor()));
 
      } catch (Exception e) {
        String msg = String.format("SinkGroup %s has been removed due to " +
            "an error during configuration", groupName);
        LOGGER.error(msg, e);
      }
    }
  }
 
  // add any unassigned sinks to solo collectors
  // todo 对未分组的 sink 进行处理
  for (Entry<String, Sink> entry : sinks.entrySet()) {
    if (!usedSinks.containsValue(entry.getKey())) {
      try {
 
        SinkProcessor pr = new DefaultSinkProcessor();
        List<Sink> sinkMap = new ArrayList<Sink>();
        sinkMap.add(entry.getValue());
        pr.setSinks(sinkMap);
        Configurables.configure(pr, new Context());
 
        //todo 创建 SinkRunner
        sinkRunnerMap.put(entry.getKey(), new SinkRunner(pr));
        
      } catch (Exception e) {
        String msg = String.format("SinkGroup %s has been removed due to " +
            "an error during configuration", entry.getKey());
        LOGGER.error(msg, e);
      }
    }
  }
 
 
}

          PropertiesFileConfigurationProvider configurationProvider = new PropertiesFileConfigurationProvider(agentName, configurationFile);
          application = new Application();
          application.handleConfigurationEvent(configurationProvider.getConfiguration());

 // 代码接上方

      final Application appReference = application;
      Runtime.getRuntime().addShutdownHook(new Thread("agent-shutdown-hook") {
        @Override
        public void run() {
          appReference.stop();
        }
      });

    } catch (Exception e) {
      logger.error("A fatal error occurred while running. Exception follows.", e);
    }
  }
}

@InterfaceAudience.Public
@InterfaceStability.Stable
public interface Channel extends LifecycleAware, NamedComponent {

  public void put(Event event) throws ChannelException;
  public Event take() throws ChannelException;
  public Transaction getTransaction();
}

public class ChannelProcessor implements Configurable {
    private static final Logger LOG = LoggerFactory.getLogger(ChannelProcessor.class);
    private final ChannelSelector selector;
    private final InterceptorChain interceptorChain;

    public ChannelProcessor(ChannelSelector selector) {
        this.selector = selector;
        this.interceptorChain = new InterceptorChain();
    }

......

    public ChannelSelector getSelector() {
        return this.selector;
    }

    public void processEventBatch(List<Event> events) {
        ...
        while(i$.hasNext()) {
            Event optChannel = (Event)i$.next();
            List tx = this.selector.getRequiredChannels(optChannel);

            ...//将event放到Required队列

            t1 = this.selector.getOptionalChannels(optChannel);

            Object eventQueue;
            ...//将event放到Optional队列

        }

        ...//event的分配操作

    }

    public void processEvent(Event event) {
        event = this.interceptorChain.intercept(event);
        if(event != null) {
            List requiredChannels = this.selector.getRequiredChannels(event);
            Iterator optionalChannels = requiredChannels.iterator();

            ...//event的分配操作

            List optionalChannels1 = this.selector.getOptionalChannels(event);
            Iterator i$1 = optionalChannels1.iterator();

            ...//event的分配操作
        }
    }
}

public interface ChannelSelector extends NamedComponent, Configurable {
    void setChannels(List<Channel> var1);

    List<Channel> getRequiredChannels(Event var1);

    List<Channel> getOptionalChannels(Event var1);

    List<Channel> getAllChannels();
}

  public static ChannelSelector create(List<Channel> channels,
      ChannelSelectorConfiguration conf) {
    String type = ChannelSelectorType.REPLICATING.toString();
    if (conf != null) {
      type = conf.getType();
    }
    ChannelSelector selector = getSelectorForType(type);
    selector.setChannels(channels);
    Configurables.configure(selector, conf);
    return selector;
  }

@InterfaceAudience.Public
@InterfaceStability.Stable
public interface Sink extends LifecycleAware, NamedComponent {
  public void setChannel(Channel channel);
  public Channel getChannel();
  /* 用来处理channel中取来的event*/
  public Status process() throws EventDeliveryException;
  public static enum Status {
    READY, BACKOFF
  }
}

public class SinkRunner implements LifecycleAware {

  ...

  @Override
  public void start() {
    SinkProcessor policy = getPolicy();

    policy.start();

    runner = new PollingRunner();

    runner.policy = policy;
    runner.counterGroup = counterGroup;
    runner.shouldStop = new AtomicBoolean();

    runnerThread = new Thread(runner);
    runnerThread.setName("SinkRunner-PollingRunner-" +
        policy.getClass().getSimpleName());
    runnerThread.start();

    lifecycleState = LifecycleState.START;
  }
  ...

}

public interface SinkProcessor extends LifecycleAware, Configurable {
  Status process() throws EventDeliveryException;
  void setSinks(List<Sink> sinks);
}

public class SinkGroup implements Configurable, ConfigurableComponent {
  List<Sink> sinks;
  SinkProcessor processor;
  SinkGroupConfiguration conf;

  public SinkGroup(List<Sink> groupSinks) {
    sinks = groupSinks;
  }

  public SinkProcessor getProcessor() {
    return processor;
  }

  @Override
  public void configure(ComponentConfiguration conf) {
    this.conf = (SinkGroupConfiguration) conf;
    processor =
        SinkProcessorFactory.getProcessor(this.conf.getProcessorContext(),
            sinks);
  }
}

public class DefaultSinkProcessor implements SinkProcessor, ConfigurableComponent {
  private Sink sink;
  private LifecycleState lifecycleState;

  @Override
  public void start() {
    Preconditions.checkNotNull(sink, "DefaultSinkProcessor sink not set");
    sink.start();
    lifecycleState = LifecycleState.START;
  }

  @Override
  public void stop() {
    Preconditions.checkNotNull(sink, "DefaultSinkProcessor sink not set");
    sink.stop();
    lifecycleState = LifecycleState.STOP;
  }

  @Override
  public LifecycleState getLifecycleState() {
    return lifecycleState;
  }

  @Override
  public void configure(Context context) {
  }

  @Override
  public Status process() throws EventDeliveryException {
    return sink.process();
  }

  @Override
  public void setSinks(List<Sink> sinks) {
    Preconditions.checkNotNull(sinks);
    Preconditions.checkArgument(sinks.size() == 1, "DefaultSinkPolicy can "
        + "only handle one sink, "
        + "try using a policy that supports multiple sinks");
    sink = sinks.get(0);
  }

  @Override
  public void configure(ComponentConfiguration conf) {

  }

}