flume拓扑结构

Flume 的三种拓扑结构，你的场景该用哪一种？

单机跑 Flume 很简单，一个 Agent、三个组件、一条链路，配置就能跑。

但生产环境往往更复杂：几十台服务器的日志要汇总、一份数据要同时给多个下游用、不同机房的采集链路要打通。这时候就需要考虑拓扑结构了。

Flume 官网定义了三种核心拓扑：简单串联、复制与多路复用、聚合。这篇文章我讲清楚每种结构是干什么的、什么场景用，以及实际选型的时候怎么判断。

简单串联

最简单的拓扑，就是一个 Agent 接一个 Agent，数据顺着链往下传。

典型场景：

• 跨网络传输：边缘机房到中心机房，中间隔了防火墙，没法直连存储，得经过一个跳板机
• 分步处理：Agent1 只做采集，Agent2 做数据清洗，Agent3 做格式转换再入库，每个 Agent 各管一段

配置示例（三个 Agent 三份配置）：

Agent1 在采集机器上，读本地文件，Sink 用 avro 发出去：

Agent1（数据源采集）

# Agent1：从文件采集数据，发送到 Agent2 的 Avro Source  
agent1.sources = execSource  
agent1.channels = memoryChannel  
agent1.sinks = avroSink  

# Source：监控本地文件  
agent1.sources.execSource.type = exec  
agent1.sources.execSource.command = tail -F /var/log/app.log  

# Sink：发送到 Agent2 的 Avro 端口（如 41414）  
agent1.sinks.avroSink.type = avro  
agent1.sinks.avroSink.hostname = agent2.example.com  
agent1.sinks.avroSink.port = 41414  

# 绑定关系  
agent1.sources.execSource.channels = memoryChannel  
agent1.sinks.avroSink.channel = memoryChannel  

Agent2（中间转发）

Agent2 在中间机器上，Source 用 avro 监听 41414，Sink 再发到 Agent3 的 41415：

# Agent2：接收 Agent1 数据，转发到 Agent3  
agent2.sources = avroSource  
agent2.channels = memoryChannel  
agent2.sinks = avroSink  

# Source：监听 Avro 端口 41414  
agent2.sources.avroSource.type = avro  
agent2.sources.avroSource.bind = 0.0.0.0  
agent2.sources.avroSource.port = 41414  

# Sink：转发到 Agent3 的 Avro 端口 41415  
agent2.sinks.avroSink.type = avro  
agent2.sinks.avroSink.hostname = agent3.example.com  
agent2.sinks.avroSink.port = 41415  

# 绑定关系  
agent2.sources.avroSource.channels = memoryChannel  
agent2.sinks.avroSink.channel = memoryChannel  

Agent3（目标存储）

Agent3 在存储机器上，Source 监听 41415，Sink 写 HDFS。

# Agent3：接收 Agent2 数据，写入 HDFS  
agent3.sources = avroSource  
agent3.channels = fileChannel  
agent3.sinks = hdfsSink  

# Source：监听 Avro 端口 41415  
agent3.sources.avroSource.type = avro  
agent3.sources.avroSource.bind = 0.0.0.0  
agent3.sources.avroSource.port = 41415  

# Sink：写入 HDFS  
agent3.sinks.hdfsSink.type = hdfs  
agent3.sinks.hdfsSink.hdfs.path = hdfs://cluster/flume/logs/%Y%m%d  

# 绑定关系  
agent3.sources.avroSource.channels = fileChannel  
agent3.sinks.hdfsSink.channel = fileChannel  

用了之后发现的问题：

• 单点故障。链上任何一个 Agent 挂了，整条链断掉。我在线上碰到过一次，中间的 Agent 因为 GC 卡住了，下游全部断粮。
• 延迟累积。每经过一个 Agent 就多一次序列化/反序列化、多一次网络传输，端到端延迟明显增加。
• 调试麻烦。数据在链上哪一段出了问题，得逐级查日志。

啥时候用它： 不得已才用。能用直接链路就别串联，非要串的话，每级都用 File Channel，别在中间环节丢数据。

复制和多路复用：一份数据多条路

一个 Source 对接多个 Channel 和 Sink，数据要么复制给所有下游，要么按条件分流。

复制模式（Replication）：同一份数据，发给所有 Sink。场景：一份日志同时存 HDFS 做归档、写 Kafka 做实时消费。

多路复用模式（Multiplexing）：按条件分发，不同数据走不同 Sink。场景：用户行为日志存 HDFS、系统监控日志存时序数据库，按日志类型分流。

配置示例

1. 复制模式（同一份数据多目标存储）

复制模式其实什么都不用配，Source 后面跟多个 Channel 默认就是复制：

# Agent：将数据同时写入 HDFS 和 Kafka  
agent.sources = tailSource  
agent.channels = hdfsChannel kafkaChannel  
agent.sinks = hdfsSink kafkaSink  

# Source：监控日志文件  
agent.sources.tailSource.type = exec  
agent.sources.tailSource.command = tail -F /var/log/app.log  
# 复制模式：数据发送到所有 Channel  
agent.sources.tailSource.channels = hdfsChannel kafkaChannel  

# Sink1：写入 HDFS  
agent.sinks.hdfsSink.type = hdfs  
agent.sinks.hdfsSink.hdfs.path = hdfs://cluster/logs/  
agent.sinks.hdfsSink.channel = hdfsChannel  

# Sink2：写入 Kafka  
agent.sinks.kafkaSink.type = org.apache.flume.sink.kafka.KafkaSink  
agent.sinks.kafkaSink.kafka.topic = app-logs  
agent.sinks.kafkaSink.channel = kafkaChannel  

# 配置 Channels  
agent.channels.hdfsChannel.type = file  
agent.channels.kafkaChannel.type = memory  

2. 多路复用模式（按条件路由）

多路复用需要配 Channel Selector：

结合自定义拦截器添加 Header，按 log_type 字段路由到不同 Sink：

# Agent：按日志类型路由到 HDFS 或 Kafka  
agent.sources = tailSource  
agent.channels = hdfsChannel kafkaChannel  
agent.sinks = hdfsSink kafkaSink  

# Source：配置拦截器添加 log_type 头信息  
agent.sources.tailSource.type = exec  
agent.sources.tailSource.command = tail -F /var/log/app.log  
agent.sources.tailSource.interceptors = typeInterceptor  
agent.sources.tailSource.interceptors.typeInterceptor.type = com.example.TypeInterceptor$Builder  

# 多路复用：按 Header 中的 log_type 路由  
agent.sources.tailSource.selector.type = multiplexing  
agent.sources.tailSource.selector.header = log_type  # 路由依据的 Header 字段  
agent.sources.tailSource.selector.mapping.user = hdfsChannel  # log_type=user → HDFS  
agent.sources.tailSource.selector.mapping.system = kafkaChannel  # log_type=system → Kafka  

# Sink 与 Channel 绑定（同复制模式）  
# ...（省略 HDFS Sink 和 Kafka Sink 配置）  

用了之后发现的问题：

• 资源消耗翻倍。两个 Sink 同时写，CPU 和网络带宽都要翻倍。我之前在一个服务器上配了三个 Sink，直接跑满了 I/O。
• 慢 Sink 拖慢全局。如果 HDFS Sink 写入慢，Kafka Sink 也得等，因为事务是绑在一起的。解决办法是把慢的 Sink 放到独立的 Sink Group 里，用不同的 Channel 隔离。

啥时候用它： 多下游消费同一份数据的场景，避免重复采集。

聚合：多对一，汇总到中心

多个边缘 Agent 采集数据，全部发到中心 Agent，由中心统一写入存储。

这是大数据集群日志采集的标配。几十台甚至上百台服务器的日志，每台跑一个 Flume Agent，都往中心 Agent 发，中心 Agent 攒够了往 HDFS 写。

配置示例

以 “3 个边缘 Agent 采集日志 → 1 个中心 Agent 聚合写入 HDFS” 为例：

边缘 Agent（如 Agent1）

# 边缘 Agent1：采集本地日志，发送到中心 Agent  
agent1.sources = execSource  
agent1.channels = memoryChannel  
agent1.sinks = avroSink  

# Source：监控本地日志  
agent1.sources.execSource.type = exec  
agent1.sources.execSource.command = tail -F /var/log/server1.log  

# Sink：发送到中心 Agent 的 Avro 端口  
agent1.sinks.avroSink.type = avro  
agent1.sinks.avroSink.hostname = central-agent.example.com  
agent1.sinks.avroSink.port = 41414  

# 绑定关系  
agent1.sources.execSource.channels = memoryChannel  
agent1.sinks.avroSink.channel = memoryChannel  

中心 Agent（聚合写入 HDFS）

# 中心 Agent：接收多个边缘 Agent 数据，写入 HDFS  
central.sources = avroSource  
central.channels = fileChannel  
central.sinks = hdfsSink  

# Source：监听 Avro 端口，接收所有边缘 Agent 数据  
central.sources.avroSource.type = avro  
central.sources.avroSource.bind = 0.0.0.0  
central.sources.avroSource.port = 41414  
# 支持高并发：增加工作线程数  
central.sources.avroSource.threads = 20  

# Sink：聚合写入 HDFS  
central.sinks.hdfsSink.type = hdfs  
central.sinks.hdfsSink.hdfs.path = hdfs://cluster/aggregated-logs/%Y%m%d/  
central.sinks.hdfsSink.hdfs.filePrefix = aggregated-  

# 通道：使用 File Channel 确保可靠性  
central.channels.fileChannel.type = file  
central.channels.fileChannel.checkpointDir = /var/flume/checkpoint  
central.channels.fileChannel.dataDirs = /var/flume/data  

# 绑定关系  
central.sources.avroSource.channels = fileChannel  
central.sinks.hdfsSink.channel = fileChannel  

用了之后发现的问题：

• 中心 Agent 是瓶颈。 所有数据汇聚到一点，单机带宽、磁盘 I/O、CPU 都可能被撑爆。我之前在中心 Agent 上遇到过网络带宽打满，导致边缘 Agent 全部积压。
• 中心 Agent 挂了全完。 边缘 Agent 的数据发不出去，Channel 积压，积满了就丢数据。

解决方案：

• 中心 Agent 部署多个实例，前面加一层负载均衡（DNS 轮询或者 Nginx 转发）
• 边缘 Agent 的 Channel 容量要大，扛得住中心短暂不可用
• 中心 Agent 的 Source 线程数调高（threads=20 甚至更高）

啥时候用它： 数据源分散、需要集中管理的场景。几乎所有的生产环境日志采集都在用这个模式

拓扑结构对比与选择建议

拓扑结构	核心优势	局限性	最佳实践场景
简单串联	配置简单，支持分步处理	单点故障风险高，延迟累积	跨网络传输、分步清洗链路
复制 / 多路复用	数据多目标分发，灵活路由	资源消耗高，需平衡各 Sink 性能	数据多副本存储、按类型分类处理
聚合	分布式数据源集中管理	中心 Agent 可能成瓶颈，需扩容	大规模集群日志采集、区域数据汇总