📚 课程目录第3章数据采集层设计

3. 数据采集层设计：日志采集、数据库CDC与消息队列选型

数据采集层，说白了就是实时数据产品的「水源地」。水源不干净，下游再好的管道、再强的处理能力都白搭。我这些年踩过的坑，十有八九都跟采集层有关——要么丢数据，要么重复采集，要么延迟高到离谱。

这一章，咱们就聊聊日志采集、数据库CDC和消息队列选型。这三个东西，是实时数据采集的三大支柱。

3.1 日志采集：Filebeat vs Flume

日志采集，最传统的场景就是采集服务器上的文本日志。你想想看，一个中型互联网公司，每天产生几十TB的日志，怎么搞？

我个人的习惯是：轻量级场景用Filebeat，复杂管道用Flume。为什么这么分？

对比维度	Filebeat	Flume
资源消耗	极低（Go编写，约10-20MB内存）	较高（Java，约200-500MB）
部署复杂度	简单，单二进制文件	需要Java环境，配置较复杂
数据处理能力	弱，仅支持简单过滤和字段提取	强，支持拦截器、路由、聚合
可靠性保障	内置背压机制，支持ACK	事务机制，端到端可靠性
适用场景	日志采集→Elasticsearch/Kafka	复杂ETL、多路分发、数据清洗

我的经验： 如果只是把Nginx日志送到Kafka，Filebeat就够了。但如果你需要在采集层做字段解析、根据日志级别分流、或者做简单的聚合统计，Flume的拦截器体系会让你省很多事。

Filebeat的配置，其实很简单。我贴一个常用的配置片段：

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/nginx/access.log
  fields:
    log_type: nginx_access
    env: production

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

嗯，这里要注意：required_acks 我一般设成1，兼顾可靠性和性能。设成-1（all）虽然更安全，但吞吐会下降30%左右。

Flume的配置就复杂一些。我曾经在一个项目里，用Flume同时采集2000台服务器的日志，还要做字段脱敏和路由分发。那配置写起来，真是一把辛酸泪。

agent.sources = tailSource
agent.channels = kafkaChannel
agent.sinks = kafkaSink

agent.sources.tailSource.type = spooldir
agent.sources.tailSource.spoolDir = /var/log/app
agent.sources.tailSource.fileHeader = true
agent.sources.tailSource.interceptors = filterInterceptor

agent.sources.tailSource.interceptors.filterInterceptor.type = regex_filter
agent.sources.tailSource.interceptors.filterInterceptor.regex = ^ERROR|WARN
agent.sources.tailSource.interceptors.filterInterceptor.excludeEvents = false

agent.channels.kafkaChannel.type = org.apache.flume.channel.kafka.KafkaChannel
agent.channels.kafkaChannel.kafka.bootstrap.servers = kafka1:9092,kafka2:9092
agent.channels.kafkaChannel.kafka.topic = app-log-topic
agent.channels.kafkaChannel.capacity = 10000
agent.channels.kafkaChannel.transactionCapacity = 1000

agent.sinks.kafkaSink.type = org.apache.flume.sink.kafka.KafkaSink
agent.sinks.kafkaSink.kafka.topic = app-log-topic
agent.sinks.kafkaSink.kafka.bootstrap.servers = kafka1:9092,kafka2:9092

避坑指南： 我曾经在生产环境遇到过Flume的KafkaChannel和KafkaSink同时使用同一个topic的情况，导致数据循环写入。后来我强制要求：Source→Channel→Sink的topic必须不同，或者Channel用File Channel做缓冲。

3.2 数据库CDC：Canal vs Debezium

数据库CDC（Change Data Capture），是实时数据产品的另一个重要水源。说白了，就是监听数据库的binlog，把数据变更实时同步出来。

我个人建议：MySQL生态用Canal，异构数据库用Debezium。为什么？

Canal：阿里巴巴开源，专为MySQL设计。性能极好，单节点可以支撑数千TPS的binlog解析。但只支持MySQL。
Debezium：基于Kafka Connect，支持MySQL、PostgreSQL、MongoDB、Oracle等。生态丰富，但性能不如Canal极致。

我记得有一次，客户要求从MySQL同步数据到HBase，延迟要小于1秒。我们用Canal，单节点扛住了8000 TPS的写入，延迟稳定在200ms以内。如果用Debezium，同样的场景，延迟会到500ms左右。

但Debezium也有它的优势。比如，你要从PostgreSQL同步数据，Canal就无能为力了。Debezium的配置，我贴一个MySQL Connector的例子：

{
  "name": "mysql-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "192.168.1.100",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "debezium_pwd",
    "database.server.id": "223344",
    "database.server.name": "my-app-db",
    "database.include.list": "order_db,user_db",
    "table.include.list": "order_db.orders,user_db.users",
    "database.history.kafka.bootstrap.servers": "kafka1:9092,kafka2:9092",
    "database.history.kafka.topic": "schema-changes.my-app-db",
    "include.schema.changes": "true",
    "snapshot.mode": "initial",
    "tombstones.on.delete": "false"
  }
}

  关键点： snapshot.mode 设成 initial，第一次启动时会做全量快照，之后只增量同步。如果你只想增量，可以设成 schema_only。但要注意，schema_only 不会同步已有数据，只同步启动后的变更。

Canal的配置相对简单，但要注意权限问题。我曾经因为binlog权限没配好，排查了整整一天。

# canal.properties
canal.serverMode = tcp
canal.destinations = example

# instance.properties
canal.instance.master.address = 192.168.1.100:3306
canal.instance.dbUsername = canal
canal.instance.dbPassword = canal_pwd
canal.instance.connectionCharset = UTF-8
canal.instance.tsdb.enable = true
canal.instance.gtidon = false

# 监听哪些表
canal.instance.filter.regex = order_db\\..*,user_db\\..*
# 排除哪些表
canal.instance.filter.black.regex = order_db\\.tmp_.*

避坑指南： 我曾经在生产环境遇到过Canal重复消费的问题。原因是Canal的ZooKeeper集群时钟不同步，导致offset写入混乱。后来我强制要求：所有Canal节点和ZooKeeper节点必须做NTP时钟同步，并且使用独立ZooKeeper集群，不要跟其他服务混用。

3.3 消息队列选型：Kafka vs Pulsar

消息队列，是数据采集层的「主动脉」。选对了，事半功倍；选错了，后面全是坑。

我个人的经验是：Kafka适合吞吐优先、架构简单的场景；Pulsar适合多租户、延迟敏感、需要存储计算分离的场景。

对比维度	Kafka	Pulsar
架构模型	存储计算耦合（Broker既存数据又处理请求）	存储计算分离（BookKeeper存数据，Broker处理请求）
吞吐性能	极高（单节点百万级消息/秒）	高（单节点数十万级消息/秒）
延迟	毫秒级（但分区数多时延迟会上升）	亚毫秒级（分区数多时延迟稳定）
多租户	弱（靠Topic命名空间隔离）	强（原生支持Tenant/Namespace）
消息保留	基于时间/大小删除	支持TTL、保留策略、积压策略
运维复杂度	中等（依赖ZooKeeper）	较高（依赖BookKeeper和ZooKeeper）

你想想看，如果你的业务是日志采集，每天几十TB的数据量，Kafka几乎是唯一选择。我见过一个场景，用3台Kafka Broker，每台8块SSD，扛住了每天100TB的日志写入，延迟稳定在50ms以内。

但如果你做的是金融交易系统，对延迟和隔离性要求极高，Pulsar的优势就出来了。Pulsar的存储计算分离架构，意味着你可以独立扩缩Broker和BookKeeper节点。我曾经帮一个客户做Pulsar集群，Broker从3台扩到10台，BookKeeper从3台扩到6台，整个过程零停机，业务无感知。

Kafka的生产者配置，我一般这样写：

Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092,kafka2:9092,kafka3:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("acks", "1");
props.put("retries", "3");
props.put("batch.size", "16384");
props.put("linger.ms", "10");
props.put("buffer.memory", "33554432");
props.put("compression.type", "snappy");

Producer producer = new KafkaProducer<>(props);

我的经验： linger.ms 设成10ms，batch.size 设成16KB，可以在延迟和吞吐之间取得很好的平衡。如果追求极致吞吐，可以把 linger.ms 提高到100ms，但延迟会相应增加。

Pulsar的生产者配置，稍微不同：

PulsarClient client = PulsarClient.builder()
    .serviceUrl("pulsar://pulsar1:6650,pulsar2:6650,pulsar3:6650")
    .build();

Producer producer = client.newProducer(Schema.STRING)
    .topic("persistent://my-tenant/my-namespace/my-topic")
    .producerName("my-producer")
    .sendTimeout(30, TimeUnit.SECONDS)
    .blockIfQueueFull(true)
    .maxPendingMessages(1000)
    .compressionType(CompressionType.LZ4)
    .create();

避坑指南： 我曾经在生产环境遇到过Pulsar生产者 blockIfQueueFull 设成false导致消息丢失的问题。当生产者队列满了，新消息会被直接丢弃。后来我强制要求：所有生产环境必须设成true，并且配合 maxPendingMessages 控制队列大小，避免OOM。

3.4 整体架构图

说了这么多，咱们用一张图把整个数据采集层的架构串起来。这张图是我在多个项目中沉淀下来的通用模式：

这张图的核心逻辑，说白了就是：数据源→采集器→消息队列→消费者。每一层都可以独立扩展，互不依赖。我见过很多团队，把采集器和消息队列混在一起，结果出了问题，排查半天都找不到根因。

  总结一下：
  日志采集：轻量用Filebeat，复杂用Flume
数据库CDC：MySQL用Canal，异构用Debezium
消息队列：吞吐优先用Kafka，多租户/低延迟用Pulsar
架构原则：每层独立，无状态设计，幂等消费

嗯，数据采集层就聊到这儿。下一层，咱们聊聊数据管道中的「清洗与转换」——那又是另一番天地了。