Flume-ng启动过程分析

Application.java

入口函数main():

    ...
    //加载flume的配置文件，初始化Sink,Source,Channel的工厂类
    PropertiesFileConfigurationProvider configurationProvider =
            new PropertiesFileConfigurationProvider(agentName,
                configurationFile);
    application = new Application();
    //configurationProvider.getConfiguration()中实例化Sink,Source,Channel
    application.handleConfigurationEvent(configurationProvider.getConfiguration());-------getConfiguration------>
            //Map用于存储所有Sink,Source,Channel
            Map<String, ChannelComponent> channelComponentMap = Maps.newHashMap();
            Map<String, SourceRunner> sourceRunnerMap = Maps.newHashMap();
            Map<String, SinkRunner> sinkRunnerMap = Maps.newHashMap();
            //先实例化channel
            loadChannels(agentConf, channelComponentMap);
            //将Source对应的channel注册到ChannelSelector，Source通过ChannelSelector获取Channel
            loadSources(agentConf, channelComponentMap, sourceRunnerMap);
            //向Sink注册Channel
            loadSinks(agentConf, channelComponentMap, sinkRunnerMap);
            ...
            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());
            }
            ...
            return conf

    ...
    //application.handleConfigurationEvent(conf)---->
            stopAllComponents();
            startAllComponents(conf);
    final Application appReference = application;
    //关闭程序时，调用的钩子
    Runtime.getRuntime().addShutdownHook(new Thread("agent-shutdown-hook") {
    @Override
    public void run() {
      appReference.stop();
    }
  });

startAllComponents(conf):

//通过LifecycleSupervisor类启动组件//启动MonitorRunnable，监控Channelfor (Entry<String, Channel> entry :
      materializedConfiguration.getChannels().entrySet()) {
      try{
        logger.info("Starting Channel " + entry.getKey());
        supervisor.supervise(entry.getValue(),
            new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START);
      } catch (Exception e){
        logger.error("Error while starting {}", entry.getValue(), e);
      }
    }

    //等待启动
    for(Channel ch: materializedConfiguration.getChannels().values()){
      while(ch.getLifecycleState() != LifecycleState.START
          && !supervisor.isComponentInErrorState(ch)){
        try {
          logger.info("Waiting for channel: " + ch.getName() +
              " to start. Sleeping for 500 ms");
          Thread.sleep(500);
        } catch (InterruptedException e) {
          logger.error("Interrupted while waiting for channel to start.", e);
          Throwables.propagate(e);
        }
      }
    }

    //启动MonitorRunnable，监控sink
    for (Entry<String, SinkRunner> entry : materializedConfiguration.getSinkRunners()
        .entrySet()) {
      try{
        logger.info("Starting Sink " + entry.getKey());
        supervisor.supervise(entry.getValue(),
          new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START);
      } catch (Exception e) {
        logger.error("Error while starting {}", entry.getValue(), e);
      }
    }

    //启动MonitorRunnable，监控source
    for (Entry<String, SourceRunner> entry : materializedConfiguration
        .getSourceRunners().entrySet()) {
      try{
        logger.info("Starting Source " + entry.getKey());
        supervisor.supervise(entry.getValue(),
          new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START);
      } catch (Exception e) {
        logger.error("Error while starting {}", entry.getValue(), e);
      }
    }

    this.loadMonitoring();

LifecycleSupervisor.java

负责启动和监控Flume组件的类，功能如：失败重启组件
LifecycleSupervisor内部比较重要的几个变量：

//监控进程的线程池
ScheduledThreadPoolExecutor monitorService
Map<LifecycleAware, ScheduledFuture<?>> monitorFutures
Map<LifecycleAware, Supervisoree> supervisedProcesses

//启动监控
  public synchronized void supervise(LifecycleAware lifecycleAware,
      SupervisorPolicy policy, LifecycleState desiredState) {
    ...

    //组件状态
    Supervisoree process = new Supervisoree();
    process.status = new Status();
    process.policy = policy;
    process.status.desiredState = desiredState;
    process.status.error = false;

    //监控线程，调用启动组件的线程。比如
    MonitorRunnable monitorRunnable = new MonitorRunnable();---->
            lifecycleAware.start();--->
                //如果是sink
                sinRunner.start()---->
                    runnerThread = new Thread(runner);
                    runnerThread.setName("SinkRunner-PollingRunner-" +
                    policy.getClass().getSimpleName());
                    runnerThread.start();---->
                        //在新线程里循环调用
                        DefaultSinkProcessor.process();---->
                            //sink从channel中取数据，进行处理
                            sink.process();



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