Apache Kafka

Kafka 是一个开源的分布式事件流平台,被数千家公司用于高性能数据管道、流式分析、数据集成和关键业务应用。它的核心是一个发布/订阅(pub/sub)系统:生产者把消息写入主题(topic),消费者从它们关心的主题读取。它远远超出了 ActiveMQ 或 RabbitMQ 这类传统消息队列。

主要用途

• 削峰缓冲——在高峰窗口吸收突发(外卖的午餐高峰、电商的闪购),让下游服务不被压垮。
• 服务解耦——把数据源和消费者分开;Kafka MirrorMaker 在数据湖和可用区之间搬运数据。
• 异步通信——把非关键工作从请求路径上卸下,改善用户侧延迟。

发布/订阅模型

不同类别的消息从各种生产者流入不同的主题(topic)。与传统队列不同,消息在被消费后仍然保留——Kafka 独立于任何单个消费者管理生命周期,所以多个消费者可以在不同的 offset 读同一个主题。

生产者架构

生产者用一个 ProducerRecord(消息包装)调用 .send()。流程:

• 序列化器(serializer)把消息转成字节流(自定义序列化器,或通过 Schema Registry 用 Protobuf/Avro)。
• 分区器(partitioner)决定目标分区,实现跨集群的分布式存储和并行处理。
• 记录在发送前先排进RecordAccumulator(内存)。两个旋钮控制时机:BATCH_SIZE_CONFIG(队列阈值)和 LINGER_MS_CONFIG(最大等待)。
• Sender 创建网络客户端并等待 acks 确认。

消费者模型

Kafka 用拉(pull)模型——消费者按自己的节奏请求数据,比推(push)更解耦。许多独立消费者可以读同一个主题,但同一个消费者组内的消费者不能共享一个分区:整个组表现为一个逻辑订阅者,每个分区恰好被一个成员拥有。

分区与副本

并行来自把一个主题拆成分散在多台服务器上的分区。DefaultPartitioner 用 hash(key) % num_partitions;自定义实现重写 partition()。

每个分区是一个追加写日志,按 1 GB 分段,每 4 KB 一条稀疏索引记录以便快速查找。保留策略(log.retention.hours/minutes/ms,默认 7 天)根据 log.cleanup.policy 删除或压实旧数据。复制通过 leader/follower 方案防范单点故障。

系统可靠性

acks

这个生产者侧设置控制一条记录在被视为写入成功前必须被多少个 broker 接收——核心的持久性/吞吐取舍:

重试与投递语义

Kafka ≥ 2.1 的 RETRIES_CONFIG 默认是 MAX_INT。重试若不加管理会有重复风险:

• At-Least-Once(至少一次)——acks=-1;消息绝不丢失但可能重复。
• At-Most-Once(至多一次)——acks=0;无重复但消息可能丢失。
• Exactly-Once(恰好一次)——At-Least-Once + 幂等。每个生产者拿到一个唯一的 Producer Id(PID),每条消息有单调递增的序列号。Broker 跟踪每个分区最大的 <PID, sequence> 并拒绝重复(enable.idempotence=true)。TransactionManager 加上类数据库的 beginTransaction()/commitTransaction() 包住 .send()。

副本

多个 follower 从 leader 同步。生产者只写 leader;follower 从它拉取。ISR 列出同步中的 follower;落后超过 replica.lag.time.max.ms 的 follower 移到 OSR。ISR + OSR = 已分配副本(AR)。leader 故障时,从 ISR 选出新 leader。每个副本跟踪自己的 LEO(Log End Offset)以衡量同步进度。

高速读写

• 集群计算——生产者和消费者都并行处理。
• 稀疏索引——.index 文件加速读取。
• 追加写日志——只做顺序写,远快于随机访问。
• 页缓存(Page Cache)——操作系统把文件元数据/索引缓存在内存,数据在磁盘;Kafka 依赖这种分层方式而非自建缓存。
• 零拷贝(Zero-copy)——数据从读缓冲直接传到 socket 缓冲,跳过用户/内核上下文切换。(SSL 下这个优势消失,因为 broker 必须解密再重新加密。)

ZooKeeper

分布式系统的协调服务。对 Kafka,它跟踪集群节点状态和主题/消息元数据。五项主要功能:

• Controller 选举——选第一个既在 AR 又在 ISR 的 broker(/kafka/controller)。
• 集群成员——列出存活的 broker(/kafka/brokers/ids)。
• 主题配置——存储完整主题细节(/kafka/brokers/topics/{topic}/partitions/{id}/state)。
• 访问控制列表(ACL)。
• 配额(Quotas)——监控每客户端的读写额度。

CLI 示例

创建一个 2 分区、复制因子 3 的主题:

# test.config → bootstrap.servers=localhost:9092
kafka-topics \
  --bootstrap-server localhost:9092 \
  --command-config ./test.config \
  --topic test1 \
  --create \
  --replication-factor 3 \
  --partitions 2

生产带 key 的消息,然后从头消费:

# 生产者
kafka-console-producer \
  --topic test1 \
  --broker-list localhost:9092 \
  --property parse.key=true \
  --property key.separator=:
# > key:{"val":0}

# 消费者
kafka-console-consumer \
  --topic test1 \
  --bootstrap-server localhost:9092 \
  --property print.key=true \
  --from-beginning

消费者组与再均衡

消费者组是 Kafka 并行消费的单位。组里每个消费者实例被分到一个互不相交的分区子集,整个组一起消费完整主题。给一个 5 分区的主题加上第 6 个消费者,那个消费者会闲着——没有多余分区可分。这个约束故意保持简单:它意味着每个分区在每个组里恰好有一个权威 offset,进度永远不含糊。

再均衡协议

只要组成员变化就触发再均衡(rebalance):有新消费者加入、一个现有的崩溃或超过 session.timeout.ms、主题加了新分区,或应用调了 unsubscribe()。再均衡期间,Group Coordinator(为每个消费者组选出的一个 broker)停止所有消费、撤销分区分配,然后重新分配。代价是一次暂停——有时几秒——期间不处理消息。因此尽量减少再均衡的频率和时长是一个重要的运维关注点。

# 调这些来减少不必要的再均衡
session.timeout.ms=45000        # 静默的消费者多久后被判定为死
heartbeat.interval.ms=15000     # 必须 < session.timeout.ms / 3
max.poll.interval.ms=300000    # 两次 poll() 之间的最大间隔,超过则被踢出
partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor

Eager vs. Cooperative 再均衡

经典的 eager 再均衡(RangeAssignor 和 RoundRobinAssignor 用的)在任何重分配开始前撤销所有分区分配——一次全局 stop-the-world 暂停。Kafka 2.4+ 引入了协作式粘性再均衡协议(CooperativeStickyAssignor):消费者只撤销需要移动的分区,保留它继续持有的那些。这通常对不涉及重分配的消费者完全消除暂停,并让分区分配保持"粘性"以最小化不必要的状态迁移。

静态组成员

Kafka 2.3+ 加了 group.instance.id,给消费者一个跨重启的持久身份。有了静态成员,一个崩溃后在 session.timeout.ms 内重新加入的消费者无需再均衡就能收回它原来的分区——对容器化负载(部署时 pod 频繁重启)是个重要优化。

日志压实(Log Compaction)

Kafka 的保留有两种口味。默认的 delete 策略只是删除比 log.retention.hours 旧(或超过大小阈值)的段。日志压实(log compaction)根本不同:它不按时间丢弃消息,而是只保留每个 key 的最新消息,形成一个行为像最终一致键值存储的压实日志。

# 为用户画像变更日志创建一个压实主题
kafka-topics \
  --bootstrap-server localhost:9092 \
  --topic user-profiles \
  --create \
  --partitions 12 \
  --replication-factor 3 \
  --config cleanup.policy=compact \
  --config min.cleanable.dirty.ratio=0.1 \
  --config segment.ms=86400000    # 每天滚动一个新段

日志清理器怎么工作

broker 的后台日志清理器(log cleaner)线程扫描每个分区的"脏"(最近写入)部分。它们在内存建一个 offset 映射(key → 最新 offset),然后只把脏部分里每个 key 的最新记录拷进一个新的"干净"段,丢弃较旧的重复。结果:压实日志只随不同的 key 增长,而不随时间增长。

一条特殊的墓碑消息(tombstone)——key 非空、value 为 null 的记录——表示删除。压实后,连墓碑最终也会被移除(在 delete.retention.ms 之后),所以读压实日志的消费者看到该 key 被彻底抹掉。

KRaft——没有 ZooKeeper 的 Kafka

ZooKeeper 多年作为 Kafka 的集群协调层:controller 选举、broker 成员、主题元数据。但它引入了一个独立的运维依赖、独立的扩展关注点,以及集群启动时的瓶颈。KRaft(Kafka Raft Metadata 模式),Kafka 2.8 引入、从 3.3 起生产可用,用 Kafka 自带的 Raft 共识实现完全替代了 ZooKeeper。

架构转变

在 KRaft 下,一部分 broker 被指定为 controller。它们运行一个专用的元数据分区(主题 __cluster_metadata),通过 Raft 协议复制。活跃的 controller 就是 Raft leader。所有集群元数据——broker 注册、主题配置、分区 leader——都活在这个单一日志里,而非散布在 ZooKeeper znode 中。

# KRaft server.properties(broker+controller 合一节点)
process.roles=broker,controller
node.id=1
controller.quorum.voters=1@kafka1:9093,2@kafka2:9093,3@kafka3:9093
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
inter.broker.listener.name=PLAINTEXT
controller.listener.names=CONTROLLER
log.dirs=/var/lib/kafka/data

KRaft 的好处

• 单一运维足迹——一个系统部署、监控、升级,而非两个。
• 启动和 leader 选举更快——元数据已在 Kafka 日志里;无需 ZooKeeper epoch 握手。大集群重启从几分钟降到几秒。
• 更高分区数——ZooKeeper 在约 20 万分区以上吃力;KRaft 瞄准百万级。元数据日志随 Kafka 自己的存储扩展,而非 ZooKeeper 内存常驻的 znode。
• 简化安全——一套 TLS/SASL 配置,而非 Kafka 和 ZooKeeper 各一套。

性能调优参数

Kafka 的默认值是保守的。真实世界的调优瞄准三个维度:生产者吞吐、消费者延迟(lag)、持久性保证。下面这些参数是最关键的杠杆。

生产者调优

# 吞吐优化的生产者
acks=all                   # 持久写到所有 ISR 副本
enable.idempotence=true    # 去重重试,生产者会话内 exactly-once
linger.ms=5               # 最多等 5ms 攒批;0 = 立即发送
batch.size=65536          # 64 KB 批;越大往返越少、延迟越高
compression.type=lz4     # lz4 CPU/压缩比最佳;snappy 也常见
buffer.memory=67108864   # 64 MB 内存缓冲;高量生产者调大
max.in.flight.requests.per.connection=5  # 开了幂等时 >1 也没问题

消费者调优

fetch.min.bytes=1024         # 等到有 1 KB 可用;减少 fetch 往返
fetch.max.wait.ms=500        # 等待 fetch.min.bytes 满足的最大时间
max.poll.records=500         # 每次 poll() 的记录数;处理慢就调小
auto.offset.reset=earliest  # 首次运行:从最旧开始;latest = 只读尾部
enable.auto.commit=false    # 手动提交以实现 exactly-once 处理语义

Broker 与主题调优

# broker 级
num.io.threads=8             # I/O 线程数;设成磁盘数
num.network.threads=3       # 网络请求线程
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
log.flush.interval.messages=10000  # 每 1 万条 fsync;交给 OS 性能最佳

# 通过 kafka-configs 做主题级覆盖
kafka-configs --bootstrap-server localhost:9092 \
  --alter --entity-type topics --entity-name orders \
  --add-config retention.ms=604800000,min.insync.replicas=2

Kafka vs. RabbitMQ vs. Apache Pulsar

选对消息系统是你能做的最高杠杆的架构决策之一。Kafka、RabbitMQ、Pulsar 各占设计空间里不同的一点。

Kafka Streams 与 Kafka Connect

Kafka 不只是消息代理——它是整个流处理生态的基础。两个组件把它延伸得最远:Kafka Streams 在你的 JVM 应用内做流处理,Kafka Connect 做即插即用的数据管道连接器。

Kafka Streams

Kafka Streams 是一个轻量客户端库(不是集群),用于构建流处理应用。不像 Apache Flink 或 Spark Streaming,没有单独的处理集群要管——库跑在你的应用进程里,从输入主题读、把结果写到输出主题。容错靠把状态(窗口聚合、连接表)存在磁盘上的 RocksDB,并由一个 Kafka changelog 主题支撑,用于崩溃后恢复。

StreamsBuilder builder = new StreamsBuilder();

// 从 "orders" 主题读,按 userId 分组,在 5 分钟窗口内计数
KStream<String, Order> orders = builder.stream("orders");
orders
  .groupByKey()
  .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
  .count()
  .toStream()
  .to("orders-per-user-5min");  // 把结果写回 Kafka

KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();

Kafka Streams 的关键概念:

• KStream——无界的记录流;每条记录是独立事件。
• KTable——把流看作变更日志;每个 key 的值是它的最新更新(像物化视图)。
• GlobalKTable——完整复制到每个实例的 KTable(适合无需重分区的广播连接)。
• Stream-Table Join——用查找表的当前状态丰富每个事件;连接是本地的,因为两者按 key 同分区。
• 窗口——tumbling(不重叠)、hopping(重叠)、session(活动驱动)、sliding 窗口,用于基于时间的聚合。
• Exactly-once——Kafka Streams 通过横跨 Kafka 读和写的事务支持端到端 exactly-once 语义。

Kafka Connect

Kafka Connect 是一个可扩展、容错的框架,用于在 Kafka 和外部系统(数据库、对象存储、搜索引擎、SaaS API)之间搬数据,无需写自定义代码。连接器有两种:source 连接器从外部系统读、写入 Kafka(如 Debezium PostgreSQL CDC、S3 source),sink 连接器从 Kafka 读、写入外部系统(如 Elasticsearch sink、Snowflake sink、JDBC sink)。

{
  "name": "postgres-orders-cdc",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "db.prod.internal",
    "database.port": "5432",
    "database.user": "debezium",
    "database.dbname": "orders",
    "table.include.list": "public.orders,public.inventory",
    "topic.prefix": "cdc",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_orders",
    "publication.name": "dbz_publication"
  }
}

Connect 作为一个 worker 集群运行,把连接器任务分摊到各 worker。若一个 worker 失败,它的任务被重新分配给存活的 worker——和消费者组再均衡完全相同的机制。这意味着 Connect 靠加 worker 横向扩展,并在节点故障时无需运维介入就能存活。

Schema Registry 与 Avro

随着 schema 演化,Kafka 主题里的裸字节会变成维护噩梦。Confluent Schema Registry(或 AWS Glue Schema Registry)存储 Avro、Protobuf 或 JSON Schema 定义,并给每个版本分配一个数字 ID。生产者通过在载荷前缀嵌入 4 字节 schema ID 来序列化消息;消费者按 ID 查 schema 来反序列化。这实现了带兼容性检查的 schema 演化:registry 拒绝与已注册版本向后或向前不兼容的新 schema 版本,防止独立部署的服务之间悄悄的数据损坏。

{
  "type": "record",
  "name": "OrderEvent",
  "namespace": "com.example.orders",
  "fields": [
    { "name": "orderId",  "type": "string" },
    { "name": "userId",   "type": "string" },
    { "name": "totalCents", "type": "long" },
    { "name": "currency", "type": "string", "default": "USD" }
  ]
}

常见坑与最佳实践

分区数——选对它

分区是 Kafka 的并行单位,创建后不能减少(只能增加)。选太少会封顶你的消费者吞吐上限;选太多浪费内存并增加再均衡时间。一个粗略起点:按你 broker 的磁盘带宽,每分区瞄准 1–3 MB/s 吞吐,并大致每个你预期峰值运行的消费者实例配 1 个分区。一个常见错误是为"保持简单"用单分区建主题——这让组里除一个外的每个消费者都闲着。

Offset 管理

用 enable.auto.commit=true 时,offset 会被周期性提交,而不管消费者是否真的处理了消息。在提交和下游写之间崩溃意味着消息被永久跳过。用 enable.auto.commit=false,只在你的处理完成且持久化后才提交 offset。要 exactly-once,把下游写和 offset 提交包进一个 Kafka 事务。

消费者 lag 监控

消费者 lag——最新生产 offset 和消费者已提交 offset 之间的差距——是 Kafka 消费者的首要运维健康指标。持续增长且不回落的 lag 表示要么消费者慢,要么生产突然飙升。用 Kafka 内建的 consumer group describe、Confluent 的 Kafka Consumer Lag Exporter,或 LinkedIn 的 Burrow 监控它。

# 查看一个消费者组所有分区的 lag
kafka-consumer-groups \
  --bootstrap-server localhost:9092 \
  --group order-processing-group \
  --describe

# 示例输出
# GROUP                   TOPIC  PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# order-processing-group  orders  0          1023440         1023441         1
# order-processing-group  orders  1          998772          999201          429  ← lag 在涨!

避免常见反模式

• 每种事件一个主题,而非所有东西一个主题——把无关事件类型混在一个主题里,迫使消费者反序列化并过滤它们不关心的消息,还阻碍独立的保留策略。
• 别在 Kafka 里存大载荷——消息保持在 1 MB 以内;更大的载荷(图片、报表)存 S3,只把引用 URL 放进 Kafka。默认 message.max.bytes 是 1 MB 是有原因的。
• 别把 Kafka 当数据库——虽然日志压实近似一个 KV 存储,Kafka 没有随机访问读路径。要查找,把压实日志物化进 Redis 或数据库。
• 给毒丸消息用死信主题——一条反序列化失败的畸形消息,若你的消费者在它上面崩溃,会永远阻塞一个分区。把失败消息路由到专用 DLQ 主题并告警。

真实使用场景

理解组织为何选 Kafka——而非更简单的队列——能巩固你在面试里何时推荐它。

• LinkedIn(Kafka 起源)——用户活动跟踪:每次页面浏览、点击、投递都发布到主题,喂给实时仪表盘、推荐模型和 A/B 测试分析。重放至关重要:一个模型重训练作业能扫 30 天的原始事件。
• Uber——数百万司机的地理位置更新以每秒数百万事件流经 Kafka 主题。Kafka 把司机 app 与调度、地图、动态定价服务解耦。主题保留 7 天,任何落后的下游服务都能追上。
• Netflix——生产数据库的 CDC 流经 Debezium 进入 Kafka,再进 Elasticsearch 做搜索、进 S3 的 Iceberg 表做分析。Kafka 的保留窗口给分析管道一个 Spark 作业失败时的恢复窗口。
• 电商结账管道——典型的系统设计例子:订单服务发布一个 OrderCreated 事件,独立的消费者分别处理库存扣减、支付、通知派发和推荐信号记录——全部独立扩展、故障隔离。