Flume日志采集工具使用详解2（样例1：接收netcat网络数据并写入日志文件）

作者：hangge | 2024-03-06 08:40

二、接收 netcat 网络数据并写入日志文件

1，整体架构

2，配置 Agent

（1）启动 Flume 任务其实就是启动一个 Agent。首先我们进入 flume 程序包中的 conf 文件夹，创建一个 Agent 配置文件：

cd conf
vi example.conf

（2）接着在配置文件中添加如下内容。我们首先定义了 Source、Channel 和 Sink 组件的名称；然后配置了 Source、Channel 和 Sink 组件的相关参数；最后把这 3 个组件连接到了一起，即告诉 Source 需要向哪个 Channel 写入数据，告诉 Sink 需要从哪个 Channel 读取数据，这样 Source、Channel 和 Sink 这 3 个组件就连接起来了。

提示：bind 参数后面指定的 ip 是四个 0，这个当前机器的通用 ip，因为一台机器可以有多个 ip，例如：内网 ip、外网 ip，如果通过 bind 参数指定某一个 ip 的话，表示就只监听通过这个 ip 发送过来的数据了，这样会有局限性，所以可以指定 0.0.0.0。

# 定义 Agent 内部 3 大组件的名称
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# 配置 Source 组件
a1.sources.r1.type = netcat
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 6666

# 配置 Sink 组件
a1.sinks.k1.type = logger

# 配置 Channel 组件
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# 将 Source 组件、Sink 组件和 Channel 组件绑定到一起
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

3，启动 Agent

（1）配置好 Agent 后就可以使用如下命令启动此 Agent：

参数说明：

agent：启动一个 Flume 的 Agent 代理。
--name：指定 Agent 的名字。
--conf：指定 Flume 配置文件的根目录。
--conf-file：指定 Agent 配置文件的位置。

bin/flume-ng agent --name a1 --conf conf --conf-file conf/example.conf

（2）使用上面命令启动成功之后，命令行窗口会被一直占用，因为 Agent 属于一个前台进程。我们也可以通过 nohup 方式启动（注意最后的 &），将 Agent 放到后台执行：

nohup bin/flume-ng agent --name a1 --conf conf --conf-file conf/example.conf &

（3）启动之后，通过“jps -m”命令可以查看到一个 Application 进程，它就是启动的 Agent。

jps -m

（4）或者使用 ps 命令也可以查看到这个进程：

ps -ef|grep flume

4，测试 Agent

（1）首先使用 telnet 连接配置的 6666 端口：

telnet 192.168.60.9 6666

（2）连接上后发送两条数据：

（3）查看 flume 日志文件，可以看到这两条数据已经被成功接收到并写入文件中：

cat flume.log

附：sink 使用 logger 时内容不全问题

1，问题描述

（1）根据设计，logger sink 将内容限制为 16 字节，从而避免屏幕充斥着过多的内容。比如我门使用 telnet 发送了一条如下 20 个字节的数据：

12345678901234567890

（2）可以看到无论是控制台的输出，还是日志文件里面的记录，消息内容都是被截断的，只保留 16 个字节。

2，解决办法

（1）logger sink 的 16 字节限制是其内部代码控制的，我们无法通过诸如修改 log4j2.xml 配置文件进行修改。如果想要查看调试的完整内容，那么我们应该使用其他的 sink。

（2）下面我们使用 file_roll sink，它会将日志写到本地文件系统中。

# 定义 Agent 内部 3 大组件的名称
a1.sources = r1
a1.sinks = k1
a1.channels = c1
 
# 配置 Source 组件
a1.sources.r1.type = netcat
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 6666
 
# 配置 Sink 组件
a1.sinks.k1.type=file_roll
a1.sinks.k1.sink.directory=/usr/local/logs
a1.sinks.k1.sink.rollInterval=60
 
# 配置 Channel 组件
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
 
# 将 Source 组件、Sink 组件和 Channel 组件绑定到一起
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

（3）上面配置，任务启动后会自动在指定目录下每隔 60 秒创建一个文件：

（4）查看文件内容可以看到消息是完整的：

大数据