图文精华

About云-梭伦科技»专题 › 技术学习(版主发帖区) › 大数据学习 › Flume › 连载型› Flume-NG启动过程源码分析（三）
分享

Flume-NG启动过程源码分析（三）

xioaxu790 2014-8-17 15:31:06 发表于 连载型 [显示全部楼层] 回帖奖励 阅读模式 关闭右栏 0 10954
About云VIP帮助找工作和提升
问题导读
1、什么是是SimpleMaterializedConfiguration的一个实例？
2、如何分析startAllComponents(conf)方法？
3、flume的所有组件均实现自什么 接口？





      本篇分析加载配置文件后各个组件是如何运行的？
　　加载完配置文件订阅者Application类会收到订阅信息执行：
  1.   @Subscribe

  2.   public synchronized void handleConfigurationEvent(MaterializedConfiguration conf) {

  3.     stopAllComponents();

  4.     startAllComponents(conf);

  5.   }
复制代码


　　MaterializedConfiguration conf就是getConfiguration()方法获取的配置信息，是SimpleMaterializedConfiguration的一个实例。

　　handleConfigurationEvent方法在前面章节(一)中有过大致分析，包括：stopAllComponents()和startAllComponents(conf)。Application中的materializedConfiguration就是MaterializedConfiguration conf，stopAllComponents()方法中的materializedConfiguration是旧的配置信息，需要先停掉旧的组件，然后startAllComponents(conf)将新的配置信息赋给materializedConfiguration并依次启动各个组件。

　　1、先看startAllComponents(conf)方法。代码如下：

  1. private void startAllComponents(MaterializedConfiguration materializedConfiguration) {//启动所有组件最基本的三大组件

  2.     logger.info("Starting new configuration:{}", materializedConfiguration);

  3. 

  4.     this.materializedConfiguration = materializedConfiguration;

  5. 

  6.     for (Entry<String, Channel> entry :

  7.       materializedConfiguration.getChannels().entrySet()) {

  8.       try{

  9.         logger.info("Starting Channel " + entry.getKey());

  10.         supervisor.supervise(entry.getValue(),

  11.             new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START);

  12.       } catch (Exception e){

  13.         logger.error("Error while starting {}", entry.getValue(), e);

  14.       }

  15.     }

  16. 

  17.     /*

  18.      * Wait for all channels to start.等待所有channel启动完毕

  19.      */

  20.     for(Channel ch: materializedConfiguration.getChannels().values()){

  21.       while(ch.getLifecycleState() != LifecycleState.START

  22.           && !supervisor.isComponentInErrorState(ch)){

  23.         try {

  24.           logger.info("Waiting for channel: " + ch.getName() +

  25.               " to start. Sleeping for 500 ms");

  26.           Thread.sleep(500);

  27.         } catch (InterruptedException e) {

  28.           logger.error("Interrupted while waiting for channel to start.", e);

  29.           Throwables.propagate(e);

  30.         }

  31.       }

  32.     }

  33. 

  34.     for (Entry<String, SinkRunner> entry : materializedConfiguration.getSinkRunners()

  35.         .entrySet()) {        //启动所有sink

  36.       try{

  37.         logger.info("Starting Sink " + entry.getKey());

  38.         supervisor.supervise(entry.getValue(),

  39.           new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START);

  40.       } catch (Exception e) {

  41.         logger.error("Error while starting {}", entry.getValue(), e);

  42.       }

  43.     }

  44. 

  45.     for (Entry<String, SourceRunner> entry : materializedConfiguration

  46.         .getSourceRunners().entrySet()) {//启动所有source

  47.       try{

  48.         logger.info("Starting Source " + entry.getKey());

  49.         supervisor.supervise(entry.getValue(),

  50.           new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START);

  51.       } catch (Exception e) {

  52.         logger.error("Error while starting {}", entry.getValue(), e);

  53.       }

  54.     }

  55. 

  56.     this.loadMonitoring();

  57.   }
复制代码


　　三大组件都是通过supervisor.supervise(entry.getValue(),new SupervisorPolicy.AlwaysRestartPolicy(), LifecycleState.START)启动的，其中，channel启动之后还要待所有的channel完全启动完毕之后才可再去启动sink和source。如果channel没有启动完毕就去启动另外俩组件，会出现错误，以为一旦sink或者source建立完毕就会立即与channel通信获取数据。稍后会分别分析sink和source的启动。

　　supervisor是LifecycleSupervisor的一个对象，该类的构造方法会构造一个有10个线程，上限是20的线程池供各大组件使用。构造方法如下：

  1. public LifecycleSupervisor() {

  2.     lifecycleState = LifecycleState.IDLE;

  3.     supervisedProcesses = new HashMap<LifecycleAware, Supervisoree>();//存储所有历史上的组件及其监控信息

  4.     monitorFutures = new HashMap<LifecycleAware, ScheduledFuture<?>>();

  5.     monitorService = new ScheduledThreadPoolExecutor(10,

  6.         new ThreadFactoryBuilder().setNameFormat(

  7.             "lifecycleSupervisor-" + Thread.currentThread().getId() + "-%d")

  8.             .build());

  9.     monitorService.setMaximumPoolSize(20);

  10.     monitorService.setKeepAliveTime(30, TimeUnit.SECONDS);

  11.     purger = new Purger();

  12.     needToPurge = false;

  13.   }
复制代码


　　supervise(LifecycleAware lifecycleAware,SupervisorPolicy policy, LifecycleState desiredState)方法则是具体执行启动各个组件的方法。flume的所有组件均实现自LifecycleAware 接口，如图：，这个接口就三个方法getLifecycleState(返回组件运行状态)、start(组件启动)、stop(停止组件)。supervise方法代码如下：
  1. public synchronized void supervise(LifecycleAware lifecycleAware,

  2.       SupervisorPolicy policy, LifecycleState desiredState) {

  3. 　　//检查线程池状态

  4.     if(this.monitorService.isShutdown()

  5.         || this.monitorService.isTerminated()

  6.         || this.monitorService.isTerminating()){

  7.       throw new FlumeException("Supervise called on " + lifecycleAware + " " +

  8.           "after shutdown has been initiated. " + lifecycleAware + " will not" +

  9.           " be started");

  10.     }

  11. 　　//如果该组件已经在监控，则拒绝二次监控

  12.     Preconditions.checkState(!supervisedProcesses.containsKey(lifecycleAware),

  13.         "Refusing to supervise " + lifecycleAware + " more than once");

  14. 

  15.     if (logger.isDebugEnabled()) {

  16.       logger.debug("Supervising service:{} policy:{} desiredState:{}",

  17.           new Object[] { lifecycleAware, policy, desiredState });

  18.     }

  19. 　　//新的组件

  20.     Supervisoree process = new Supervisoree();

  21.     process.status = new Status();

  22. 

  23.     process.policy = policy;

  24.     process.status.desiredState = desiredState;

  25.     process.status.error = false;

  26. 

  27.     MonitorRunnable monitorRunnable = new MonitorRunnable();

  28.     monitorRunnable.lifecycleAware = lifecycleAware;//组件

  29.     monitorRunnable.supervisoree = process;

  30.     monitorRunnable.monitorService = monitorService;

  31. 

  32.     supervisedProcesses.put(lifecycleAware, process);

  33.     //创建并执行一个在给定初始延迟后首次启用的定期操作，随后，在每一次执行终止和下一次执行开始之间都存在给定的延迟。如果任务的任一执行遇到异常，就会取消后续执行。

  34.     ScheduledFuture<?> future = monitorService.scheduleWithFixedDelay(

  35.         monitorRunnable, 0, 3, TimeUnit.SECONDS);  //启动MonitorRunnable,结束之后3秒再重新启动，可以用于重试

  36.     monitorFutures.put(lifecycleAware, future);

  37.   }
复制代码




　　该方法首先monitorService是否是正常运行状态；然后构造Supervisoree process = new Supervisoree()，进行赋值并构造一个监控进程MonitorRunnable，放入线程池去执行。

　　MonitorRunnable.run()方法：

  1. public void run() {

  2.       logger.debug("checking process:{} supervisoree:{}", lifecycleAware,

  3.           supervisoree);

  4. 

  5.       long now = System.currentTimeMillis();//获取现在的时间戳

  6. 

  7.       try {

  8.         if (supervisoree.status.firstSeen == null) {

  9.           logger.debug("first time seeing {}", lifecycleAware);

  10. 　　　　　　//如果这个组件是是初次受监控

  11.           supervisoree.status.firstSeen = now;

  12.         }

  13. 　　　　　//如果这个组件已经监控过

  14.         supervisoree.status.lastSeen = now;

  15.         synchronized (lifecycleAware) {//锁住组件

  16.           if (supervisoree.status.discard) {//该组件已经停止监控

  17.             // Unsupervise has already been called on this.

  18.             logger.info("Component has already been stopped {}", lifecycleAware);

  19.             return;//直接返回

  20.           } else if (supervisoree.status.error) {//该组件是错误状态

  21.             logger.info("Component {} is in error state, and Flume will not"

  22.                 + "attempt to change its state", lifecycleAware);

  23.             return;//直接返回

  24.           }

  25. 

  26.           supervisoree.status.lastSeenState = lifecycleAware.getLifecycleState();//获取组件最新状态,没运行start()方法之前是LifecycleState.IDLE状态

  27. 

  28.           if (!lifecycleAware.getLifecycleState().equals(

  29.               supervisoree.status.desiredState)) {//该组件最新状态和期望的状态不一致

  30. 

  31.             logger.debug("Want to transition {} from {} to {} (failures:{})",

  32.                 new Object[] { lifecycleAware, supervisoree.status.lastSeenState,

  33.                     supervisoree.status.desiredState,

  34.                     supervisoree.status.failures });

  35. 

  36.             switch (supervisoree.status.desiredState) {//根据状态执行相应的操作

  37.               case START:

  38.                 try {

  39.                   lifecycleAware.start();   //启动组件，同时其状态也会变为LifecycleState.START

  40.                 } catch (Throwable e) {

  41.                   logger.error("Unable to start " + lifecycleAware

  42.                       + " - Exception follows.", e);

  43.                   if (e instanceof Error) {

  44.                     // This component can never recover, shut it down.

  45.                     supervisoree.status.desiredState = LifecycleState.STOP;

  46.                     try {

  47.                       lifecycleAware.stop();

  48.                       logger.warn("Component {} stopped, since it could not be"

  49.                           + "successfully started due to missing dependencies",

  50.                           lifecycleAware);

  51.                     } catch (Throwable e1) {

  52.                       logger.error("Unsuccessful attempt to "

  53.                           + "shutdown component: {} due to missing dependencies."

  54.                           + " Please shutdown the agent"

  55.                           + "or disable this component, or the agent will be"

  56.                           + "in an undefined state.", e1);

  57.                       supervisoree.status.error = true;

  58.                       if (e1 instanceof Error) {

  59.                         throw (Error) e1;

  60.                       }

  61.                       // Set the state to stop, so that the conf poller can

  62.                       // proceed.

  63.                     }

  64.                   }

  65.                   supervisoree.status.failures++;//启动错误失败次数+1

  66.                 }

  67.                 break;

  68.               case STOP:

  69.                 try {

  70.                   lifecycleAware.stop();　　　　//停止组件

  71.                 } catch (Throwable e) {

  72.                   logger.error("Unable to stop " + lifecycleAware

  73.                       + " - Exception follows.", e);

  74.                   if (e instanceof Error) {

  75.                     throw (Error) e;

  76.                   }

  77.                   supervisoree.status.failures++;　　//组件停止错误，错误次数+1

  78.                 }

  79.                 break;

  80.               default:

  81.                 logger.warn("I refuse to acknowledge {} as a desired state",

  82.                     supervisoree.status.desiredState);

  83.             }

  84. 　　　　　　　//两种SupervisorPolicy(AlwaysRestartPolicy和OnceOnlyPolicy)后者还未使用过，前者表示可以重新启动的组件，后者表示只能运行一次的组件

  85.             if (!supervisoree.policy.isValid(lifecycleAware, supervisoree.status)) {

  86.               logger.error(

  87.                   "Policy {} of {} has been violated - supervisor should exit!",

  88.                   supervisoree.policy, lifecycleAware);

  89.             }

  90.           }

  91.         }

  92.       } catch(Throwable t) {

  93.         logger.error("Unexpected error", t);

  94.       }

  95.       logger.debug("Status check complete");

  96.     }
复制代码



　　上面的 lifecycleAware.stop()和lifecycleAware.start()就是执行的sink、source、channel等的对应方法。

　　这里的start需要注意如果是channel则是直接执行start方法；如果是sink或者PollableSource的实现类，则会在start()方法中启动一个线程来循环的调用process()方法来从channel拿数据(sink)或者向channel送数据(source);如果是EventDrivenSource的实现类，则没有process()方法，通过执行start()来执行想channel中送数据的操作(可以在此添加线程来实现相应的逻辑)。

　　2、stopAllComponents()方法。顾名思义，就是停止所有组件的方法。该方法代码如下：

  1. private void stopAllComponents() {

  2.     if (this.materializedConfiguration != null) {

  3.       logger.info("Shutting down configuration: {}", this.materializedConfiguration);

  4.       for (Entry<String, SourceRunner> entry : this.materializedConfiguration

  5.           .getSourceRunners().entrySet()) {

  6.         try{

  7.           logger.info("Stopping Source " + entry.getKey());

  8.           supervisor.unsupervise(entry.getValue());

  9.         } catch (Exception e){

  10.           logger.error("Error while stopping {}", entry.getValue(), e);

  11.         }

  12.       }

  13. 

  14.       for (Entry<String, SinkRunner> entry :

  15.         this.materializedConfiguration.getSinkRunners().entrySet()) {

  16.         try{

  17.           logger.info("Stopping Sink " + entry.getKey());

  18.           supervisor.unsupervise(entry.getValue());

  19.         } catch (Exception e){

  20.           logger.error("Error while stopping {}", entry.getValue(), e);

  21.         }

  22.       }

  23. 

  24.       for (Entry<String, Channel> entry :

  25.         this.materializedConfiguration.getChannels().entrySet()) {

  26.         try{

  27.           logger.info("Stopping Channel " + entry.getKey());

  28.           supervisor.unsupervise(entry.getValue());

  29.         } catch (Exception e){

  30.           logger.error("Error while stopping {}", entry.getValue(), e);

  31.         }

  32.       }

  33.     }

  34.     if(monitorServer != null) {

  35.       monitorServer.stop();

  36.     }

  37.   }
复制代码




　　首先，需要注意的是，stopAllComponents()放在startAllComponents(MaterializedConfiguration materializedConfiguration)方法之前的原因，由于配置文件的动态加载这一特性的存在，使得每次加载之前都要先把旧的组件停掉，然后才能去加载最新配置文件中的配置；

　　其次，首次执行stopAllComponents()时，由于配置文件尚未赋值，所以并不会执行停止所有组件的操作以及停止monitorServer。再次加载时会依照顺序依次停止对source、sink以及channel的监控，通过supervisor.unsupervise(entry.getValue())停止对其的监控，然后停止monitorServer。supervisor.unsupervise方法如下：
  1. public synchronized void unsupervise(LifecycleAware lifecycleAware) {

  2. 

  3.     Preconditions.checkState(supervisedProcesses.containsKey(lifecycleAware),

  4.         "Unaware of " + lifecycleAware + " - can not unsupervise");

  5. 

  6.     logger.debug("Unsupervising service:{}", lifecycleAware);

  7. 

  8.     synchronized (lifecycleAware) {

  9.     Supervisoree supervisoree = supervisedProcesses.get(lifecycleAware);

  10.     supervisoree.status.discard = true;

  11.       this.setDesiredState(lifecycleAware, LifecycleState.STOP);

  12.       logger.info("Stopping component: {}", lifecycleAware);

  13.       lifecycleAware.stop();

  14.     }

  15.     supervisedProcesses.remove(lifecycleAware);

  16.     //We need to do this because a reconfiguration simply unsupervises old

  17.     //components and supervises new ones.

  18.     monitorFutures.get(lifecycleAware).cancel(false);

  19.     //purges are expensive, so it is done only once every 2 hours.

  20.     needToPurge = true;

  21.     monitorFutures.remove(lifecycleAware);

  22.   }
复制代码


　　该方法首先会检查正在运行的组件当中是否有此组件supervisedProcesses.containsKey(lifecycleAware)；如果存在，则对此组件标记为已取消监控supervisoree.status.discard = true；将状态设置为STOP，并停止组件lifecycleAware.stop()；然后从删除此组件的监控记录，包括从记录正在处于监控的组件的结构supervisedProcesses以及记录组件及其对应的运行线程的结构monitorFutures中删除相应的组件信息，并且needToPurge = true会使得两小时执行一次的线程池清理操作。

　　有一个问题就是，sink和source是如何找到对应的channel的呢？？其实前面章节就已经讲解过，分别在AbstractConfigurationProvider.loadSources方法中通过ChannelSelector配置source对应的channel，而在source中通过getChannelProcessor()获取channels,通过channelProcessor.processEventBatch(eventList)将events发送到channel中；而在AbstractConfigurationProvider.loadSinks方法中sink.setChannel(channelComponent.channel)来设置此sink对应的channel，然后在sink的实现类中通过getChannel()获取设置的channel，并使用channel.take()从channel中获取event进行处理。

　　以上三节是Flume-NG的启动、配置文件的加载、配置文件的动态加载、组件的执行的整个流程。文中的疏漏之处，请各位指教，我依然会后续继续完善这些内容的。





本文转载自：http://www.cnblogs.com/lxf20061900/p/3679240.html
关注公众号，获取大数据、人工智能20套、区块链资源5阶段等资源，随时更新，获取最新技术资源
回复

使用道具 举报

自定义广告语

没找到任何评论，期待你打破沉寂

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
xioaxu790

金牌会员

关注

  • 851

    主题

  • 1000

    帖子

  • 108

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2024 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000