采集端 Agent 监听目录，定时采集数据发送到汇聚端 Agent。汇聚端按一定的规则将采集文件持久化到本地目录。 典型场景描述：每天凌晨3点扫描主机 host1，/data/logs/ 目录下的文件，采集新生成的日志文件，发送到汇聚端。

需要注意的是：此场景采集的数据都是静态文件，即，如果被采集的文件在采集过程中发生变化（文件更新），可能会导致传输数据错误等问题。

可以制定监听目录采集数据支持:

• 定时扫描目录采集，支持配置unix cron表达式的方式，配置采集定时任务
• 指定序列化方式，默认采用按行序列化的方式
• 支持正则表达式过滤采集文件名
• 断点续传
• 流量控制（限流）

需要自定义开发 source 以及自定义 interceptor（流量控制）实现上述需求。

汇聚端即接收端，将接收到的数据持久化到本地目录，支持：

• 输出文件名与源文件名一致
• 按主机ip/主机名分类输出归档文件

需要开发自定义 sink 实现上述需求

整个采集的流程与模块设计如下图所示：

主要分为三个模块，采集端自定义source，中间自定义拦截器进行流量限制，接收端自定义sink，自动归档文件到本地。具体使用配置如下：

自定义source 配置如下：

加粗黑体部分为必须配置的项目，黄色标记部分为定时配置。

支持两种配置定时任务的方式

1. 标准Unix cron expression 配置定时任务
2. 定义初始delay值 ( initialDelay )，与定时循环周期 (pollNewDelay) 的方式

在1，2方式都配置的情况下，第1种 - cron 表达式的优先级更高，如果不配置cron，则以第二种方式的默认值定时执行任务（立即执行采集任务，采集完毕后，24小时后再次扫描）。

ignorePattern 配置正则，配置需要过滤的文件即不需要采集的文件。默认值为^$,即不过滤文件。例如可以配置 ignorePattern : ^.*.tmp$ ,过滤 .tmp 文件。

设计了flume消息通知机制，当采集的文件全部传输完毕后，发送一条空消息，在消息header 加入采集完毕的标记 fileDone，接收端每次会检测是否收到结束标记，收到即将文件重命名。

自定义sink 配置如下：

doneFilesTag 的配置需要与 source 中的 fileDoneHeaderKey 对应上，目前默认值为：fileDnoe。

自定义interceptor 配置如下：

该 interceptor 配置在 source端，即数据发送端，限制每秒数据发送的流量。

将项目打包，将以下jar包拷贝至 flume lib 目录下即可。

• flume-source-sink-1.0-SNAPSHOT.jar
• cron4j-2.2.5.jar
• cron-utils-4.1.0.jar