在Flume原理剖析和安装部署章节，我们最后提到NetCat Source的实例，实现了监听一个指定的网络端口，只要在应用程序向这个端口里面写数据，这个NetCat Source组件就能获取到信息。本章内容继续讲解Flume Source的其他几个常用组件，Avro Source、Spool Source、Exec Source，后两者是属于自动监控读取文件的source组件。因为本文只测试Source组件，从而这里默认的Channel是 MemoryChannel，sink为LoggerSink。

一、Avro Source实战

Avro Source可以定制avro-client发送一个指定的文件给Flume agent，Avro源使用Avro RPC机制，Flume主要的RPC Source也是 Avro Source，它使用Netty-Avro inter-process的通信（IPC）协议来通信，因此可以用java或JVM语言发送数据到Avro Source端。它的配置文件主要包含三个参数：

• type： Avro source的别名是avro，也可以使用完整类别名称，org.apache.flume.source.AvroSource；
• bind： 绑定的IP地址或主机名。使用0.0.0.0绑定机器所有端口
• port： 绑定监听端口端口

利用Avro Source可以实现多级流动、扇出流、扇入流等效果，下面放一张flume官方经典图，多级流动，如下图所示。另外也可以接受通过flume提供的Avro客户端发送的日志信息（本文的示例就是使用avro-client发送日志信息）。

1、cd /usr/flume/conf 在conf目录下新建avro.conf ，vi avro.conf 。如果你之前已经写过了netcat.conf 你可以使用命令 cp netcat.conf avro.conf 之后修改avro里面的内容即可，不需要每次都进行全段全段的撰写，节约时间。

#example.conf: A single-node flume configuration
#It is Just for test flume spool Example

#Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

#定制source，绑定channel、主机以及端口
a1.sources.r1.type = avro
a1.sources.r1.channels = c1
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 4141

#Describe the sink
a1.sinks.k1.type = logger

#use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

#Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

整个配置文件内容分为三个大部分：1、从整体上描述代理agent中sources、sinks、channels所涉及到的组件；2、详细描述agent中每一个source、sink与channel的具体实现：即在描述source的时候，需要指定source到底是什么类型的，即这个source是接受文件的、还是接受http的、还是接受thrift的；对于sink则是描述使用什么类型的sink，本例中使用的是Logger Sink，将数据写入日志文件；对于channel需要指定是内存等；3、通过channel将source与sink连接起来。

2、使用命令，启动flume agent

[root@master bin]# flume-ng agent -n a1 -c /usr/flume/conf/ --conf-file /usr/flume/conf/avro.conf -Dflume.root.logger=INFO,console

参数解释：

-n 指定agent名称，要与配置文件一致（等价于--name)

-c 指定配置文件位置（等价于 --conf）

-f 指定agent配置文件（等价 --conf-file）

-Dflume.root.logger=INFO,console 设置日志等级

3、成功启动之后，如下所示，系统程序界面会一直在等待状态，等待接收avro-client发送的日志数据：

4、我们使用SSH 远程连接工具Secure CRT 重新Clone一下新Session，启动avro-client，这里需要指定发给给agent的文件具体位置，并指定端口

[root@master ~]# flume-ng avro-client -c /usr/flume/conf/ -H master -p 4141 -F /usr/flume/logs/avro_test.log

5、最后我们来查看一下结果，在运行上面的命令之后，agent界面中会显示avro_test.log中的文件内容，下图的字比较小，为了更好的理解，我把几张图拼在一起。

二、Spool Source实战

Spool Source监听一个指定的目录，即只要应用程序想这个指定的目录中添加新的文件，source组件就可以获取到该信息，并解析该文件的内容，然后写入到channel。等待sink处理完之后，标记该文件已完成处理，文件名添加.completed后缀。虽然是自动监控整个目录，但是只能监控文件，如果以追加的方式向已被处理的文件中添加内容，source并不能识别；换言之，如果在被spool source监控的目录下出现了一个新文件，则立即会被处理。

但值得注意的是：（1）拷贝到spool目录下的文件不可以再打开编辑。（2）spool目录下不可包含相应的子目录，即无法监控子目录的文件夹变动。

1、添加配置文件内容cd /usr/flume/conf 在conf目录下新建spool.conf ，vi spool.conf 。如果你之前已经写过了avro.conf 你可以使用命令 cp avro.conf spool.conf之后修改spool里面的内容即可，不需要每次都进行全段全段的撰写，节约时间。

#example.conf: A single-node flume configuration
#It is Just for test flume spool Example

#Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

#指定source 绑定channel和需要监控的文件夹
a1.sources.r1.type = spooldir
a1.sources.r1.channels = c1
a1.sources.r1.spoolDir = /usr/flume/logs
a1.sources.r1.fileHeader = true

#Describe the sink
a1.sinks.k1.type = logger

#use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

#Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

2、启动。使用命令，启动flume agent

[root@master bin]# flume-ng agent -n a1 -c /usr/flume/conf/ --conf-file /usr/flume/conf/spool.conf -Dflume.root.logger=INFO,console  

参数选项就不在这里做过多解释，如有需要，请参考Avro Source实战内容

3、成功启动之后，如下所示，系统程序界面会一直处于监控状态，监控文件夹下文件是否出现变动。在这个界面中明显已经开始读完一个名为avro_test.log的文件，headers为正在读的文件的位置，body为一行的内容，但是显示的只有16位，这里面用黄色框标出的即这个文件夹中的内容，即我们在上面测试用的avro_test.log文件当时存放的正式logs文件夹下。

4、为了测试其工作方式，我们在其他会话窗口创建一个test_spool.log文件，放入到logs文件夹下，并观察agent窗口的变化。

从上图可以看出，测试用的test_spool.log一旦放入到被监测的logs文件夹下之后，会立即被处理，并且被打上处理完成的标记。Tips：我们在测试的时候会通过vi命令来创建一个文件，也可以echo "spool test" >> test_spool.log 以追加的方式写入新内容；或者使用cat命令，cat其他文件的内容追加到文件中。

三、Exec Source实战

监听一个指定的命令，获取一条命令的结果作为它的数据源，source组件从这个命令的结果中取数据。常用的是tail -F 【file】指令，即只要应用程序向日志(文件)里面写数据，source组件就可以获取到日志(文件)中最新的内容 ，EXEC执行一个给定的命令获得输出的源,如果要使用tail命令，必选使得file足够大才能看到输出内容

1、添加配置文件内容cd /usr/flume/conf 在conf目录下新建exec.conf ，vi exec.conf

#example.conf: A single-node flume configuration
#It is Just for test flume exec Example

#Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

#配置source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /usr/flume/logs/exec_tail_test.log
a1.sources.r1.channels = c1

#Describe the sink
a1.sinks.k1.type = logger

#use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

#Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

2、启动。使用命令，启动flume agent

[root@master ~]# flume-ng agent -n a1 -c /usr/flume/conf/ -f /usr/flume/conf/exec.conf -Dflume.root.logger=INFO,console

3、成功启动之后，如下所示，系统程序界面会一直处于监控状态，监控该命令下文件是否出现变动，从上面的日志信息可以看出，相关组件进程均已成功启动。

4、如果要使用tail命令，必选使得file足够大才能看到输出内容，因此为了让agent界面能够监控到tail之后的结果，我们往监控的文件中循环插入100条数据：

5、查看agent中的数据