Flume-NG源码分析-整体结构及配置载入分析

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");

if (commandLine.hasOption('z') || commandLine.hasOption("zkConnString")) {
  isZkConfigured = true;
}

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 {
    PropertiesFileConfigurationProvider configurationProvider =
      new PropertiesFileConfigurationProvider(
        agentName, configurationFile);
    application = new Application();
    application.handleConfigurationEvent(configurationProvider
      .getConfiguration());
  }
}
application.start();

public interface LifecycleAware {

  /**
   * <p>
   * Starts a service or component.
   * </p>
   * @throws LifecycleException
   * @throws InterruptedException
   */
  public void start();

  /**
   * <p>
   * Stops a service or component.
   * </p>
   * @throws LifecycleException
   * @throws InterruptedException
   */
  public void stop();

  /**
   * <p>
   * Return the current state of the service or component.
   * </p>
   */
  public LifecycleState getLifecycleState();

}

public synchronized void start() {
  for(LifecycleAware component : components) {
    supervisor.supervise(component,
        new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START);
  }
}

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

eventBus.post(getConfiguration())。

public MaterializedConfiguration getConfiguration() {
 //初始化三大组件的配置Map，source，channel，sink
  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;
}

@Override
public FlumeConfiguration getFlumeConfiguration() {
  BufferedReader reader = null;
  try {
    reader = new BufferedReader(new FileReader(file));
    Properties properties = new Properties();
    properties.load(reader);
    return new FlumeConfiguration(toMap(properties));
  } catch (IOException ex) {
    LOGGER.error("Unable to load file:" + file
        + " (I/O failure) - Exception follows.", ex);
  } finally {
    if (reader != null) {
      try {
        reader.close();
      } catch (IOException ex) {
        LOGGER.warn(
            "Unable to close file reader for file: " + file, ex);
      }
    }
  }
  return new FlumeConfiguration(new HashMap<String, String>());
}

cnck = parseConfigKey(key, BasicConfigurationConstants.CONFIG_SINKGROUPS_PREFIX);

public final class BasicConfigurationConstants {

public static final String CONFIG_SOURCES = "sources";
public static final String CONFIG_SOURCES_PREFIX = CONFIG_SOURCES + ".";
public static final String CONFIG_SOURCE_CHANNELSELECTOR_PREFIX = "selector.";

public static final String CONFIG_SINKS = "sinks";
public static final String CONFIG_SINKS_PREFIX = CONFIG_SINKS + ".";
public static final String CONFIG_SINK_PROCESSOR_PREFIX = "processor.";

public static final String CONFIG_SINKGROUPS = "sinkgroups";
public static final String CONFIG_SINKGROUPS_PREFIX = CONFIG_SINKGROUPS + ".";

public static final String CONFIG_CHANNEL = "channel";
public static final String CONFIG_CHANNELS = "channels";
public static final String CONFIG_CHANNELS_PREFIX = CONFIG_CHANNELS + ".";

public static final String CONFIG_CONFIG = "config";
public static final String CONFIG_TYPE = "type";

private BasicConfigurationConstants() {
  // disable explicit object creation
}