4、日志传输与缓冲机制：Kafka 在日志链路中的角色

日志系统里，Kafka 到底扮演什么角色？

说白了，它就是个「超级缓冲池」。

我刚开始接触日志系统时，总觉得直接写文件不就完了？后来线上一次日志洪峰，直接把后端服务打挂了。嗯，从那以后我才真正理解——没有缓冲的日志系统，就是定时炸弹。

4.1 Kafka 在日志链路中的核心定位

日志从产生到落地，通常要经过：

  • 采集端（如 Filebeat、Logstash）
  • 缓冲层（Kafka）
  • 消费端（如 Logstash、Flink、ES）

Kafka 夹在中间，干三件事：

  1. 削峰填谷——采集端突发的日志量，不会直接压垮下游
  2. 解耦——采集和消费可以各自独立扩缩容
  3. 持久化——日志不会因为消费端挂了就丢失

我个人习惯：只要日志量超过 1000 TPS，就必须上 Kafka。低于这个量，用 Redis 队列也能凑合，但一旦上了规模，Kafka 的吞吐优势就体现出来了。

4.2 生产者/消费者模型：谁在发，谁在收？

Kafka 的模型其实很简单：

  • 生产者（Producer）：把日志消息发到 Kafka 的 Topic 里
  • 消费者（Consumer）：从 Topic 里拉取消息，做后续处理
  • Topic：消息的分类，相当于一个「日志桶」
  • Partition：Topic 的分片，用来并行处理

我画了一张图，帮你理清关系：

生产者 Kafka 集群 Topic: app-logs Part-0 Part-1 Part-2 消费者组 A 消费者组 B 消费者组 C 图例： 生产者 Topic/Partition 消费者组 数据流向

你想想看，一个 Topic 有 3 个 Partition，消费者组里有 3 个消费者，那正好一人消费一个 Partition。如果消费者只有 2 个，那其中一个就要扛两个 Partition 的活。

避坑指南：我曾经在生产环境遇到过消费者数量超过 Partition 数的情况，结果多出来的消费者一直空闲，白白浪费资源。所以消费者数量 ≤ Partition 数量，这是铁律。

4.3 数据持久化：日志不会丢的秘密

Kafka 的持久化机制，我总结为三个关键词：

机制 说明 我的经验
磁盘顺序写 消息追加到文件末尾，不随机读写 吞吐量能到百万级/秒
分段存储 每个 Partition 分成多个 Segment 文件 方便清理过期数据
索引文件 每个 Segment 配一个索引，快速定位 查历史日志时很有用

具体来说，Kafka 的消息默认存在磁盘上，保留 7 天（可配置）。这意味着：

  • 消费者挂了，重启后还能从上次的位置继续消费
  • 多个消费者组可以各自维护自己的消费进度
  • 日志不会因为消费端重启而丢失

我个人习惯：日志场景下，我一般设置保留 3 天。太长了浪费磁盘，太短了排查问题时找不到历史数据。3 天是个平衡点。

4.4 重试策略：消息发送失败了怎么办？

生产者发送消息时，可能会遇到网络抖动、Broker 宕机等情况。Kafka 提供了三种重试策略：

4.4.1 同步重试

发送失败后立即重试，直到成功或达到最大重试次数。

// 生产者配置示例
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");  // 等待所有副本确认
props.put("retries", 3);   // 重试次数
props.put("max.in.flight.requests.per.connection", 1);  // 防止乱序

4.4.2 异步重试 + 回调

不阻塞主流程，失败时通过回调处理。

producer.send(new ProducerRecord<>("app-logs", key, value), 
    (metadata, exception) -> {
        if (exception != null) {
            // 记录失败日志，后续补偿
            log.error("发送失败，消息ID: {}", key);
            // 可以写入死信队列
        }
    });

4.4.3 幂等性保证

开启幂等性后，Kafka 会自动去重，避免重复消息。

props.put("enable.idempotence", true);  // 开启幂等性

注意：我曾经遇到过一个坑——开启了幂等性，但没设置 max.in.flight.requests.per.connection=1，结果在高并发下还是出现了少量重复消息。后来查文档才发现，幂等性需要配合这个参数才能完全生效。

4.5 日志链路中的完整配置示例

下面是我在项目中常用的日志传输配置，你可以直接参考：

# 生产者端（Filebeat 或 Logstash 输出到 Kafka）
output.kafka:
  hosts: ["kafka1:9092", "kafka2:9092", "kafka3:9092"]
  topic: "app-logs"
  partition.round_robin:
    reachable_only: true
  compression: gzip
  max_message_bytes: 10485760  # 10MB
  required_acks: 1
  retries: 5

# 消费者端（Logstash 从 Kafka 消费）
input {
  kafka {
    bootstrap_servers => "kafka1:9092,kafka2:9092,kafka3:9092"
    topics => ["app-logs"]
    group_id => "logstash-consumer-group"
    auto_offset_reset => "latest"
    consumer_threads => 3
    fetch_max_bytes => 52428800  # 50MB
  }
}

小技巧：auto_offset_reset 设置为 latest，表示只消费新消息。如果你要回溯历史日志，可以临时改成 earliest，但记得改回来，否则会重复消费大量数据。

4.6 测试验证要点

自动化测试时，我重点关注这几个点：

  1. 消息不丢：生产者发送 10 万条，消费者收到 10 万条
  2. 消息不乱：同一个 Partition 内的消息顺序一致
  3. 故障恢复：杀掉一个 Broker，生产和消费是否正常
  4. 性能压测：持续写入 30 分钟，观察吞吐量和延迟

嗯，Kafka 在日志链路中的角色，说白了就是「可靠的中转站」。它不负责日志的解析和存储，但保证了日志在传输过程中不丢、不乱、不堵。下一节我们会聊到日志的格式化与结构化，到时候你会看到，Kafka 传过来的原始日志，是怎么变成我们熟悉的 JSON 格式的。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →