4、数据采集：日志采集（Filebeat/Flume）、消息队列（Kafka）的选型与配置

数据采集是整个实时指标链路的第一道关卡。说白了，如果源头的数据都采不全、采不准，后面再牛的流计算引擎也是白搭。这一章我们聊聊日志采集工具和消息队列的选型与配置，我会结合自己踩过的坑，给你一些实在的建议。

4.1 日志采集器：Filebeat vs Flume

先说说日志采集。市面上主流的就是 Filebeat 和 Flume。很多人纠结选哪个，其实没那么复杂——看场景。

4.1.1 Filebeat：轻量、简单、Go 语言编写

Filebeat 是 Elastic 家的产品，用 Go 写的。它的特点就一个字：轻。占用资源极少，部署起来跟喝水一样简单。

• 资源占用： 我记得有一次在 512MB 的机器上跑 Filebeat，内存占用才 20MB 左右。Flume 的话，没个 512MB 起步你都不敢跑。
• 配置方式： 一个 YAML 文件搞定，没有复杂的依赖。
• 适用场景： 单纯的日志采集、转发到 Elasticsearch 或 Kafka。

我个人习惯把 Filebeat 当作“边车”部署在每台业务机器上。它负责监听日志文件，一行一行地读，然后发到 Kafka。

# filebeat.yml 核心配置示例
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/app/*.log
  fields:
    app_id: "order-service"
    env: "prod"

output.kafka:
  hosts: ["kafka1:9092", "kafka2:9092"]
  topic: "app-logs"
  partition.round_robin:
    reachable_only: true
  required_acks: 1
  compression: gzip
  max_message_bytes: 1000000

4.1.2 Flume：重量级、Java 生态、灵活

Flume 是 Apache 旗下的老牌日志采集框架。它比 Filebeat 重得多，但胜在灵活。它的架构是 Source → Channel → Sink，你可以像搭积木一样组合各种组件。

• 优势： 支持拦截器（Interceptor）、自定义 Source/Sink、事务保证。如果你需要从非文件源（比如 Taildir、Syslog、Kafka Source）采集，Flume 更合适。
• 劣势： 配置繁琐，依赖 Java 环境，内存占用高。
• 适用场景： 复杂的 ETL 场景、多数据源汇聚、需要事务性保证的采集。

我曾经在一个老项目里用过 Flume，当时需要从多个 Flume Agent 级联采集，还要做数据过滤和格式转换。Flume 的拦截器确实好用，但维护成本也高。后来新项目我全换成了 Filebeat + Kafka Connect，清爽多了。

# flume.conf 核心配置示例
agent.sources = tailSource
agent.channels = kafkaChannel
agent.sinks = kafkaSink

# 配置 Taildir Source（支持断点续传）
agent.sources.tailSource.type = spooldir
agent.sources.tailSource.spoolDir = /var/log/app/
agent.sources.tailSource.fileHeader = true
agent.sources.tailSource.interceptors = i1
agent.sources.tailSource.interceptors.i1.type = timestamp

# 配置 Channel（使用内存 Channel，注意容量）
agent.channels.kafkaChannel.type = memory
agent.channels.kafkaChannel.capacity = 10000
agent.channels.kafkaChannel.transactionCapacity = 1000

# 配置 Kafka Sink
agent.sinks.kafkaSink.type = org.apache.flume.sink.kafka.KafkaSink
agent.sinks.kafkaSink.kafka.topic = app-logs
agent.sinks.kafkaSink.kafka.bootstrap.servers = kafka1:9092,kafka2:9092
agent.sinks.kafkaSink.kafka.producer.acks = 1

4.1.3 选型对比

4.2 消息队列：Kafka 的选型与配置

日志采集上来之后，一般不会直接喂给流计算引擎，而是先放到消息队列里。为什么？因为要解耦。采集端和消费端的速度不可能完全匹配，中间加个队列，两边都能松口气。

消息队列有很多：RabbitMQ、RocketMQ、Pulsar……但实时指标领域，Kafka 是事实标准。原因很简单：高吞吐、持久化、分区并行。

4.2.1 Kafka 核心概念速览

• Topic： 消息的分类。比如“订单日志”一个 Topic，“用户行为”一个 Topic。
• Partition： Topic 的分片。每个 Partition 内部有序，Partition 之间无序。分区数决定了并行度。
• Consumer Group： 消费组。同一个组内的消费者共同消费一个 Topic，每个 Partition 只能被组内一个消费者消费。
• Offset： 偏移量。消费者消费的位置，Kafka 靠它实现断点续传。

4.2.2 选型要点

选 Kafka 版本和配置时，我一般关注这几个点：

1. 版本选择： 别追新。我建议用 2.8.x 或 3.x 的稳定版。3.0 之后去掉了 ZooKeeper 依赖（用 KRaft），但生产环境我还没完全放心，目前还是用 2.8 + ZooKeeper 的组合。
2. 分区数： 分区不是越多越好。每个分区对应一个日志文件，分区太多会导致文件句柄暴增、Leader 选举变慢。我一般按 Topic 的吞吐量估算：单分区吞吐约 10MB/s，目标吞吐除以 10 再留 20% 余量。
3. 副本数： 生产环境至少 2 副本，核心 Topic 建议 3 副本。副本多了写性能会下降，但数据安全更重要。
4. 消息保留策略： 实时指标场景，消息一般保留 24-72 小时。配置 log.retention.hours 或 log.retention.bytes。

# server.properties 核心配置
# 基础配置
broker.id=1
listeners=PLAINTEXT://0.0.0.0:9092
log.dirs=/data/kafka/logs

# 分区与副本
num.partitions=3
default.replication.factor=2
min.insync.replicas=1

# 消息保留
log.retention.hours=72
log.retention.bytes=1073741824  # 1GB

# 性能调优
num.network.threads=8
num.io.threads=16
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600

4.2.3 生产环境配置清单

4.3 整体链路架构

下面这张图展示了从日志产生到进入消息队列的完整链路。你可以看到，Filebeat 部署在业务机器上，采集日志后直接写入 Kafka。Kafka 作为缓冲层，下游的 Flink 或 Spark Streaming 再从 Kafka 消费。

4.4 避坑与最佳实践

最后，分享几个我在项目中踩过的坑，希望能帮你少走弯路。