Kafka Partition Key设计：消息顺序与负载均衡

1. 概述

在 Kafka 中，partition key 决定了消息落入哪个 partition。由此，它决定了两件事：哪些消息保持正确的顺序，以及负载如何在各个 consumer 之间分配。

选择错误的 key 通常会导致两种后果之一。第一种是同一订单的事件被混乱处理。第二种是一个“热” partition 承载了几乎所有流量，而其他 partition 则闲置。

读完本文后，你将理解 Kafka 如何将 key 映射到 partition。你还将学会根据顺序要求选择 key，如何处理热 key，以及如何在 Spring Boot 3.x 中配置 producer/consumer，以避免意外破坏消息顺序。

2. Kafka 如何选择 partition

2.1. Key 不为 `null`：哈希后取余

当消息有 key 时，producer 按以下公式计算 partition：

partition = toPositive(murmur2(keyBytes)) % numPartitions

我们可以使用 kafka-clients 中提供的实用函数来重现这个逻辑：

import org.apache.kafka.common.utils.Utils;
import java.nio.charset.StandardCharsets;

public final class PartitionCalculator {

    private PartitionCalculator() {}

    public static int partitionFor(String key, int numPartitions) {
        byte[] keyBytes = key.getBytes(StandardCharsets.UTF_8);
        return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
    }

    public static void main(String[] args) {
        for (String orderId : new String[]{"order-1001", "order-1002", "order-1001"}) {
            System.out.println(orderId + " -> partition " + partitionFor(orderId, 6));
        }
    }
}

逐部分解释：

Key 通过 StringSerializer 转换为字节，即 UTF-8。我们在示例中使用 UTF-8 以与该行为一致。
Utils.murmur2 是 default partitioner 使用的哈希函数。
toPositive 移除符号位以确保结果为正。随后的 % 运算得出 partition 编号。

运行结果如下所示（具体数字取决于 key）：

order-1001 -> partition 3
order-1002 -> partition 0
order-1001 -> partition 3

关键点：相同的 key 和相同的 partition 数量总是导致相同的 partition。 Kafka 只保证 partition 内的顺序。因此，如果要保持某组消息的顺序，这些消息必须使用相同的 key。

2.2. Key 为 `null`：没有顺序保证

从 Kafka 3.3 开始，没有 key 的消息根据“uniform sticky”机制分配。Producer 将一个批次放入一个 partition，然后切换到另一个 partition，考虑每个 broker 的速度。负载分配相当均匀，但两条连续的消息可能在不同的 partition 中。这样就没有顺序保证了。

注意： 哈希取决于 key 的字节而不是逻辑值。如果一个服务使用 LongSerializer 发送 key 1001，另一个服务使用 StringSerializer 发送 "1001"，这两条消息可能会进入不同的 partition。请在整个系统中统一 key 的 serializer。

3. 使用 Spring Boot 发送带 key 的消息

添加依赖：

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

在 application.yml 中配置 producer：

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all
      properties:
        enable.idempotence: true
        max.in.flight.requests.per.connection: 5

各设置的含义：

acks: all 和 enable.idempotence: true 有助于重试时不会创建重复记录。
更重要的是，启用幂等性时，broker 会拒绝顺序错误的批次。因此，即使发生重试且 max.in.flight 达到 5，partition 内的顺序仍然得以保持。

发送事件的服务：

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

public record OrderEvent(String orderId, String customerId, String type) {}

@Service
public class OrderEventPublisher {

    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;

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

    public void publish(OrderEvent event) {
        kafkaTemplate.send("order-events", event.orderId(), event)
                .whenComplete((result, ex) -> {
                    if (ex == null) {
                        var md = result.getRecordMetadata();
                        System.out.printf("%s %s -> p%d@%d%n",
                                event.orderId(), event.type(), md.partition(), md.offset());
                    }
                });
    }
}

send(topic, key, value) 的第二个参数是 key。这里我们使用 orderId。如果依次为同一订单发送 CREATED、PAID、SHIPPED，输出将如下所示：

order-1001 CREATED -> p3@120
order-1001 PAID -> p3@121
order-1001 SHIPPED -> p3@122

三个事件都在 partition 3 中，offset 依次增加。Consumer 将按正确的顺序读取它们。

4. 选择 key：从顺序要求开始

第一个问题不是“哪个 key 分布最均匀”。正确的问题是：哪些消息必须按顺序处理？ Key 应该是满足这一要求的最小组的标识符。

业务要求	合适的 Key	备注
一个订单的事件必须按顺序	`orderId`	许多 key，负载均衡好
客户钱包余额加减必须按顺序	`customerId` 或 `walletId`	如果有大客户可能遇到热 key
访问日志，不需要顺序	`null`	负载最均匀
按产品的库存	`sku`	“热”产品可能导致负载不均

伴随的两个原则：

Key 的粒度越细，负载均衡越好。 orderId 有数百万个值，所以分布均匀。countryCode 只有几个值，所以少数几个 partition 会承载几乎所有负载。
Key 的值数应该远大于 partition 数。 如果只有 10 个 key 而 topic 有 12 个 partition，肯定会有空 partition。通常也会有 partition 承载多个较重的 key。

5. 热 partition 及其处理

假设 key 是 customerId，一个企业客户产生了 40% 的流量。包含该客户的 partition 将出现延迟。添加更多 consumer 也无济于事，因为 一个 partition 在同一时刻只能被 group 中的一个 consumer 读取。

5.1. Salting：将热 key 分散到多个 bucket

想法是在 key 中添加后缀，将热 key 分散到多个 partition：

import java.util.Set;

public class OrderKeyStrategy {

    private final Set<String> hotCustomers;
    private final int buckets;

    public OrderKeyStrategy(Set<String> hotCustomers, int buckets) {
        this.hotCustomers = hotCustomers;
        this.buckets = buckets;
    }

    public String keyFor(OrderEvent event) {
        if (!hotCustomers.contains(event.customerId())) {
            return event.customerId();
        }
        int bucket = Math.floorMod(event.orderId().hashCode(), buckets);
        return event.customerId() + "#" + bucket;
    }
}

对于普通客户，key 保持为 customerId。对于热客户，bucket 由 orderId 计算，所以 key 的形式为 cust-big#0 … cust-big#7。

需要明确的结果：

同一订单的所有事件仍然有相同的 key，因此仍然保持顺序。
同一客户的不同订单之间的顺序不再得到保证。

Salting 只有在业务接受这种折衷时才能使用。如果你需要完全按客户顺序（例如钱包余额），salting 会导致数据错误。

注意： 不要使用随机值或时间戳作为 bucket。这样做会导致同一订单的事件落入多个 partition，顺序会被完全破坏。

5.2. 测试 key 策略

下面的测试检查两个主要行为：同一订单进入同一 partition，以及热客户分散到多个 partition。

import org.junit.jupiter.api.Test;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import static org.assertj.core.api.Assertions.assertThat;

class OrderKeyStrategyTest {

    private static final int PARTITIONS = 12;
    private final OrderKeyStrategy strategy = new OrderKeyStrategy(Set.of("cust-big"), 8);

    @Test
    void sameOrder_alwaysGoesToSamePartition() {
        var created = new OrderEvent("order-1", "cust-big", "CREATED");
        var paid = new OrderEvent("order-1", "cust-big", "PAID");

        int p1 = PartitionCalculator.partitionFor(strategy.keyFor(created), PARTITIONS);
        int p2 = PartitionCalculator.partitionFor(strategy.keyFor(paid), PARTITIONS);

        assertThat(p1).isEqualTo(p2);
    }

    @Test
    void hotCustomer_isSpreadAcrossPartitions() {
        Set<Integer> partitions = IntStream.range(0, 1_000)
                .mapToObj(i -> new OrderEvent("order-" + i, "cust-big", "CREATED"))
                .map(e -> PartitionCalculator.partitionFor(strategy.keyFor(e), PARTITIONS))
                .collect(Collectors.toSet());

        assertThat(partitions).hasSizeGreaterThan(1);
    }

    @Test
    void normalCustomer_keepsCustomerIdAsKey() {
        var event = new OrderEvent("order-9", "cust-small", "CREATED");
        assertThat(strategy.keyFor(event)).isEqualTo("cust-small");
    }
}

测试使用了第 2.1 节中的 PartitionCalculator，它准确重现了 producer 的公式。因此我们不需要建立 broker，仍然可以验证消息实际上会进入哪个 partition。

6. Consumer 端：不要自己破坏顺序

Producer 保持顺序还不够。Consumer 也可能破坏它。

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class OrderEventListener {

    @KafkaListener(topics = "order-events", groupId = "order-service", concurrency = "6")
    public void onEvent(OrderEvent event) {
        // 在 listener 的线程中顺序处理
        System.out.println(Thread.currentThread().getName() + " " + event);
    }
}

concurrency = "6" 创建 6 个 consumer 线程。每个 partition 被分配给恰好一个线程，因此 partition 内的顺序得以保持。如果 topic 有 6 个 partition，每个线程获得一个 partition。如果设置 concurrency 大于 partition 数，多余的线程将处于空闲。

Consumer 端常见的错误：

将消息推送到 listener 中的单独线程池（executor.submit(...)）。同一订单的两个事件可能并行运行，PAID 可能在 CREATED 之前完成。如果需要比 partition 级别更深的并行化，请按 key 并行化，每个 key 一个顺序队列。不要按单个消息并行化。
通过发送到另一个 topic 进行重试（retry topic，non-blocking retry）。错误消息被分离出主流，相同 key 的后续消息继续被处理。对于需要严格顺序的流，应使用带 BackOff 的 DefaultErrorHandler 进行阻塞重试。这种方式接受 partition 暂时停止。

7. 改变 partition 数：隐藏的陷阱

由于公式中有 % numPartitions，将 partition 从 6 增加到 12 会导致大多数 key 转移到不同的 partition：

public class RepartitionDemo {
    public static void main(String[] args) {
        String key = "order-1001";
        System.out.println("6 partitions  -> " + PartitionCalculator.partitionFor(key, 6));
        System.out.println("12 partitions -> " + PartitionCalculator.partitionFor(key, 12));
    }
}

6 partitions  -> 3
12 partitions -> 9

order-1001 的旧事件在 partition 3 中，可能还没有被完全处理。新事件进入 partition 9。不同的 consumer 可能同时处理它们，导致顺序颠倒。

降低风险的方法：

从一开始就预估足够的 partition 数。 根据目标吞吐量除以单个 consumer 的吞吐量，然后乘以安全系数。
如果必须增加 partition，停止 producer，等待 consumer 处理完 lag，增加 partition，然后重新启动 producer。
或者创建一个新 topic 有更多 partition，然后逐步将流转移到新 topic。

注意： Kafka 不支持减少 partition 数。选择过多的 partition 也有成本：消耗 broker 资源、rebalance 变慢、打开更多文件。预留足够但不过量，不要为了“保险”而设置 1000 个 partition。

8. 何时需要自定义 partitioner

你很少需要自己编写 Partitioner。大多数情况下，选择正确的 key 就足够了。自定义 partitioner 只有在需要显式映射时才合理，例如为 VIP 客户保留几个 partition。

反过来，自定义 partitioner 有一些缺点：

写入该 topic 的每个 producer，包括用其他语言编写的服务，都必须使用相同的逻辑。否则顺序会被破坏。
Kafka Streams 和 Kafka Connect 等工具默认使用 murmur2。它们可能与你的 partition 方式不匹配。

因此，在考虑 partitioner 之前，优先通过 key 设计来解决问题。

9. 结论

Partition key 是一个设计决策，而不仅仅是 send() 的一个参数。需要记住的主要要点：

Kafka 只保证 partition 内的顺序。相同的 key 和相同的 partition 数会产生相同的 partition，按照 murmur2 哈希取余公式。
选择 key 应该是需要保持顺序的最小组的标识符。Key 的值越多，负载分配越均匀。
通过 salting 处理热 key，但仅当业务接受 bucket 之间的顺序丢失时。
在 producer 端启用幂等性。在 consumer 端，不要随意将消息推送到线程池，在使用 non-blocking retry 前要仔细考虑。
避免在运行中的 topic 上增加 partition，如果流需要顺序的话。从一开始就要预留足够的 partition。

掌握这些要点，你就能设计既保持正确顺序又能在流量增加时扩展的 topic。

Kafka Partition Key设计：消息顺序与负载均衡

1. 概述

2. Kafka 如何选择 partition

2.1. Key 不为 null：哈希后取余

2.2. Key 为 null：没有顺序保证

3. 使用 Spring Boot 发送带 key 的消息

4. 选择 key：从顺序要求开始

5. 热 partition 及其处理

5.1. Salting：将热 key 分散到多个 bucket

5.2. 测试 key 策略

6. Consumer 端：不要自己破坏顺序

7. 改变 partition 数：隐藏的陷阱

8. 何时需要自定义 partitioner

9. 结论

同主题文章

Kafka 中的 Consumer Group 与 Rebalance 在 Spring Boot 中的应用

Kafka producer：acks、幂等性和批处理

Saga pattern 与 Kafka 分布式事务：预订旅游套餐示例

Schema Registry 和 Avro/Protobuf 在 Kafka 与 Spring Boot 中的应用

事件溯源和 Kafka：保存事件而非状态

2.1. Key 不为 `null`：哈希后取余

2.2. Key 为 `null`：没有顺序保证