Apache Kafka

Apache Kafka 是一种用于持久化事件流的分布式追加日志系统。生产者将记录追加到主题分区，代理复制并保留这些记录，消费者则通过偏移量读取记录。因此，当多个相互独立的应用需要观察相同的有序历史，同时又不希望耦合它们的可用性或处理速率时，Kafka 非常适用。

集群角色和元数据 #

Kafka 集群包含两种服务器角色：

代理存储分区副本，并处理 produce 和 fetch 请求。
KRaft 控制器维护集群元数据、代理成员关系、主题配置、副本放置和分区领导权。

在开发集群中，一个进程可以执行其中一个或两个角色。生产集群会分离控制器和代理角色，因为数据平面负载不应干扰元数据仲裁决策。ZooKeeper 属于 Kafka 的旧版元数据架构，在 Kafka 4.x 中已不存在。

客户端接收的是引导地址列表，而不是完整的路由表。元数据请求会标识当前负责每个分区的代理，之后 produce 和 fetch 请求会直接发送给相关的领导者。因此，每个公布的代理地址都必须能从客户端网络中解析并访问；向外部客户端公布仅容器内部可用的主机名，是常见的部署错误。

Kafka 在 TCP 之上使用版本化的二进制协议。org.apache.kafka:kafka-clients 工件提供 Java 生产者、消费者和管理 API。其他语言的客户端实现相同的协议。

主题、分区和记录 #

主题被划分为有序分区。每条记录可以包含：

一个键；
一个值；
一个时间戳；以及
零个或多个标头。

在协议边界，键和值都是字节数组。序列化器将应用对象转换为这些字节，反序列化器执行逆向操作。

代理会为追加到某个分区的记录分配单调递增的偏移量。偏移量标识该分区内的位置，并不是全局有序的事件 ID。Kafka 只保证单个分区内的顺序，不保证主题范围内的顺序。

如果生产者未指定分区，分区器会使用记录键和当前主题元数据。因此，只要分区数量不变，相等的非 null 键就会被路由到同一个分区。这个特性支持以账户 ID 或设备 ID 等键实现实体级顺序。增加分区数量可能改变键到分区的映射，因此扩展分区需要进行明确的顺序和迁移分析。

每个分区都以段文件的形式存储在代理日志目录下。段文件包含二进制记录批次以及偏移量和时间戳索引；它们不是按行组织的文本日志。

保留和压缩 #

消费不会删除记录。代理根据主题的清理策略删除记录：

cleanup.policy=delete 在达到配置的时间或大小边界后删除旧段。
cleanup.policy=compact 保留每个键的最新值。
cleanup.policy=compact,delete 同时应用两种策略。

压缩会保留偏移量，但不会生成稠密的连续序列。因此，消费者可能会遇到偏移量之间的间隔。值为 null 的带键记录称为墓碑记录；经过 delete.retention.ms 后，压缩可能同时删除该墓碑记录和该键的旧值。

使用管理工具配置压缩状态主题：

bin/kafka-topics.sh --bootstrap-server broker:9092 \
  --create --topic customer-state \
  --partitions 12 --replication-factor 3 \
  --config cleanup.policy=compact \
  --config min.cleanable.dirty.ratio=0.5

压缩是异步的。应用不得假设替换值或墓碑记录追加后，过时值会立即消失。

复制和持久性 #

一个分区具有一个领导者和一个或多个副本。生产者和消费者通常与领导者通信。跟随者获取领导者的日志，当领导权发生变化时，控制器会选出替代领导者。

多个相互独立的设置共同决定已确认写入的持久性：

replication.factor 决定存储该分区的代理数量。
acks=all 使生产者等待领导者满足同步副本要求。
min.insync.replicas 在同步副本数量过少时拒绝 acks=all 写入。
允许非干净领导者选举会使过时副本成为领导者，因此可能丢失已确认的记录。

例如，复制因子为 3 且 min.insync.replicas=2 时，可以容忍一个副本不可用，同时保持双副本确认规则。但写入不能容忍两个副本不可用。这个选择使可用性与持久性相互关联。

副本放置必须跨越主机、机架或可用区等故障域。将三个副本放在同一台物理主机上，并不能提供主机故障容错能力。

创建和检查主题 #

主题数量和分区数量属于运行容量决策，因此应显式创建重要主题，而不是依赖代理端的自动创建：

bin/kafka-topics.sh --bootstrap-server broker:9092 \
  --create --topic orders \
  --partitions 12 --replication-factor 3 \
  --config min.insync.replicas=2 \
  --config retention.ms=604800000

bin/kafka-topics.sh --bootstrap-server broker:9092 \
  --describe --topic orders

bin/kafka-configs.sh --bootstrap-server broker:9092 \
  --entity-type topics --entity-name orders --describe

Java Admin API 提供相同的控制平面：

java

import java.util.List;
import java.util.Map;
import java.util.Properties;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;

Properties config = new Properties();
config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");

try (Admin admin = Admin.create(config)) {
    NewTopic orders = new NewTopic("orders", 12, (short) 3)
        .configs(Map.of(
            "min.insync.replicas", "2",
            "retention.ms", "604800000"
        ));
    admin.createTopics(List.of(orders)).all().get();
}

createTopics 是异步的；等待返回的 future 可以显现授权、验证和超时失败。

Java 客户端依赖 #

生产者、消费者和 Admin 客户端位于同一个工件中：

xml

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>${kafka.version}</version>
</dependency>

请从 Kafka 的兼容性文档中选择客户端版本。协议协商允许混用许多不同版本的客户端和代理，但新的 API 和记录格式仍然要求代理达到最低版本。

生成记录 #

KafkaProducer 是线程安全的。它会将记录累积到各分区批次中，压缩每个批次，并异步发送请求。因此，长期运行的生产者比每条记录创建一个生产者具有更好的性能。

java

import java.nio.charset.StandardCharsets;
import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

Properties config = new Properties();
config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
    StringSerializer.class.getName());
config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
    StringSerializer.class.getName());
config.put(ProducerConfig.ACKS_CONFIG, "all");
config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
config.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "zstd");
config.put(ProducerConfig.LINGER_MS_CONFIG, 5);

try (KafkaProducer<String, String> producer = new KafkaProducer<>(config)) {
    ProducerRecord<String, String> record =
        new ProducerRecord<>("orders", "customer-42", "{\"orderId\":123}");
    record.headers().add("schema-version",
        "1".getBytes(StandardCharsets.UTF_8));

    producer.send(record, (metadata, error) -> {
        if (error != null) {
            System.err.println("Produce failed: " + error);
            return;
        }
        System.out.printf("%s-%d@%d%n",
            metadata.topic(), metadata.partition(), metadata.offset());
    });
}

关闭生产者会刷新未完成的发送。应用应传播或记录回调失败；客户端缓冲区接受一条记录，并不意味着代理已经确认该记录。

重要的吞吐量和延迟控制项包括：

batch.size：一个分区批次累积的最大字节数；
linger.ms：部分填充的批次可等待的时间；
compression.type：应用于整个批次的压缩方式；
buffer.memory：生产者为未发送记录提供的内存；
delivery.timeout.ms：完成发送的上限时间；以及
max.request.size：客户端接受的最大请求大小。

较大的批次可以提高压缩率和请求效率，但会消耗内存，并可能增加延迟。代理和主题的消息大小限制也必须能够接受生成的批次。

幂等性和重试 #

网络故障可能导致生产者无法确定代理是否已经追加请求。盲目重试可能因此产生重复记录。幂等生产会附加生产者身份和序列号，使代理能够拒绝生产者会话内的重复重试。

幂等性不会对两个相互独立的业务请求、由不同生产者实例发送的记录，或应用级重放后重新发送的记录进行去重。稳定的业务 ID 和下游幂等性仍然是必要的。

消费记录 #

KafkaConsumer 不是线程安全的。每个消费者应由一个应用线程独占，该线程负责调用 poll、处理分配回调并提交偏移量。并行处理需要在已完成的工作和偏移量提交之间进行谨慎协调。

java

import java.time.Duration;
import java.util.List;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

Properties config = new Properties();
config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
config.put(ConsumerConfig.GROUP_ID_CONFIG, "order-indexer");
config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
    StringDeserializer.class.getName());
config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
    StringDeserializer.class.getName());
config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(config)) {
    consumer.subscribe(List.of("orders"));

    while (!Thread.currentThread().isInterrupted()) {
        ConsumerRecords<String, String> records =
            consumer.poll(Duration.ofSeconds(1));

        for (ConsumerRecord<String, String> record : records) {
            indexOrder(record.key(), record.value());
        }

        // Commits the next offsets after the complete batch succeeds.
        consumer.commitSync();
    }
}

该示例提供至少一次处理语义：如果 indexOrder 成功后、commitSync 执行前发生故障，就会发生重放。因此，indexOrder 必须是幂等的，或者目标系统必须通过稳定事件 ID 进行去重。

auto.offset.reset 仅在组没有有效的已提交偏移量时生效。它不会强制已建立的组重新读取主题。若要有意进行重放，请使用 seek、新的组 ID 或 consumer-groups 重置操作。

消费者组和再平衡 #

具有相同 group.id 的消费者组成一个组。每个分区在该组中最多分配给一个活跃消费者：

消费者少于分区时，一些消费者会拥有多个分区；
数量相等时，每个消费者拥有一个分区；以及
消费者多于分区时，多余消费者处于空闲状态。

不同的组会分别接收主题历史的独立副本。因此，欺诈检测器和搜索索引器应使用不同的组 ID。

成员关系或订阅发生变化时，分配会更新。协作式分配器可以减少分区移动，而静态成员关系可以避免短时间重启期间不必要的重新分配。但应用仍必须停止处理已撤销的分区，并且只提交已完成的偏移量。

使用以下命令检查分配和滞后：

bin/kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --describe --group order-indexer

滞后量是 log-end-offset - committed-offset。它衡量的是未提交距离，而不是处理延迟。低流量分区可能具有较小的滞后量，但其中最早的未处理记录可能已经很旧，因此应同时监控记录滞后和基于时间的服务目标。

max.poll.interval.ms 限制两次 poll 调用之间的时间。同步处理大型批次可能超过该间隔，并使消费者从组中移除。可以使用 max.poll.records 限制批次，为合理的慢速工作增加间隔，或在暂停分区并仅提交连续完成的偏移量的同时，将工作转移到执行器。

交付语义 #

Kafka 应用可以实现三种结果：

至多一次： 在处理前推进或提交，接受故障后的数据丢失。
至少一次： 在提交前处理，接受重放和重复。
Kafka 精确一次处理： 使用 Kafka 事务原子地提交已生成的记录和已消费的偏移量。

Kafka 事务不会使数据库写入、电子邮件或 HTTP 调用具备原子性。外部作用需要幂等键、 outbox/inbox 设计、去重或其他协调协议。

事务性 Consume-Transform-Produce #

每个运行中的生产者实例都需要一个稳定且唯一的 transactional.id。并发复用同一个 ID 会隔离旧生产者，从而阻止两个实例以同一个事务所有者的身份提交。

java

import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.ProducerFencedException;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

Properties consumerConfig = new Properties();
consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, "order-normalizer");
consumerConfig.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
consumerConfig.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
    StringDeserializer.class.getName());
consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
    StringDeserializer.class.getName());

Properties producerConfig = new Properties();
producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
producerConfig.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,
    "order-normalizer-instance-1");
producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
    StringSerializer.class.getName());
producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
    StringSerializer.class.getName());

try (KafkaConsumer<String, String> consumer =
         new KafkaConsumer<>(consumerConfig);
     KafkaProducer<String, String> producer =
         new KafkaProducer<>(producerConfig)) {
    consumer.subscribe(List.of("orders"));
    producer.initTransactions();

    while (true) {
        var records = consumer.poll(Duration.ofSeconds(1));
        if (records.isEmpty()) {
            continue;
        }

        try {
            producer.beginTransaction();
            Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();

            for (var record : records) {
                producer.send(new ProducerRecord<>(
                    "normalized-orders", record.key(),
                    normalize(record.value())));
                offsets.put(
                    new TopicPartition(record.topic(), record.partition()),
                    new OffsetAndMetadata(record.offset() + 1)
                );
            }

            producer.sendOffsetsToTransaction(
                offsets, consumer.groupMetadata());
            producer.commitTransaction();
        } catch (ProducerFencedException fatal) {
            throw fatal;
        } catch (KafkaException retryableBatchFailure) {
            producer.abortTransaction();
        }
    }
}

必须隐藏已中止记录的消费者应使用 isolation.level=read_committed。事务会增加协调成本和延迟，因此应在确实需要 Kafka 读-处理-写行为具备原子性时使用，而不应默认替代幂等处理。

Kafka Streams #

Kafka Streams 是可嵌入的 Java 库，而不是独立的集群服务。每个应用实例加入一个消费者组，将工作划分为任务，并存储由变更日志主题支持的本地状态。

java

import java.util.Properties;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;

Properties config = new Properties();
config.put(StreamsConfig.APPLICATION_ID_CONFIG, "paid-order-counts");
config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,
    Serdes.String().getClass());
config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,
    Serdes.String().getClass());
config.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, "exactly_once_v2");

StreamsBuilder builder = new StreamsBuilder();
builder.<String, String>stream("orders")
    .filter((customerId, json) -> json.contains("\"status\":\"PAID\""))
    .groupByKey()
    .count()
    .toStream()
    .to("paid-orders-per-customer");

KafkaStreams streams = new KafkaStreams(builder.build(), config);
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
streams.start();

更改 application.id 会创建新的处理身份，并使用不同的消费者组、内部主题和状态目录。应将其视为持久化的应用状态，而不是显示名称。

Kafka Connect 和复制 #

Kafka Connect 运行可复用的源连接器和汇连接器。源连接器从外部系统导入记录；汇连接器导出 Kafka 记录。分布式工作器将连接器配置、偏移量和状态存储在 Kafka 主题中，并在工作器之间重新平衡连接器任务。

Connect 简化了标准集成，但不会消除定义以下内容的必要性：

记录键和分区策略；
模式兼容性；
删除和墓碑记录语义；
重试和死信行为；
目标系统的幂等性；以及
凭据和网络边界。

MirrorMaker 2 使用 Connect 在 Kafka 集群之间复制主题。复制是异步的，并不会将两个集群变成一个同步共识系统。灾难恢复流程必须考虑复制滞后、消费者偏移量转换、主题配置和故障恢复切换。

模式和演进 #

Kafka 存储字节，不强制应用模式。Avro、Protobuf、JSON Schema 或其他编码可以定义记录结构。模式注册表可以将模式 ID 与负载关联，并拒绝不兼容的变更。

兼容性规则应反映数据的重放方式：

新消费者可能会遇到保留数月的旧记录；
生产者发生变化时，旧消费者可能仍在运行；
压缩主题可能无限期保留旧值；以及
从归档恢复可能重新引入旧模式。

添加具有定义默认值的可选字段，比改变字段含义或重新使用字段名称更容易演进。即使序列化格式允许机械式变更，事件名称和字段语义仍然构成长期契约。

安全性 #

Kafka 将传输安全、身份验证和授权分离：

TLS 加密流量，并向客户端验证代理身份。
mTLS 或 SASL 机制验证客户端身份。
ACL 为主题、组、事务 ID 和集群资源上的操作授权。

可以通过属性提供 TLS 客户端配置：

properties

security.protocol=SSL
ssl.truststore.location=/run/secrets/kafka.truststore.p12
ssl.truststore.password=${TRUSTSTORE_PASSWORD}
ssl.keystore.location=/run/secrets/client.keystore.p12
ssl.keystore.password=${KEYSTORE_PASSWORD}
ssl.key.password=${KEY_PASSWORD}

不要将真实密码放入已提交的属性文件。应通过部署平台的密钥机制注入密码，并限制文件系统访问权限。

分区和容量规划 #

分区数量限制消费者组的并行度，但更多分区并非没有代价。每个分区都会增加元数据、打开的文件、拉取副本的工作量、选举和恢复流量。可根据以下因素估算初始分区数量：

峰值入口和出口吞吐量；
最大有效消费者并发数；
每个代理的网络和磁盘吞吐量；
键的倾斜程度；
复制和恢复目标；以及
在可以接受重新分区之前的预期增长量。

保留容量大致为：

text

ingress bytes/second × retention seconds × replication factor

还应为段粒度、索引、压缩、临时不均衡、重新分配和代理恢复预留余量。能够处理稳定状态流量、但无法在恢复目标内重建故障代理的集群，容量不足。

运维和故障诊断 #

至少监控：

离线分区和副本不足的分区；
控制器和代理的可用性；
produce 和 fetch 的延迟及错误率；
请求处理线程和网络线程的饱和度；
磁盘使用量和磁盘延迟；
副本拉取滞后；
消费者偏移量滞后和记录年龄；
事务和身份验证失败；以及
分区不均衡和重新分配进度。

有用的诊断命令包括：

# Observe records without joining an application consumer group.
bin/kafka-console-consumer.sh --bootstrap-server broker:9092 \
  --topic orders --from-beginning \
  --property print.key=true --property print.offset=true

# Inspect one group's committed offsets and lag.
bin/kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --group order-indexer --describe

# Read broker API compatibility information.
bin/kafka-broker-api-versions.sh --bootstrap-server broker:9092

不可用的领导者、同步副本不足、无法访问的公布监听器、授权失败以及过载的代理，都可能在应用侧表现为发送超时。在更改重试上限之前，应将客户端异常与代理指标、分区状态和控制器事件进行关联。

Apache Kafka

集群角色和元数据 #

主题、分区和记录 #

保留和压缩 #

复制和持久性 #

创建和检查主题 #

Java 客户端依赖 #

生成记录 #

幂等性和重试 #

消费记录 #

消费者组和再平衡 #

交付语义 #

事务性 Consume-Transform-Produce #

Kafka Streams #

Kafka Connect 和复制 #

模式和演进 #

安全性 #

分区和容量规划 #

运维和故障诊断 #

参考资料 #

评论