4. 消息队列解耦：在 Logstash 与 Elasticsearch 之间引入 Kafka

在日志采集与检索的经典链路中，Logstash 直接向 Elasticsearch 写入数据会引发两个核心问题：背压传导与突发流量冲击。当 ES 集群因合并、GC 或网络抖动导致写入变慢时，Logstash 的输出线程会被阻塞，进而反压到输入插件，造成整个采集管道阻塞甚至 OOM。引入 Kafka 作为缓冲层，可以将 Logstash 与 ES 解耦，实现削峰填谷与异步处理，显著提升系统稳定性。

4.1 为什么需要消息队列解耦

直接耦合架构在流量突发或下游抖动时，故障会沿链路向上传播。下表对比了直接写入与引入 Kafka 后的关键差异：

4.2 引入 Kafka 后的组件拓扑

新的日志平台拓扑在 Logstash 与 Elasticsearch 之间插入 Kafka 集群，形成「生产者 → 缓冲 → 消费者」的三层架构。下图展示了核心组件与数据流向：

4.3 核心配置实现

以下配置展示了 Logstash 生产端将数据写入 Kafka，以及消费端从 Kafka 拉取数据写入 Elasticsearch 的完整流程。

4.3.1 生产端 Logstash 配置（input → kafka output）

# producer-logstash.conf
input {
  beats {
    port => 5044
    client_inactivity_timeout => 60
  }
}

filter {
  # 通用日志解析
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
  }
  # 添加元数据
  mutate {
    add_field => { "[@metadata][topic]" => "logstash-logs" }
  }
}

output {
  kafka {
    topic_id => "logstash-logs"
    bootstrap_servers => "kafka1:9092,kafka2:9092,kafka3:9092"
    codec => "json"
    # 批量发送，提升吞吐
    batch_size => 16384
    linger_ms => 100
    # 至少写入 1 个副本，保证不丢数据
    acks => "all"
    compression_type => "gzip"
  }
}

4.3.2 消费端 Logstash 配置（kafka input → elasticsearch output）

# consumer-logstash.conf
input {
  kafka {
    topics => ["logstash-logs"]
    bootstrap_servers => "kafka1:9092,kafka2:9092,kafka3:9092"
    group_id => "logstash-es-consumer"
    consumer_threads => 4
    # 从最新偏移量开始消费
    auto_offset_reset => "latest"
    # 批量拉取
    fetch_max_wait_ms => 500
    fetch_min_bytes => 524288  # 512KB
    codec => "json"
  }
}

filter {
  # 消费端可做二次处理，如字段裁剪、类型转换
  mutate {
    remove_field => ["@version", "host"]
  }
}

output {
  elasticsearch {
    hosts => ["http://es-node1:9200", "http://es-node2:9200"]
    index => "logs-%{+YYYY.MM.dd}"
    # 批量写入
    flush_size => 500
    idle_flush_time => 5
    # 失败重试
    retry_on_failure => true
    # 使用 ES 的 bulk API
    bulk_path => "/_bulk"
  }
}

4.4 削峰填谷效果验证

引入 Kafka 后，系统在突发流量下的表现差异显著。以下为压测数据对比（模拟 10 倍峰值流量）：

从数据可以看出，Kafka 作为缓冲层将突发流量「削峰」为平稳的写入流，ES 集群不再直接面对流量尖刺，整体稳定性大幅提升。

4.5 生产环境最佳实践

• Topic 分区规划：分区数 = max(Logstash 消费线程数, ES 数据节点数 × 2)。分区过多会增加 Kafka 元数据开销，过少则降低并行度。
• 消息保留策略：根据日志重放需求设置 retention.ms（建议 24~72 小时）。如果日志量极大，可启用 log.retention.bytes 按容量清理。
• 监控关键指标：Kafka 消费者 Lag（滞后量）、Logstash 输出队列深度、ES 索引速率。Lag 持续增长说明消费端能力不足，需扩容消费者或优化 ES 写入。
• 幂等写入：Kafka 生产端启用 enable.idempotence=true，避免网络重试导致的消息重复。消费端 Logstash 配合 ES 的 doc_as_upsert 实现去重。
• 压缩与批量：生产端开启 gzip 压缩（压缩比可达 5:1），消费端设置合理的 fetch_min_bytes 和 flush_size，减少网络交互次数。