Spring Boot + Kafka 亿级日志收集系统：零丢失传输与实时分析架构

关键词：Spring Boot、Kafka、亿级日志、零丢失、实时分析、Exactly-Once、背压、熔断、可观测性

一、总体设计思路

二、整体架构（Logical View）

┌─────────────┐     ┌─────────────┐     ┌──────────────┐
│ Filebeat    │────▶│  Kafka      │────▶│  Flink SQL    │
│   / SDK     │     │  Cluster    │     │  /Flink Job   │
└─────────────┘     └─────────────┘     └──────────────┘
       ▲                    │                     │
       │        ┌───────────┴───────────┐         │
       │        │  Spring Boot LogHub   │         │
       │        │  (Producer + Admin)   │         │
       └────────┴───────────────────────┘         ▼
                                        ┌────────────────┐
                                        │ ClickHouse/ES  │
                                        │   Dashboard    │
                                        └────────────────┘

三、日志采集端实现

3.1 轻量级 Java SDK（零侵入）

// 依赖：spring-kafka 3.x
@Component
public final class LogKafkaProducer {

    private final KafkaTemplate<String, byte[]> kafkaTemplate;

    @Value("${log.topic}")
    private String topic;

    // 使用 ByteArraySerializer，减少一次序列化
    public void send(LogEvent event) {
        ProducerRecord<String, byte[]> record =
                new ProducerRecord<>(topic, null, event.getTimestamp(),
                                     event.getTraceId(), ProtostuffUtils.serialize(event));
        // 异步回调+重试
        kafkaTemplate.send(record).addCallback(
            r -> {},
            ex -> log.error("send failed, will retry", ex));
    }
}

3.2 幂等与事务配置

spring:
  kafka:
    producer:
      bootstrap-servers: kfk1:9092,kfk2:9092,kfk3:9092
      retries: 5
      acks: all
      enable-idempotence: true
      transaction-id-prefix: log-tx-
      compression-type: lz4
      linger-ms: 5
      batch-size: 64KB

四、Kafka 集群可靠性配置

五、实时处理：Spring Boot Stream → Flink

5.1 Spring Boot 消费者（快速过滤）

@KafkaListener(id = "alarm-consumer", topics = "log.error",
               concurrency = "8")
public void listen(ConsumerRecord<String, byte[]> rec,
                   Acknowledgment ack) {
    LogEvent e = ProtostuffUtils.deserialize(rec.value(), LogEvent.class);
    if (e.getLevel() == Level.ERROR) {
        alertService.push(e);
    }
    ack.acknowledge();
}

5.2 Flink SQL 实时聚合（核心链路）

CREATE TABLE log_source (
  traceId STRING,
  `timestamp` TIMESTAMP_LTZ(3),
  level STRING,
  host STRING,
  message STRING,
  -- 使用 Kafka 连接器
  WATERMARK FOR `timestamp` AS `timestamp` - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'log.all',
  'properties.bootstrap.servers' = 'kfk1:9092',
  'format' = 'json',
  'scan.startup.mode' = 'latest-offset'
);

-- 每 30 秒统计 ERROR 日志 Top 10 主机
SELECT host, COUNT(*) AS err_cnt
FROM log_source
WHERE level = 'ERROR'
GROUP BY host, TUMBLE(`timestamp`, INTERVAL '30' SECOND)
ORDER BY err_cnt DESC
LIMIT 10;

六、零丢失机制深度剖析

七、可观测与报警

7.1 指标采集（Micrometer → Prometheus）

@Configuration
public class KafkaMetrics {
    @Bean
    public MeterRegistryCustomizer<PrometheusMeterRegistry> kafkaMetrics() {
        return registry -> {
            // 绑定 Kafka Producer JMX
            new KafkaClientMetrics(kafkaTemplate.getProducerFactory().getConfigurationProperties())
                .bindTo(registry);
        };
    }
}

7.2 关键报警规则（PromQL）

# ISR 副本不足
- alert: KafkaISRShrink
  expr: kafka_cluster_partition_in_sync_replica < 2
  for: 1m

# 消费延迟 > 5s
- alert: ConsumerLag
  expr: kafka_consumer_lag_sum > 5000

八、亿级场景压测结果

九、常见坑与最佳实践

1. 分区数 = 目标吞吐(MB/s) ÷ 单分区吞吐(≈10 MB/s) 再向上取整。
2. 避免大消息：单条 > 1 MB 时开启 compression.type=lz4。
3. 避免长 GC：-XX:+UseZGC + 512 k 堆外 buffer。
4. 监控 > 调优：先可观测再优化。

十、一键启动 Demo

git clone https://github.com/your-org/spring-kafka-log.git
cd docker && docker-compose up -d
# 访问 http://localhost:8080 生成日志
# Grafana: http://localhost:3000 (admin/admin)