在1.5.0的flume版本中开始提供这个功能，判断配置文件的更新时间戳来reload服务
原理：
1）在启动中使用EventBus.register注册Application对象，同时Application有一个Subscribe的方法handleConfigurationEvent（参数是MaterializedConfiguration对象）
2）定义了一个计划任务线程池，检测到文件更新情况（判断文件的更新时间）
3）如果检测到文件有更新会使用EventBus.post方法发送这个event(MaterializedConfiguration对象)
4）调用Application.handleConfigurationEvent重启各个组件

下面看具体实现：
在org.apache.flume.node.Application的main方法中：
支持reload时

1
2
3
4
5
6
7
         EventBus eventBus =  new  EventBus(agentName +  "-event-bus"  );  //实例化一个EventBus对象
         PollingPropertiesFileConfigurationProvider configurationProvider =
             new  PollingPropertiesFileConfigurationProvider(agentName,
                 configurationFile, eventBus,  30 );  //默认的检测文件是否更新的interval时间为30s
         components.add(configurationProvider);  //添加到启动列表中，在start方法中会启动PollingPropertiesFileConfigurationProvider 的计划任务线程池
         application =  new  Application(components);
         eventBus.register(application);  //向EventBus中注册Application对象，让Application对象作为时间的监听者

org.apache.flume.node.PollingPropertiesFileConfigurationProvider //扩展了PropertiesFileConfigurationProvider类并实现了LifecycleAware接口
PollingPropertiesFileConfigurationProvider是一个有生命周期概念的监控配置更改的服务：
start方法：

1
2
3
4
5
6
7
8
9
10
11
12
   public  void  start() {
....
     executorService = Executors.newSingleThreadScheduledExecutor(
             new  ThreadFactoryBuilder().setNameFormat( "conf-file-poller-%d"  )
                 .build());  //定义一个单线程的任务调度池
     FileWatcherRunnable fileWatcherRunnable =
         new  FileWatcherRunnable(file , counterGroup );  //构建一个FileWatcherRunnable服务对象
     executorService.scheduleWithFixedDelay(fileWatcherRunnable,  0 , interval,
         TimeUnit. SECONDS);  //以interval为时间间隔运行FileWatcherRunnable
     lifecycleState = LifecycleState. START;  // 设置为START状态
...
   }

FileWatcherRunnable是一个实现了Runnable 接口的线程类：
其主要的run方法

1
2
3
4
5
6
7
8
9
10
11
12
13
     public  void  run() {
...
       long  lastModified = file.lastModified();  //调用File.lastModified获取文件的上一次更新时的时间戳
       if  (lastModified > lastChange) {  //初始lastChange 为0
...
         lastChange = lastModified;  // 设置本次的lastChange 为更新时间，如果下一次更新文件，lastModified 会大于这个lastChange 
         try  {
           eventBus.post(getConfiguration());
         } 
.....
       }
     }
   }

在检测到更新后调用eventBus.post(getConfiguration())向监听者发送信息，这里为Application对象，监听者调用注释为Subscribe的方法处理信息：

1
2
3
4
5
@Subscribe
public  synchronized  void  handleConfigurationEvent(MaterializedConfiguration conf) {  //reload服务
   stopAllComponents();
   startAllComponents(conf);
}