《Java 消息中间件·Kafka 内核与客户端》3.3 分区器与键设计

默认分区器已经不是一个类

如果你按 3.x 的印象去找 org.apache.kafka.clients.producer.internals.DefaultPartitioner，在 4.3.0 的 sources jar 里会找不到。实际列出来的分区相关文件只有三个：

org/apache/kafka/clients/producer/RoundRobinPartitioner.java
org/apache/kafka/clients/producer/internals/BuiltInPartitioner.java
org/apache/kafka/clients/producer/Partitioner.java

没有 DefaultPartitioner.java。默认逻辑被合并进了 BuiltInPartitioner，而它甚至不实现 Partitioner 接口。源码注释第一句就说明了：

/**
 * Built-in default partitioner.  Note, that this is just a utility class that is used directly from
 * RecordAccumulator, it does not implement the Partitioner interface.
 */

这不是命名游戏，它改变了你排查问题的入口。自定义分区器走 Partitioner 接口，默认逻辑走 RecordAccumulator 内部的另一条路径。两者在 KafkaProducer.partition() 里分叉。

分区选择的优先级

KafkaProducer.partition()（第 1469 行）的完整逻辑只有四步：

private int partition(ProducerRecord<K, V> record, byte[] serializedKey, byte[] serializedValue, Cluster cluster) {
    if (record.partition() != null)
        return record.partition();

    if (partitionerPlugin.get() != null) {
        int customPartition = partitionerPlugin.get().partition(
            record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster);
        if (customPartition < 0) {
            throw new IllegalArgumentException(String.format(
                "The partitioner generated an invalid partition number: %d. Partition number should always be non-negative.", customPartition));
        }
        return customPartition;
    }

    if (serializedKey != null && !partitionerIgnoreKeys) {
        // hash the keyBytes to choose a partition
        return BuiltInPartitioner.partitionForKey(serializedKey, cluster.partitionsForTopic(record.topic()).size());
    } else {
        return RecordMetadata.UNKNOWN_PARTITION;
    }
}

优先级是：

1. 显式指定了 partition，直接用；
2. 配了 partitioner.class，调用自定义分区器；
3. 有 key 且没开 partitioner.ignore.keys，按 key 哈希；
4. 其余情况返回 UNKNOWN_PARTITION，交给 RecordAccumulator 的粘性逻辑。

第 3 步的哈希函数是公开的：

public static int partitionForKey(final byte[] serializedKey, final int numPartitions) {
    return Utils.toPositive(Utils.murmur2(serializedKey)) % numPartitions;
}

murmur2 的结果取正再对分区数取模。这里有一个容易忽略的点：分区数变化会改变几乎所有 key 的落点。从 6 个分区扩到 7 个，同一个 key 的 % 7 结果和 % 6 不同，原本同分区的两条消息可能被分到不同分区。扩分区不是无代价操作，第 11.3 节会展开。

key 为 null 时的粘性行为

返回 UNKNOWN_PARTITION 之后，RecordAccumulator 用 BuiltInPartitioner 选一个分区并「粘住」，直到这个分区累计写入的字节数达到阈值。

BuiltInPartitioner.updatePartitionInfo 的核心判断：

if (producedBytes >= stickyBatchSize && enableSwitch || producedBytes >= stickyBatchSize * 2) {
    // We've produced enough to this partition, switch to next.
    StickyPartitionInfo newPartitionInfo = new StickyPartitionInfo(nextPartition(cluster));
    stickyPartitionInfo.set(newPartitionInfo);
}

stickyBatchSize 就是生产者的 batch.size。源码注释解释了为什么有 * 2 的上限：高 linger.ms 时，如果批次还没准备好就切换分区，会产生 12 KB、4 KB 这样的碎片批次；允许在 1 倍到 2 倍之间切换，可以把批次边界对齐。

这意味着：key 为 null 时不是轮询。消息会一段一段地落在同一个分区，每段的大小由 batch.size 决定。默认 16 KB 意味着低流量下可能几百条消息都在一个分区，然后突然换到下一个。

实测：null key 的粘性序列

6 个分区、batch.size=1024、linger.ms=0、同步发 300 条。程序记录每条消息的 RecordMetadata.partition()，输出前 40 条：

mode=null count=300
first40=3,3,3,3,3,3,3,3,3,3,3,3,3,3,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2,2
p3=28 p2=93 p4=74 p1=40 p5=39 p0=26

前 14 条全在 p3，接着至少 26 条全在 p2。300 条的最终分布是 26 到 93，最大/最小比约 3.6 倍。

这组数据要表达两件事：

1. 短窗口内严重不均：前 40 条只用了 2 个分区，另外 4 个一条没有；
2. 长窗口内也不保证均匀：300 条已经不少，仍然有 3.6 倍差距。

这是粘性分区的设计目标——用短时不均换更大的批次——不是 bug。如果你的业务要求「每条消息尽快分散到所有分区」，粘性会让部分分区短暂空闲；如果要求吞吐，粘性是对的。

实测：key 的分布

同样 6 个分区，换成 key=book-0 到 book-299：

mode=keyed count=300
p4=43 p2=45 p5=41 p0=56 p1=63 p3=52

41 到 63，最大/最小比约 1.5 倍。这是 300 个样本下的正常波动，没有明显倾斜。

这个结果值得单独说，因为它反驳了一个常见说法：「单调递增的 key 会倾斜」。book-0、book-1 是单调的，但 murmur2 把它们打散了。单调本身不是问题，问题是 key 的分布和基数。

把 key 基数降到 5：

mode=lowcard count=300
first40=4,2,5,0,1,4,2,5,0,1,4,2,5,0,1,4,2,5,0,1,4,2,5,0,1,4,2,5,0,1,4,2,5,0,1,4,2,5,0,1
p4=60 p2=60 p5=60 p0=60 p1=60

5 个 key 只落在 5 个分区，p3 一条没有。key 基数小于分区数时，必然有分区闲置。这不是哈希不均，是信息量不够：5 个不同的 key 最多产生 5 个不同的哈希值。

再把 80% 的流量压到一个 key：

mode=skew count=300
p5=251 p1=13 p2=9 p4=5 p3=11 p0=11

book-bestseller 这一个 key 落在 p5，带走 251 条（83.7%）。这才是真正的热点倾斜：不是 key 单调，是 key 的流量集中。

自定义 Partitioner

当默认规则不满足需求时，实现 Partitioner 接口。4.3.0 的接口很小：

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

Configurable 带来 configure(Map<String, ?> configs)，Closeable 带来 close()。一个把 hot- 前缀固定路由到 0 号分区的实现：

public final class HotKeyPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        int partitions = cluster.partitionsForTopic(topic).size();
        String k = key == null ? "" : key.toString();
        if (k.startsWith("hot-")) {
            return 0;
        }
        return 1 + Math.floorMod(k.hashCode(), Math.max(1, partitions - 1));
    }

    @Override
    public void configure(Map<String, ?> configs) {
    }

    @Override
    public void close() {
    }
}

配置方式是 partitioner.class=lab.HotKeyPartitioner。实测 300 条、其中一半 key 带 hot- 前缀：

mode=hot count=300
p0=150 p1=30 p3=30 p5=30 p2=30 p4=30

150 条精确落在 p0，其余 150 条在 5 个分区上各 30 条。自定义分区器的行为完全由代码决定，Kafka 只做一件事：检查返回值不小于 0。源码里没有上界检查，返回 partitions 或更大的值会走到更深处才失败。写自定义分区器时，自己保证返回值在 [0, partitions) 内。

两个边界要注意：

• 自定义分区器与自适应分区互斥。KafkaProducer 构造时，只有 partitionerPlugin.get() == null 才会启用 adaptive partitioning；配了自定义分区器，partitioner.adaptive.partitioning.enable 不生效。
• partitioner.ignore.keys 对自定义分区器无效。源码文档写明「this setting has no effect if a custom partitioner is used」。忽略 key 的逻辑在默认路径里，不在接口调用前。

有 key 但想忽略 key

有些场景 key 只是用来做审计或路由标记，不希望它影响分区。4.3.0 提供了 partitioner.ignore.keys=true：

mode=ignore-keys count=300
first40=0,0,0,...,5,5,5,...,4,4,4,...0
p0=50 p5=50 p4=49 p1=30 p3=61 p2=60

序列同样是成段的，因为忽略 key 之后就退回粘性逻辑。分布 30 到 61，和 null key 的形态一致。

这个配置在 3.x 就有了，但很多人不知道。它的价值在于：不必为了「保留 key 做业务标识」而牺牲分区均衡。

RoundRobinPartitioner 与它的坑

如果确实需要逐条轮询，可以显式配：

partitioner.class=org.apache.kafka.clients.producer.RoundRobinPartitioner

它是 4.3.0 里仍然保留的显式实现。但 ProducerConfig.PARTITIONER_CLASS_DOC 自己写了一个警告：

Note: There's a known issue that will cause uneven distribution when a new batch is created.
See KAFKA-9965 for more detail.

也就是说，轮询分区器在批次边界上仍然可能不均。它不是「严格轮流」的保证。本机没有单独测这个类，不给数字；要用的话，先在自己的流量形态下看分区计数。

key 设计的四条原则

第三行是唯一需要业务妥协的。加盐之后，「同一本书的事件有序」不再成立，消费端要么按盐分别处理，要么引入版本号做冲突解决。这个取舍属于卷二和卷三，本节只把机制摆出来。

怎么验证分区落点

最直接的方法是在客户端读 RecordMetadata：

RecordMetadata md = producer.send(new ProducerRecord<>(topic, key, value)).get();
System.out.println("partition=" + md.partition() + " offset=" + md.offset());

批量验证时不要逐条打印，把分区号累加到 Map<Integer, Integer>，最后输出计数。本节所有分布数据都是这么来的。

服务端侧可以用 kafka-dump-log.sh 看每个分区目录里的记录数和 key，或者用 kafka-consumer-groups.sh --describe 看每个分区的积压。前者适合确认「消息到底写到了哪」，后者适合确认「消费有没有偏」。

回到 libraryd

libraryd 的六个主题里，五个用 bookId 作 key，只有 library.loan.overdue 用 loanId。这个设计是有意的：

• 借阅、归还、预约、履约、库存调整都围绕同一本书，用 bookId 保证同一本书的事件顺序；
• 逾期针对的是一笔借阅，用 loanId 可以让逾期事件独立分区，不受热门书影响。

代价是热门书的单分区上限。如果某本书一天借出几千次，它的所有事件都落在同一个分区，这个分区会成为瓶颈。解决办法不是简单加分区——加分区会改变所有 key 的落点，破坏已有的顺序假设。更稳妥的做法是给热门 bookId 加后缀拆流，并在消费端按业务键合并，这需要卷二讨论的去重和顺序处理配合。

下一节离开客户端，进入服务端：KRaft 怎么把「哪个 broker 是 leader」这件事变成一条可以复制的日志。

阅读导航：上一节 3.2 acks、重试与幂等 ｜ 下一节 4.1 副本、LEO 与高水位