4、生产者分区策略：默认分区器、自定义分区器、消息键（Key）与分区路由的关系

分区策略是 Kafka 生产者将消息发送到特定分区的决策逻辑。它直接决定了消息的分布均匀性、顺序性保证以及下游消费者的负载均衡效果。本章深入剖析默认分区器的工作机制、自定义分区器的实现方法，以及消息键（Key）在分区路由中的核心作用。

4.1 分区路由的基本流程

生产者发送消息时，分区路由发生在 Producer.send() 调用内部，经过以下决策链：

1. 指定分区：如果 ProducerRecord 显式指定了分区号，则直接使用该分区，不再执行分区器逻辑。
2. 消息键非空：如果未指定分区但消息键不为 null，则使用分区器对键进行哈希计算，映射到目标分区。
3. 消息键为空：如果未指定分区且消息键为 null，则使用分区器的轮询（Round-Robin）或粘性（Sticky）策略选择分区。

4.2 默认分区器（DefaultPartitioner）

Kafka 2.4 版本之后，默认分区器采用 粘性分区（Sticky Partitioner） 策略，替代了旧版的轮询策略。其核心行为如下：

4.2.1 粘性分区策略详解

当消息键为空时，默认分区器不会为每条消息单独计算分区，而是：

• 选择一个分区作为“粘性分区”，持续向其发送一批消息。
• 当批次填满或达到 linger.ms 时间后，再切换到下一个分区。
• 相比旧版轮询策略，减少了分区切换频率和网络请求次数，显著提升吞吐量。

4.3 消息键（Key）与分区路由的关系

消息键是分区路由的核心输入。其路由规则遵循以下公式：

targetPartition = Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions

关键特性：

• 确定性：相同的 Key 始终路由到同一分区（分区数不变的前提下）。
• 顺序保证：同一 Key 的消息在分区内保持发送顺序，消费者可感知到严格顺序。
• 数据倾斜风险：如果 Key 分布不均匀（如热点 Key），会导致部分分区数据量远大于其他分区。

4.4 自定义分区器

当默认分区器无法满足业务需求时（如需要基于业务字段路由、实现加权分配、避免数据倾斜），可以实现 Partitioner 接口自定义分区逻辑。

4.4.1 接口定义

public interface Partitioner extends Configurable, Closeable {
    int partition(String topic, Object key, byte[] keyBytes,
                  Object value, byte[] valueBytes, Cluster cluster);
    void close();
    void configure(Map<String, ?> configs);
}

4.4.2 实现示例：基于业务字段的加权分区器

public class WeightedPartitioner implements Partitioner {
    private Map<Integer, Integer> partitionWeights;
    private int totalWeight;

    @Override
    public void configure(Map<String, ?> configs) {
        // 从配置中读取分区权重映射
        partitionWeights = parseWeights(configs);
        totalWeight = partitionWeights.values().stream().mapToInt(Integer::intValue).sum();
    }

    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();

        if (keyBytes == null) {
            // 无键时使用粘性策略
            return StickyPartitionCache.partition(topic, cluster);
        }

        // 基于 Key 哈希 + 权重映射
        int hash = Utils.toPositive(Utils.murmur2(keyBytes));
        int weightedIndex = hash % totalWeight;
        int cumulativeWeight = 0;

        for (Map.Entry<Integer, Integer> entry : partitionWeights.entrySet()) {
            cumulativeWeight += entry.getValue();
            if (weightedIndex < cumulativeWeight) {
                return entry.getKey();
            }
        }
        return hash % numPartitions;
    }

    @Override
    public void close() {}
}

4.4.3 生产者配置中注册自定义分区器

Properties props = new Properties();
props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, 
          "com.example.WeightedPartitioner");
// 传递自定义配置
props.put("partition.weight.0", "3");
props.put("partition.weight.1", "2");
props.put("partition.weight.2", "1");

4.5 分区策略对比与选型

4.6 分区路由流程图

下图展示了生产者发送消息时分区路由的完整决策路径：

4.7 分区策略选择决策树

在实际生产环境中，可根据以下决策树快速选择合适的分区策略：

4.8 生产实践建议