Kafka 基础：Topic、Partition 和 Offset

1. 概述

Kafka 是一个分布式消息代理系统，用于在多个服务之间传输高吞吐量的数据。在编写第一个 producer/consumer 之前，你需要掌握三个核心概念：topic、partition、offset。本文解释了每个概念，并附带具体的 Spring Boot 示例。读完本文后，你将理解 Kafka 中数据的组织和读取方式，避免初次使用时的常见错误。

2. 什么是 Topic

Topic 类似于包含消息的“表”或“通道”。Producer 将消息发送到 topic，consumer 从 topic 读取消息。一个 topic 可以同时有多个 producer 和多个 consumer。

例如，在电商系统中，你可能有以下 topic：

orders.created
orders.cancelled
payments.processed

每个 topic 包含与特定类型事件相关的消息。Kafka 不强制规定消息的 schema——你可以自由选择格式（JSON、Avro、Protobuf……）。

通过 CLI 创建 topic 来更好地理解：

kafka-topics.sh --create \
  --topic orders.created \
  --bootstrap-server localhost:9092 \
  --partitions 3 \
  --replication-factor 1

这条命令创建了一个 topic orders.created，包含 3 个 partition。这正是接下来要讨论的概念。

3. Partition——并行化的单位

3.1. 为什么需要 Partition

一个 topic 被分成多个 partition。每个 partition 是一个独立的日志文件，消息被追加到末尾，永远不会被修改或删除。Partition 让 Kafka 能够：

• 将数据分散到多个 broker。
• 允许多个 consumer 并行读取（每个 consumer 读取一个 partition）。

Topic: orders.created (3 partitions)

Partition 0: [msg0][msg1][msg4][msg7]...
Partition 1: [msg2][msg5][msg8]...
Partition 2: [msg3][msg6][msg9]...

3.2. Key 决定选择哪个 Partition

发送消息时，Kafka 使用 key 来计算 partition（通过哈希函数），确保相同 key 的消息总是进入同一个 partition——从而保持相关消息的顺序。

// 假设按 orderId 作为 key 发送事件
ProducerRecord<String, String> record =
        new ProducerRecord<>("orders.created", "order-123", payloadJson);
kafkaTemplate.send(record);

这行代码创建了一个 ProducerRecord，key 为 "order-123"。这个订单的所有事件都会进入同一个 partition，因此 consumer 可以按发送顺序读取它们。

注意： 如果不提供 key（key = null），Kafka 会以轮询方式分发消息（或从 Kafka 2.4+ 开始使用 sticky partitioner），不保证消息之间的顺序。

3.3. 顺序只在单个 Partition 内得到保证

这是最容易引起混淆的地方：Kafka 仅在单个 partition 内保证顺序，不保证整个 topic 的顺序。如果你需要为一组消息保证绝对顺序（例如某个 orderId 的所有事件），应该为这些消息使用相同的 key。

4. Offset——Partition 中的读取位置

partition 中的每条消息都有一个递增的序号，称为 offset，从 0 开始。

Partition 0:
offset:  0    1    2    3    4
message: m0   m1   m2   m3   m4

Offset 是 partition 范围内的唯一编号，而非整个 topic。Consumer 使用 offset 来知道“已经读到了哪里”。

4.1. Consumer Group 和 Offset Commit

Kafka 通过 consumer group 跟踪已读的 offset。每个 consumer group 对每个 partition 都有独立的 offset，保存在内部 topic __consumer_offsets 中。

# application.properties
spring.kafka.consumer.group-id=order-service
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.enable-auto-commit=false

• group-id：标识 consumer group。多个具有相同 group-id 的实例将共享该 topic 的 partition。
• auto-offset-reset=earliest：如果还没有保存 offset（新的 consumer group），则从 topic 的开头开始读取。
• enable-auto-commit=false：禁用自动提交，以便手动控制何时提交 offset（避免在中途处理错误时丢失消息）。

4.2. 同一 Group 中的多个 Consumer

Kafka 按照以下原则将 partition 分配给同一 group 中的 consumer：在任何时刻，一个 partition 只能被该 group 中的一个 consumer 读取。

Topic orders.created 有 3 个 partition
Consumer group "order-service" 有 2 个实例：

Consumer A -> partition 0, partition 1
Consumer B -> partition 2

如果你为一个有 3 个 partition 的 topic 运行 4 个 consumer，第 4 个 consumer 将 空闲，不会获得任何 partition。因此有用的 consumer 数量的上限 = partition 数量。

5. 在 Spring Boot 中设置 Kafka

5.1. 依赖

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

5.2. Producer 发送消息

@Service
public class OrderEventProducer {

    private final KafkaTemplate<String, String> kafkaTemplate;

    public OrderEventProducer(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void publishOrderCreated(String orderId, String payloadJson) {
        kafkaTemplate.send("orders.created", orderId, payloadJson);
    }
}

kafkaTemplate.send(topic, key, value) 将消息发送到 topic orders.created，使用 orderId 作为 key。Spring Boot 根据 spring.kafka.bootstrap-servers 配置自动创建 KafkaTemplate bean。

5.3. Consumer 读取消息和手动提交 Offset

@Component
public class OrderEventConsumer {

    @KafkaListener(topics = "orders.created", groupId = "order-service")
    public void onOrderCreated(
            ConsumerRecord<String, String> record,
            Acknowledgment ack) {

        System.out.printf("partition=%d offset=%d key=%s value=%s%n",
                record.partition(), record.offset(), record.key(), record.value());

        // 在这里处理业务逻辑
        ack.acknowledge(); // 在处理成功后提交 offset
    }
}

• record.partition() 和 record.offset() 告诉你这条消息在哪个 partition，位置是多少。
• ack.acknowledge() 仅在处理完成后提交 offset——如果服务在中途崩溃，消息将在下次重启时被重新读取（这是因为 enable-auto-commit=false 结合 AckMode.MANUAL）。

需要添加以下配置以使 Acknowledgment 工作：

spring.kafka.listener.ack-mode=manual

试运行时的输出（发送 3 条带不同 key 的消息）：

partition=1 offset=0 key=order-123 value={"status":"CREATED"}
partition=0 offset=0 key=order-456 value={"status":"CREATED"}
partition=2 offset=0 key=order-789 value={"status":"CREATED"}

每个 key 由于哈希值不同而进入不同的 partition——这与第 3.2 节的理论相符。

6. 使用 JUnit 5 进行测试

使用 spring-kafka-test 在单元测试中运行嵌入式 Kafka，不需要真实的 broker。

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka-test</artifactId>
    <scope>test</scope>
</dependency>

@SpringBootTest
@EmbeddedKafka(partitions = 1, topics = "orders.created")
class OrderEventProducerTest {

    @Autowired
    private OrderEventProducer producer;

    @Autowired
    private EmbeddedKafkaBroker embeddedKafka;

    @Test
    void shouldPublishMessageWithGivenKey() throws Exception {
        Consumer<String, String> consumer = createTestConsumer();
        embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "orders.created");

        producer.publishOrderCreated("order-123", "{\"status\":\"CREATED\"}");

        ConsumerRecord<String, String> record =
                KafkaTestUtils.getSingleRecord(consumer, "orders.created");

        assertThat(record.key()).isEqualTo("order-123");
        assertThat(record.value()).contains("CREATED");
        assertThat(record.offset()).isEqualTo(0L);
    }

    private Consumer<String, String> createTestConsumer() {
        Map<String, Object> props = KafkaTestUtils.consumerProps(
                "test-group", "true", embeddedKafka);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        DefaultKafkaConsumerFactory<String, String> factory =
                new DefaultKafkaConsumerFactory<>(props);
        return factory.createConsumer();
    }
}

@EmbeddedKafka 启动一个仅用于测试的内存中 Kafka broker。KafkaTestUtils.getSingleRecord 从 topic 获取第一条消息，我们断言 key、value 和 offset 符合预期——offset 为 0，因为这是唯一 partition 的第一条消息。

7. 常见错误

• 混淆 partition 数和 consumer 数：创建过多的 consumer 会导致某些 consumer 无所事事。Partition 数决定了最大的并行程度。
• 在需要保持顺序时不设置 key：如果你需要同一个实体（订单、用户等）的事件按正确的顺序处理，必须使用 key，不能为 null。
• 在已投入生产后增加 partition：增加已有数据的 topic 的 partition 数量会改变 key 的哈希方式 → 相同 key 的消息可能会进入不同的 partition，破坏了旧数据的顺序保证。应该从一开始就确定合理的 partition 数量。
• 默认的自动提交（enable-auto-commit=true）：offset 按固定周期（默认 5 秒）提交，与是否处理完消息无关——当 consumer 在处理中途崩溃时容易丢失消息。

注意： 如果你的应用不需要严格的顺序且吞吐量不是问题，你可以跳过 key/partition 的优化——Kafka 使用默认配置仍然可以正常工作。在没有实际需求时不要过早优化。

8. 结论

Topic 是按主题存放消息的地方，partition 是用于并行化和分散的较小单元，offset 是每个 partition 中的读取位置。深入理解这三个概念可以帮助你正确设计：选择合理的 key 来保持顺序、选择与 consumer 数量相适应的 partition 数、控制 offset 提交以避免在出错时丢失数据。这是在深入研究更高级主题（如 rebalancing、exactly-once 语义或 Kafka Streams）之前必须掌握的基础。