Kafka

Kafka 是分布式流处理平台，用于高吞吐量数据管道、流分析和数据集成场景。核心特点：高吞吐量、持久化、分布式、可扩展。

核心特性

特性	说明
高吞吐量	即使存储 TB 级消息，仍保持稳定性能
持久化	消息存储在磁盘，集群内复制防止数据丢失
分布式	数据分散在多个 broker 节点
可扩展	支持在线扩展，无需停机

基本概念

概念	说明
Producer	生产者，负责发送消息到 topic
Consumer	消费者，从 topic 拉取消息
Broker	Kafka 节点
Topic	消息主题，用于分类消息
Partition	分区，topic 的物理分组，提高并行度
Replica	副本，分区数据的备份
Leader	主副本，处理所有读写请求
Follower	从副本，从 leader 同步数据
ISR	In-Sync Replicas，与 leader 保持同步的副本集合

Partition 数量

Partition 数量决定了消费者的最大并行度。建议设置为 broker 数量的 2-3 倍。

集群架构

生产者将消息发送到 topic，Kafka 将消息均匀分布在 partition
所有 replica 会选举一个 leader，生产和消费都跟 leader 打交道
写数据时，leader 负责将数据同步到所有 follower
读数据时，直接读 leader 上的数据即可

为什么只能读写 leader？

如果可以随意读写每个 follower，就要处理数据一致性问题，系统复杂度太高。

高可用机制

ISR（In-Sync Replicas）

ISR 是与 leader 保持同步的副本集合。只有 ISR 中的副本才有资格被选为新的 leader。

副本机制

配置参数	说明
`replication.factor`	每个 partition 的副本数，必须 > 1
`min.insync.replicas`	leader 感知到的最小 follower 数

消费者组

消费者组是一组共同消费同一 topic 的消费者：

Topic: test
  ├── Partition 0 ──→ Consumer A (Group 1)
  ├── Partition 1 ──→ Consumer B (Group 1)
  └── Partition 2 ──→ Consumer C (Group 1)

规则：

同一个 partition 只能被组内一个消费者消费
Partition 数量 >= 消费者数量（多余的消费者会闲置）

偏移量管理

概念	说明
Current Offset	消费者当前消费到的位置
Committed Offset	已提交给 broker 的偏移量
Lag	未消费消息数量

核心原则：关闭自动提交，手动提交偏移量

props.put("enable.auto.commit", false);

可靠性保证

生产者配置

Properties props = new Properties();
props.put("acks", "all");           // 必须所有副本同步成功
props.put("retries", Integer.MAX_VALUE);  // 无限重试
props.put("enable.idempotence", true);    // 启用幂等性

acks 参数：

值	说明	可靠性	性能
`0`	不等待	可能丢失	最快
`1`	只等 leader 确认	可能丢失	中等
`all`	等所有 ISR 副本确认	最高	最慢

事务语义

语义	说明	场景
At-most-once	最多一次，可能丢失	日志收集
At-least-once	最少一次，可能重复	账单计算
Exactly-once	精确一次	支付系统

顺序保证

范围	说明
单 Partition 内	消息按顺序存储和消费
跨 Partition	不保证顺序

按 Key 分区可保证同一 Key 的消息有序：

producer.send(new ProducerRecord<>("order-events", orderId, message));

与传统 MQ 对比

特性	Kafka	RabbitMQ
吞吐量	百万级/秒	万级/秒
消息持久化	磁盘	内存+磁盘
消息回溯	支持 offset 回溯	不支持
延迟	毫秒级	微秒级

Kafka 适用场景： 日志收集与分析、实时流处理、点击流分析、事件溯源。

Spring Boot 整合

spring:
  kafka:
    bootstrap-servers: 127.0.0.1:9092
    producer:
      acks: all
      retries: 3
    consumer:
      auto-offset-reset: earliest
      enable-auto-commit: false

@Service
public class OrderConsumer {

    @KafkaListener(topics = "order-events", groupId = "order-group")
    public void listen(ConsumerRecord<String, String> record) {
        Order order = parseOrder(record.value());
        process(order);
    }
}