1. 项目背景与整体架构设计思路提到Flume搞日志采集的同学应该都不陌生。单机模式往下塞日志、怼到HDFS或者Kafka这类操作网上教程一抓一大把照着配一遍基本能跑通。但如果业务规模稍微上来一点——比如源服务器分散在几个机房、日志量一天几个TB、目标存储又不在同一台机器上——单机模式马上就会露怯要么采集源agent负载过高要么数据传输链路单点故障要么下游存储端因为并发压力直接被拖垮。这时候就得考虑把Flume的部署方式从“单兵作战”升级成“多节点协同”也就是标题里说的“Flume跨服务器采集数据多个Agent串联”。先说个最常见的误区很多刚开始接触Flume跨服务器采集的同学以为串联就是把两台服务器上分别装好Flume各自配一套source和sink然后靠网络“自动打通”。实际不是。Flume本身不自带“远程Agent发现”这种机制它只认数据通道。要实现跨服务器流转核心思路就是让一台Agent的Sink变成另一台Agent的Source——换句话说前一个节点通过特定协议把数据推出去后一个节点开启对应该协议的Source监听端口把数据接进来。用官方术语讲就是source/sink之间的channel机制在起核心作用。跨服务器串联有好几种玩法可以用Avro协议做Source/Sink配对把数据从采集机推到中转机也可以加个Kafka Channel或者Kafka Source做解耦缓冲还可以用File Channel保证数据落盘不丢。但最经典、也最能讲清楚原理、最方便排查问题的方案就是架构上做两层采集层和汇聚层。采集层部署在每台源服务器上负责监听业务日志文件汇聚层单独部署一台或几台中转服务器负责把多路数据汇总后统一打到最终目的地。这个思路和我实际在做的某个跨平台系统项目里日志收集模块的落地方式完全一致。为什么选“串联”而不是“并联”因为并联要求所有源agent直接把数据推给HDFS或者Kafka这会造成目标端连接数暴涨而且任何一个源节点的网络抖动都可能直接把数据憋在本地甚至丢失。串联之后汇聚层作为一个天然的缓冲和转发节点可以有效控制下游连接数量同时便于集中配置sink的批量参数、压缩参数和重试策略。这一点我们从最开始就得想明白不然越往后改架构越痛苦。2. 核心配置逐段拆解从采集端到汇聚端2.1 环境抽象与版本选择先说环境。我这里以一个模拟项目X来做演示涉及三台服务器分别承担不同角色源服务器 node-01装有业务服务日志写在 /data/logs/business.log作为采集层Agent。汇聚服务器 node-02独立中转节点负责接收来自 node-01 的数据统一处理后转发出去。存储服务器 node-03可以理解为集群里的HDFS NameNode或者某台Kafka Broker作为最终数据落地目标。实际上这三台机器可以是物理机、虚拟机或者云主机只要内网能互通且端口开放就行。关于Flume版本我用的是Flume 1.9.0搭配JDK 1.8。很多公司生产环境还在用老版本1.6或者1.7不过建议尽早升到1.9因为1.7以下有些Source在新版操作系统上有兼容性问题而且1.9默认支持更强的Sink负载均衡组。安装路径规划成 /opt/flume数据目录规划成 /data/flume其中 /data/flume/checkpoint 和 /data/flume/data 是我们特别关心的两个目录——前者放File Channel的checkpoint文件后者放实际缓存的数据块。生产环境务必给这个目录留够磁盘空间一般建议按单日最大日志量的两倍来做容量规划。这个比例不是我拍脑袋而是File Channel一旦达到capacity上限source就会执行背压机制暂停读取文件如果磁盘再被写满那整个采集链路就会雪崩到时候想恢复都很被动。2.2 采集端Agent配置监听文件并推送第一台服务器上的flume配置如下文件名我放在 /opt/flume/conf/node-01.conf。这里用的是tail -F效果的TaildirSource加Avro Sink# node-01.conf agent1.sources src1 agent1.channels ch1 agent1.sinks sink1 # 配置source agent1.sources.src1.type TAILDIR agent1.sources.src1.channels ch1 agent1.sources.src1.positionFile /data/flume/taildir_position.json agent1.sources.src1.filegroups f1 agent1.sources.src1.filegroups.f1 /data/logs/business.log agent1.sources.src1.fileHeader false # 配置channel agent1.channels.ch1.type FILE agent1.channels.ch1.checkpointDir /data/flume/checkpoint agent1.channels.ch1.dataDirs /data/flume/data agent1.channels.ch1.capacity 1000000 agent1.channels.ch1.transactionCapacity 10000 agent1.channels.ch1.keep-alive 30 # 配置sink agent1.sinks.sink1.type avro agent1.sinks.sink1.hostname node-02 agent1.sinks.sink1.port 4141 agent1.sinks.sink1.channel ch1 agent1.sinks.sink1.batch-size 500 agent1.sinks.sink1.connect-timeout 30000 agent1.sinks.sink1.request-timeout 30000拆开说说每个关键点TaildirSource是现在最推荐的source类型可以断点续传重启之后不会从文件头重新读。positionFile那个json文件专门记录每个文件读到哪个偏移量了。如果哪天发现重启后日志重复采集优先看这个文件和实际文件inode的对应关系对不对。filegroups支持正则表达式我用绝对路径图省事你也可以写 /data/logs/.*.log 来匹配多个文件。Channel我用File Channel而不是默认的Memory Channel。原因很简单Memory Channel一旦agent进程被kill掉或者机器宕机channel里还没发出去的数据就全没了。File Channel是把数据先写到磁盘再转发虽然吞吐量稍微低一点但数据可靠性高了一个档次。在跨服务器采集这种链路上我宁愿慢一点也不想丢数据之后开发找上门来说“日志少了一段”。capacity和transactionCapacity这两个参数是配合使用的。capacity是channel里最多存多少条eventtransactionCapacity是每次事务从channel取多少条交给sink。我这边配的是sink批量大小batch-size为500transactionCapacity至少要大于等于这个数不然会报“无法从channel读取这么多事件”的错。实际上transactionCapacity比batch-size大一点更稳比如10000比500大很多这样即使多条source同时写入竞争也不容易卡事务。Avro Sink的目的地是node-02的4141端口。hostname就写内网IP别写主机名——除非你对DNS解析特别有把握否则很可能因解析失败导致连接异常这类问题在网络环境复杂的公司里非常常见。connect-timeout和request-timeout我拉得比较长因为跨服务器链路上偶尔的网络抖动如果超时太短sink会频繁进入重试循环白白消耗资源。2.3 汇聚端Agent配置Avro Source接收并下推第二台服务器node-02上的配置如下# node-02.conf agent2.sources src2 agent2.channels ch2 agent2.sinks sink2 agent2.sources.src2.type avro agent2.sources.src2.bind 0.0.0.0 agent2.sources.src2.port 4141 agent2.sources.src2.channels ch2 agent2.sources.src2.threads 8 agent2.channels.ch2.type FILE agent2.channels.ch2.checkpointDir /data/flume/checkpoint_node02 agent2.channels.ch2.dataDirs /data/flume/data_node02 agent2.channels.ch2.capacity 2000000 agent2.channels.ch2.transactionCapacity 20000 agent2.channels.ch2.keep-alive 30 agent2.sinks.sink2.type hdfs agent2.sinks.sink2.hdfs.path /user/logs/%Y%m%d agent2.sinks.sink2.hdfs.filePrefix business agent2.sinks.sink2.hdfs.fileType DataStream agent2.sinks.sink2.hdfs.writeFormat Text agent2.sinks.sink2.hdfs.rollInterval 3600 agent2.sinks.sink2.hdfs.rollSize 268435456 agent2.sinks.sink2.hdfs.rollCount 0 agent2.sinks.sink2.hdfs.batchSize 1000 agent2.sinks.sink2.hdfs.idleTimeout 60 agent2.sinks.sink2.channel ch2Avro Source的bind我写的是0.0.0.0意思是监听所有网卡接口。你要是只希望接收集群内某个网段的数据也可以写成具体IP这样安全性更高。port 4141要和前面sink里指定的端口一致端口冲突是常见的低级错误启动时如果报“Address already in use”八成是端口被别的服务占了。threads参数用于指定接收数据时的处理线程数这里设8你可以按源agent的数量和字节吞吐量做调整源端多的话这个值可以继续往上加——但注意这个参数并不是越大越好因为线程多了会让CPU上下文切换成本上升我一般建议先按源agent数量的1.5倍来设。HDFS Sink的 hdfs.path 用了日期格式化符会自动按天生成目录比如 /user/logs/20240601。这个方式配合后续离线数仓的按天分区很顺手。rollInterval是3600秒即一小时强制执行滚动生成新文件rollSize是268435456约256MB意思是文件达到这个大小就滚动。两者哪个先触发都可以。rollCount设0是表示不按event条数滚动因为日志量波动大的时候按条数滚动容易产生大量小文件对HDFS的NameNode不友好严重影响集群元数据性能。这个小文件问题很多教程不会提但实战中踩过坑的人应该都懂有时候一天能生成上万个几十KB的小文件给集群带来的压力比日志本身大得多。细看还有一个小参数 batchSize它控制每次批量写HDFS的事件数。这里设1000注意和前面的transactionCapacity匹配确保单次事务能容纳一个batch的数据。如果不是对应关系良好偶尔会看到超时或者批量写失败的情况查起来又麻烦又隐蔽。2.4 中转链路的可选变体为什么要用Avro协议而不是别的写到这里很多人可能会问为什么中间跨服务器的传输层不直接用Kafka或者别的消息队列做桥接偏要用Avro Source/Sink核心原因是Avro协议是Flume原生支持的远程传输方案它走的是Netty连接自带事件封装、批量和故障重试。配置上只需要指定hostname和port不需要维护额外的依赖组件也不需要在node-01上多部署一套客户端。如果你的日志链路只需要从A点到B点Avro是最省事的方案链路短、配置少、好排查。但这并不妨碍我们引入消息队列来增强链路缓冲能力。我可以直说如果业务量到了单机Flume承受不住的程度或者经过中转的数据多路复用、多个团队都要消费那在Avro链路之间插入Kafka是很成熟的做法。方式也不复杂——node-01的sink配成Kafka Sink直接把数据丢进某个topicnode-02用Kafka Source订阅这个topic然后再写HDFS或者推给下游系统。我最近重构的某个跨平台系统项目里就用了一个相对“复古”的写法中继agent仍然配的是Avro Source但source的channel后面多接了一个Kafka Sink作为备份分流这样既保留了Avro链路的主干又给实时计算留了通路。但标题里既然说“多个agent串联”那我们优先还是基于Avro做演示。你理解了这条链路后面怎么替换组件都顺理成章。3. 启动链路与验证手段详解3.1 按顺序启动agent节点启动顺序很讲究不能乱。正确顺序是先启动汇聚端node-02再启动采集端node-01。原因很好理解node-01启动后sink会立刻尝试连接node-02的4141端口如果node-02还没起来sink会进入重试循环并打印大量ERROR日志。这倒不会让agent崩溃但会让日志文件变得很嘈杂而且你要是这时候去观察node-01的状态很容易误判成“链路故障”。反过来先把node-02起好node-01启动后首次连接就能成功链路干净又顺滑。启动命令用的是flume-ng脚本我习惯通过指定conf文件和topology名称来管理# node-02上执行 /opt/flume/bin/flume-ng agent \ --name agent2 \ --conf /opt/flume/conf \ --conf-file /opt/flume/conf/node-02.conf \ -Dflume.root.loggerINFO,console # node-01上执行 /opt/flume/bin/flume-ng agent \ --name agent1 \ --conf /opt/flume/conf \ --conf-file /opt/flume/conf/node-01.conf \ -Dflume.root.loggerINFO,console 注意--name后面的名字必须和配置文件里agent前缀完全一致。比如node-01.conf里前缀是agent1--name就必须是agent1。如果不一致Flume启动后会直接找不到对应的配置段。这个事故发生的频率极高每次帮别人排查十次里有三四次是这里写错了。3.2 验证链路是否打通启动完成后第一步先看两端日志有没有报错。node-01这边如果出现“Avro sink 连接成功”之类的INFO日志说明到node-02的链路已经建立。node-02这边如果出现“Avro source 启动并绑定端口”的日志说明source端监听正常。第二步是往业务日志文件里写几行测试数据echo test line 1 /data/logs/business.log echo test line 2 /data/logs/business.log然后去node-03上查看HDFS目录确认文件是否生成、内容是否完整hdfs dfs -ls /user/logs/ hdfs dfs -cat /user/logs/20240601/*.0如果你没有HDFS环境也可以先临时把node-02的sink改成logger类型效果就是把event内容打在agent2的日志里。这个trick在链路排查阶段非常好用配置如下agent2.sinks.sink2.type logger agent2.sinks.sink2.channel ch2看到logger打出来的event带上了我们从node-01 push过去的数据说明采集端、channel、avro链路、汇聚端这一段全部正常。这时候再切回HDFS Sink也不迟。3.3 模拟源端日志滚动业务日志一般都会有滚动逻辑比如按大小切分、按天切分生成带时间戳或者序号的新文件。Flume的TaildirSource对文件滚动处理得很完善文件被rename之后source会根据position文件继续从断点处读取新文件内容不会丢数据也不会重读。为了验证这一点我们可以手动模拟一下滚动mv /data/logs/business.log /data/logs/business.log.20240601 touch /data/logs/business.log echo new log after roll /data/logs/business.log正常情况下node-03的HDFS目录里应该同时包含滚动前的数据和滚动后新写入的数据。如果出现重启后重复读取老数据的现象去看taildir_position.json的内容确认文件inode和position是否更新到了正确值。这个验证环节在我们当时做模拟项目X时帮了大忙正好把一套有bug的日志滚动脚本暴露了出来。4. 常见问题与排查技巧实录4.1 File Channel满导致采集停止典型现象node-01日志一直打印“FileChannel has reached its capacity”业务日志文件停止采集source处于背压状态。原因不外乎两种一是下游node-02处理速度跟不上产生速度写HDFS太慢二是sink到node-02的连接中断数据全堵在channel里。排查步骤我先看node-02的agent2日志有没有大量重试或者网络异常。然后再看node-01的channel目录占用du -sh /data/flume/data/*这个命令能迅速确认channel里堆积了多少数据。如果堆积很多而node-02又正常十有八九是HDFS Sink写文件太慢——检查hdfs.batchSize是不是设置太小或者HDFS节点磁盘是不是满了。有一次我排查到最后发现是HDFS的DataNode磁盘使用率到了98%sink一直在重试这时需要做的是紧急扩容磁盘或者临时把数据导入到备用路径。4.2 Avro连接超时与端口不通如果node-01启动后一直报类似“Connection refused”或者超时先别急着怀疑Flume配置。在node-01上用nc或者telnet测一下到node-02端口通不通telnet node-02 4141不同公司内网防火墙策略差异很大经常出现的情况是Flume配置没问题但节点间安全组或防火墙没放行端口。有些环境还涉及内外网卡路由问题node-01通过内网IP能ping通node-02但指定端口就是不通。这时候用nc来测nc -vz node-02 4141测完端口通还不放心的话直接在node-02上抓包看有没有SYN包到达tcpdump -nn -i eth0 tcp port 4141抓包是终极大法能看到包在传输但连接建立不了基本可以判定是防火墙策略丢包或拦截。我之前就遇到过云主机安全组明明放行了端口但底层网络ACL还在拦截的情况不抓包真的排查不出来。4.3 event重复或丢失Avro传输本身有事务机制理论上不会丢数据。但如果你在channel之间混用了Memory Channel一旦agent异常重启未落盘的那部分事件就会消失。反过来重复数据更多是出现在TaildirSource的positionFile和实际文件系统状态不同步的时候。遇到重复数据我的一般处理方法是停掉node-01的agent记录下当前positionFile的位置然后把业务日志文件复制一份到备份目录再清空日志文件等agent重新启动观察新写入的数据是否正常。如果接着出现大量重复就直接删除positionFile让source重新建立索引——不过这个操作要谨慎因为会让所有文件从头读一遍产生海量重复数据一般只在数据量小且能接受补采的情况下用。4.4 多个源端汇聚到同一汇聚端时的资源竞争当源agent数量增多都往node-02的同一个Avro Source端口推数据时可能会出现连接数过多导致网络吞吐性能下降的情况。这时的优化手段是破开单端口限制在node-02上配置多个Avro Source监听不同端口不同源agent根据自己的业务线走对应端口agent2.sources src_1 src_2 agent2.sources.src_1.type avro agent2.sources.src_1.bind 0.0.0.0 agent2.sources.src_1.port 4141 agent2.sources.src_1.channels ch2 agent2.sources.src_2.type avro agent2.sources.src_2.bind 0.0.0.0 agent2.sources.src_2.port 4142 agent2.sources.src_2.channels ch2对应地不同源agent的sink分别指定不同的端口。这样一来多个连接可以被分发到不同的Netty工作线程组降低单端口线程队列的争抢。物理上还可以配合多网卡绑定的方式让流量分开走。这个优化在我经历的项目里效果很明显——原先20个源端怼一个端口经常出现500ms以上的波动拆成3个Channel后平滑了很多。5. 链路性能调优与后续扩展方向串联架构跑通只是第一步让它稳定、高效地支撑起业务日志量才是真正考验。前面我们把channel容量、batchSize等都配置了初版但生产上还需要做几项调优。首先是sink的batch-size和网络吞吐的关系。Avro Sink内的event会按batch-size打包后发送batch越大单次网络IO传输的数据量越大、效率越高但batch过大会导致内存占用升高、单个批次失败重试的开销也变大。通常按单条日志体积来定——如果日志单条平均1KB左右500的batch大约500KB是一个比较合适的值单条日志体积大batch可以适当调低到200防止单个批次超过网络包的最大传输单元导致分片过多。然后是Source端的interceptor使用。在很多业务场景中我们不仅要把日志运过去还要做字段清洗、时间戳规范化、甚至简单的脱敏。Flume的interceptor机制正好干这个。在node-01的配置里加上agent1.sources.src1.interceptors i1 agent1.sources.src1.interceptors.i1.type regex_filter agent1.sources.src1.interceptors.i1.regex \\[INFO\\] agent1.sources.src1.interceptors.i1.excludeEvents false注意interceptor是在source端、写入channel之前执行的。这样做的好处是减少无效日志对channel空间的占用。但如果你对日志还需要保留原始审计数据那拦截器就得慎用否则审计数据缺失会很麻烦。再往后说扩展方向如果后续链路需要分发给多个下游比如既写HDFS做离线分析又要给实时计算平台供数可以在汇聚端配置sink group实现sink级负载均衡或故障转移agent2.sinks hdfsSink kafkaSink agent2.sinkgroups sg1 agent2.sinkgroups.sg1.sinks hdfsSink kafkaSink agent2.sinkgroups.sg1.processor.type load_balance agent2.sinkgroups.sg1.processor.backoff true agent2.sinkgroups.sg1.processor.selector ROUND_ROBIN这里需要说句实话load_balance模式下如果某个sink挂了而其他sink正常processor会在backoff控制下尝试恢复有一定容错性。但如果两个sink里有一个长期处于不可用状态链路整体的吞吐会被拖慢因为processor会不断尝试失败的sink。所以如果两侧下游稳定性差异很大我反而更倾向于用failover模式把可靠性更高的那个sink设成优先处理另一个作为冷备。这一块没有绝对最优完全看你的下游系统谁更可靠。从我个人的实际经验来说Flume的跨服务器串联链路最忌讳一步到位搭得特别大组件一多排查困难。先跑通最小闭环确认每一段都正常再去加组件、加拦截器、加负载均衡。与其花时间研究各种炫酷架构不如先把业务日志原原本本、不重不漏地运到目的地。这个基础打牢了后面再谈数据清洗、数据分发的优化才有意义。