“分区器不就是一个hash取模嘛，有啥好写的？”如果你这么想，多半是还没碰到真正需要自定义分区的场景。我在生产环境吃过一次大亏：多租户数据全挤在少数几个分区上，下游消费根本扛不住，集群热点报警一条接一条。从那时候起，我才认真把Kafka生产者的Partitioner API从头到尾撸了一遍，也自己动手写过好几个版本。今天就把这部分的源码逻辑、实现要点和踩坑记录一次讲清楚。
这篇文章适合两类人：一类是刚接触Kafka、想知道partitioner.class背后到底发生了什么的新手；另一类是已经用默认分区器跑了一阵子，但遇到数据倾斜、多租户路由、按地域聚合等需求，正准备自己实现Partitioner接口的开发者。我会从接口源码开始拆，再给出一个可以直接抄作业的实现，最后把我实际排查过的问题列成清单。
1. 分区器在生产者链路中到底处于什么位置
1.1 一条消息从send()到落盘经历了什么
很多人用Kafka发送消息，只看到producer.send(new ProducerRecord<>(topic, key, value))这一行，以为消息就是直接“发”出去的。实际上从你调用send()到消息真正到达broker，中间要过好几道工序：拦截器链、序列化器、分区器、累加器、Sender线程，最后才通过网络发出。
把这条链路拆开看，顺序是这样的：
- 拦截器（Interceptor）对ProducerRecord做前置处理，比如追加公共字段、做埋点。
- 序列化器（Serializer）把key和value从对象转成字节数组：这一步就是
key.serializer和value.serializer干的活。 - 分区器（Partitioner）根据key的字节数组、value的字节数组以及当前topic的元数据，计算出一个分区号。
- 累加器（RecordAccumulator）把消息封装进ProducerBatch，放到对应分区的双端队列里。
- Sender线程从队列里取出攒好的批次，发往对应的broker。
也就是说，分区器不是负责“把消息发到哪个broker”的，它只负责决定“这条消息应该进哪个分区编号”。真正把分区号映射到broker地址的，是Kafka的元数据机制。分区器返回的是一个int，比如5，那这条消息就会进入Topic的第5号分区，至于第5号分区的leader在哪个broker上，跟你没直接关系，Producer会自动查元数据。
1.2 partition()在哪个环节被调用，参数从哪来
关键调用发生在KafkaProducer.doSend()方法内部。在序列化完成之后、写入累加器之前，会执行类似这样的一段逻辑：
int partition = this.partitioner.partition( record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster);注意这里传入的serializedKey和serializedValue是已经经过序列化器处理的字节数组，而record.key()和record.value()是原始对象。所以你的分区器实现里，既可以直接看原始key对象，也可以直接对key的字节数组做哈希，这完全取决于你的需求。
还有一个容易忽略的点：在调用分区器之前，doSend()会先调用waitOnMetadata()。这个方法会确保当前Producer的元数据缓存里已经有这个topic的分区列表，如果没有，它会触发一次metadata请求并等待结果。所以正常情况下，你的分区器在.partition()里拿到的cluster.partitionsForTopic(topic)不会返回null。但我在自定义分区器里仍然习惯做一次判空，后面会解释为什么。
1.3 分区结果直接影响什么
分区号决定了三件大事：
第一，消息落在哪个分区，直接决定了同一批下游消费者能否按照你的业务规则聚合数据。比如按用户ID哈希分区，就能保证同一个用户的所有订单消息都进同一个分区，下游消费时天然按用户聚合。
第二，分区号影响Producer端batch的组装。Kafka的吞吐之所以高，很大程度上依赖批量发送。如果你的分区器让消息过于分散，每个分区的batch都攒不满，linger.ms还没来得及发挥作用就不得不发出去，那整个发送性能会肉眼可见地下降。
第三，分区结果跟幂等、事务的机制也有关系。开启幂等后，Producer会为每个分区维护序列号，如果分区路由不稳定或者分区数变化，会影响序列号的管理和broker端的去重判断。所以分区器不是随手写写就完事的，它牵一发动全身。
2. 逐行拆解Partitioner接口与默认实现
2.1 接口里那四个方法到底干什么
打开org.apache.kafka.clients.producer.Partitioner接口，你会看到这样的签名：
public interface Partitioner extends Configurable, Closeable { int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster); void close(); default void onNewBatch(String topic, Cluster cluster, int prevPartition) {} }先说你可能忽略的背景：Configurable和Closeable是两个父接口。Closeable好理解，Producer关闭时会调用close()，你可以在这里释放资源。Configurable带来的configure(Map<String, ?> configs)方法也很重要，Kafka会把你配置文件里所有带prefix的Producer配置原样传进来，让分区器能读取自定义参数。
再说三个核心方法：
partition()是主入口，每条消息发送时都会调用。它的返回值为int，代表分区号。根据Kafka的约定，如果返回负数会直接抛异常。这个方法的六个参数里，topic是主题名，key和value是序列化前的原始对象，keyBytes和valueBytes是序列化后的字节数组，cluster是当前元数据快照，里面封装了所有topic的分区信息。
onNewBatch()是Kafka 2.4新增的。它的触发时机是：当累加器为某个分区新创建一个ProducerBatch时，会回调一次。这个回调的意义是什么呢？用一句话说：它给了分区器一个“切换分区”的机会。默认的粘性分区（sticky partition）策略就是靠它实现的，在无key消息的场景下，Producer会一直把一个分区粘到底，直到当前batch满了或超时，才在onNewBatch()里切换到下一个分区。
close()不用多说，Producer关闭、分区器即将被回收前调用一次。如果你的分区器持有连接池、缓存了某些资源，记得在这里释放。
2.2 DefaultPartitioner有key和无key的两条路
默认分区器DefaultPartitioner的逻辑，说穿了就两条分支。
当keyBytes != null时，走哈希取模：
return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;这里有两个点值得注意。第一是Utils.murmur2()，这是Kafka自己实现的MurmurHash2算法，seed是一个固定值。用同样的key和同样的分区数，算出来的分区号永远是确定的。第二是Utils.toPositive()，这个工具方法的作用是把murmur2返回的int的符号位去掉，转成正数，因为murmur2的结果可能是个负数，直接取模会得到负分区号。
当keyBytes == null时，走粘性分区策略。这里不是每条消息随机选分区，而是查缓存：
return stickyPartitionCache.partition(topic, cluster);StickyPartitionCache内部维护了一个Map<String, Integer>，记录每个topic当前正在使用的分区号。如果缓存里没有，就随机选一个放进去。接下来持续使用这个分区号，直到新batch创建时onNewBatch()回调触发，才会计算下一个分区。
为什么社区要把无key消息从“每条随机”改成“粘性分区”？我举个例子你就懂了。假设分区数很多，如果每条无key消息都随机落在不同分区，每个分区的batch都只有一两条，攒不满就得发出去。结果就是小请求不断，网络往返增加，broker端的写入放大严重。改成粘性策略后，一批消息会集中在同一个分区，batch能攒得更满，压缩率更高，整个Producer的吞吐量会明显提升。
2.3 新版接口的变化迹象
你可能在网上搜到过老版本的分区器代码，有人重写partition()时用的是List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);，然后自己遍历。在Kafka 3.x的某些小版本里，partition()的签名还加过一个Cluster resourceCluster之类的参数，这其实是为Kafka的鉴权资源和元数据隔离做的扩展。我的建议是：你要以项目里实际引入的kafka-clientsjar包源码为准，不要照抄网上的签名。写代码之前，先在你的IDE里打开Partitioner.class看一眼再动手，免得编译期就报签名不匹配。
3. 手写一个订单按城市分区的生产级分区器
3.1 先想清楚需求，别一上来就写代码
假设现在有个电商订单系统，订单消息要发给Kafka。下游有一堆消费者，有的按城市维度做物流统计，有的按城市做实时大屏。如果用默认的key哈希，因为订单号是全局唯一的，哈希结果会把城市打散，同一个城市的订单会分布在各个分区里，下游每个消费者都要消费全部分区才能聚合出城市维度数据，处理效率很低。
这时候合理的做法是自定义分区器：保证同一个城市的订单进入同一个分区，不同城市的数据相对均匀地分布在多个分区上，并且当新增分区时，尽量不影响已有规则的可解释性。
我把这个需求拆解成三点：
- 能从订单里提取出城市维度。订单的key设计成了
城市编码_订单号，比如1001_ORD20250101001，城市编码是前四位。 - 同一个城市必须路由到同一个分区。
- 不同城市之间要尽量均匀分布，避免哈希碰撞严重倾斜。
3.2 完整实现代码
基于上面的需求，我写了一个可以直接用的小例子：
package com.example.kafka.partitioner; import org.apache.kafka.clients.producer.Partitioner; import org.apache.kafka.common.Cluster; import org.apache.kafka.common.PartitionInfo; import java.util.List; import java.util.Map; import java.util.concurrent.ThreadLocalRandom; public class CityOrderPartitioner implements Partitioner { private static final int CITY_CODE_INDEX = 0; private static final int CITY_CODE_LENGTH = 4; // 触发新建batch时是否切换分区，默认开启 private boolean stickyEnabled = true; @Override public void configure(Map<String, ?> configs) { Object sticky = configs.get("city.partitioner.sticky.enabled"); if (sticky != null) { stickyEnabled = Boolean.parseBoolean(sticky.toString()); } } @Override public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { List<PartitionInfo> partitions = cluster.partitionsForTopic(topic); // 防御性判断：正常调用链不会为空，但在异常场景下避免NPE if (partitions == null || partitions.isEmpty()) { throw new IllegalStateException("topic partition metadata is empty: " + topic); } int numPartitions = partitions.size(); if (key == null || keyBytes == null) { return nextPartition(numPartitions); } String keyStr = key.toString(); // 只取前四位城市编码 String cityCode = keyStr.substring(CITY_CODE_INDEX, Math.min(CITY_CODE_LENGTH, keyStr.length())); int cityHash = cityCode.hashCode() & 0x7fffffff; return cityHash % numPartitions; } private int nextPartition(int numPartitions) { // 无key场景使用随机分区 return ThreadLocalRandom.current().nextInt(numPartitions); } @Override public void onNewBatch(String topic, Cluster cluster, int prevPartition) { // 如果你有粘性分区需求，可以在此实现切换逻辑 // 这里保持空实现，无key消息始终随机 } @Override public void close() { // 不需要释放资源 } }这个实现有几个关键点：
我把无key消息的路由逻辑直接放进了nextPartition()，用ThreadLocalRandom生成随机分区号。为什么不用Utils.murmur2那种静态哈希？因为无key消息本身没有任何业务含义，它的目的就是均匀分散，随机是合理的。
cityCode.hashCode() & 0x7fffffff这一步，相当于把Java String的hashCode转成正数再取模，避免负数分区号。这里没有用Kafka的Utils.toPositive()，是为了让这个类不依赖Kafka内部类也能单独测试，如果你项目里能访问到Utils，直接用也完全没问题。
3.3 配置接入与验证方法
写完分区器后，在Producer配置里加一行：
props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.example.kafka.partitioner.CityOrderPartitioner"); props.put("city.partitioner.sticky.enabled", "true");注意第二行，自定义配置不需要在ProducerConfig里有预定义，Kafka会原样传给configure()。这是分区器扩展里很实用的能力，你可以通过配置来动态控制分区器行为，而不需要改代码重新发版。
验证分区器是否生效，最直接的方法是写一个测试生产者和一个测试消费者。生产者的消息key带上城市编码前缀，消费者的record.partition()可以直接打印分区号。我建议你故意构造一批不同城市编码的key，发给一个3分区或6分区的topic，然后肉眼观察每个分区收到的城市分布是否符合预期。
另外还有一个更彻底的验证手段：临时把partitioner.class改成你自己的类，然后看Producer日志里的Created new producer输出，确认加载的类名确实是你写的那个。如果类名没变，通常是你配置文件写错或者类没有打进依赖里。
3.4 线程安全与性能要注意
这部分是最容易翻车的。KafkaProducer本身是线程安全的，允许从多个线程并发调用send()。这意味着同一个Partitioner实例也会被多个线程同时调用。你的partition()方法必须是无状态的，或者读写共享状态时必须保证线程安全。
我见过有人把HashMap直接当成缓存用在分区器里，上线后偶发数据路由错误，排查了大半天才定位到是并发写入导致HashMap死循环。我的建议很简单：要么像上面这个例子一样，所有共享数据在启动阶段就初始化成只读的；要么用ConcurrentHashMap，并且不要在partition()里做重量级更新操作。
性能方面也要注意。partition()是每条消息发送路径上的必经之路，它的耗时直接叠加在发送延迟上。不要在分区器里做网络请求、读文件、远程调用这类事。如果确实需要加载用户映射表或城市表，在configure()阶段加载到内存就行，不要每来一条消息就去查一次数据库。
4. 实际项目中踩过的坑与排查技巧
4.1 自定义分区器没生效，问题出在哪
最典型的症状是：分区器明明写了、配置也挂了，但消息的分区分布跟默认行为没有任何差别。我遇到过一次，最后发现是classpath里同时存在两个不同版本的kafka-clientsjar，导致Partitioner接口签名不同，加载时直接走了老版本的默认实现。
排查思路是三个方向：第一，确认你的Producer实际加载的依赖版本；第二，在configure()里加一条日志，如果这个日志没打出来，说明分区器压根没被初始化；第三，用mvn dependency:tree或gradle dependencies查一下有没有多余的传递依赖把旧版本的客户端带进来。
4.2 消息全跑到一个分区，数据倾斜了
自定义分区器的常见问题就是分区分配极度不均匀。尤其当你拿业务字段直接取模，而业务字段的分布本身就不均匀时，比如订单id都带同一个前缀，哈希全撞一起，结果就是某些分区被压满，其他分区很闲。
解决思路有两个。一个是引入“二次哈希”，先用业务字段算出一个中间值，再用这个中间值取模分区；另一个是使用一致性哈希环，让分区数量变化时路由结果尽量稳定。我自己在多个项目里的经验是：如果下游不在乎某个业务维度的数据必须严格连续，用二次哈希就好；如果下游有严格的数据局部性要求，比如同一城市必须进同一分区，那宁可牺牲均匀性也要保证这个约束，后续靠扩容分区数来缓解。
4.3 key为null时的行为跟你想象的不一样
如果你在自定义分区器里对key为null的消息用了类似key.hashCode()的代码，那会直接触发NullPointerException。默认分区器对key为null是走粘性策略的，不管你就意味着丢失这部分消息的路由能力。
我在自己的分区器里对null key做了单独处理，用随机分区兜底。但也见过一种做法：把null key统一路由到某一个固定分区，这样方便后面排查异常数据，缺点是可能造成单一分区热点。怎么取舍取决于你的业务，我建议常规情况下就用随机，只有在调试阶段才临时打到固定分区。
4.4 分区扩容后路由错乱怎么办
这个话题经常被忽略。假设你的topic原来6个分区，key哈希取模后落在hash % 6。某天你扩容到了12个分区，同一批key的取模结果大概率全部变了，对应的分区号跟扩容前完全对不上。如果下游是按照分区号做了状态存储，扩容之后就会读不到老数据。
这里没有完美的解决方案，常规做法是三种：一是接受路由变化，下游在扩容窗口期做数据迁移或重放；二是采用一致性哈希，让扩容时只有一部分key受影响；三是把分区数直接规划到位，前期不要频繁扩容。我在生产项目里一般选第二种，自定义分区器接一个一致性哈希环，虽然代码量多一点，但后续扩容的心理负担小很多。
4.5 onNewBatch的副作用
onNewBatch()是个很容易被误解的方法。很多人以为它是“每来一批消息就调用一次”，于是把分区切换逻辑写在这里，结果发现分区切换过于频繁，或者某些逻辑被重复触发。
它的真实触发时机是：累加器新创建一个ProducerBatch，同时这个batch将要使用你分区器返回的分区号之前。如果你在这个方法里做了重量级操作，等于在每条消息攒batch的路径上加了额外开销。如果不需要粘性分区，保持空实现即可，不需要为了“看起来完整”硬写逻辑。
5. 效果验证与生产落地建议
5.1 怎么确认你的分区器工作正常
我把验证方法分成三个层次，从简单到复杂：
第一层是控制台验证。用kafka-console-consumer消费一个测试topic，打开--property print.partition=true，直接看每条消息落在哪个分区。像我的城市分区器，可以用不同城市编码的key发一批消息，肉眼检查同城市是否同分区。
第二层是埋点验证。在分区器里临时加一行日志，打印topic、key、返回的分区号，跑一小段时间，统计一下分区分布是否均匀。注意这个日志上线时要关掉，否则每条消息都打日志，性能直接崩掉。
第三层是监控验证。如果集群已经接入了监控系统，重点看每个分区的写入速率和Producer端的request指标。正常情况下各分区写入速率应该是相对均衡的，如果某个分区的速率明显高于其他分区，说明你的分区规则可能有问题。
5.2 分区器与序列化器、幂等、事务的搭配
有一些搭配细节是新手容易踩的。
第一个细节是keyBytes的来源。分区器拿到的keyBytes是key经过key.serializer序列化后的结果，所以如果你在分区器里直接用keyBytes做哈希，本质上跟你的key序列化方式无关，任何key类型都能算。如果在分区器里读key.toString()，那你就假设了key是String类型，一旦key类型换成别的对象，这行代码就得改。默认分区器只用keyBytes，原因就是它不关心key的具体类型，只要字节数组参与哈希就行。
第二个细节是幂等Producer对分区器没有特殊要求，但对分区路由的稳定性有要求。幂等机制靠broker端的序列号做去重，序列号是跟“Producer ID + 分区号”绑定的。如果你的自定义分区器对同一批key返回了不稳定结果，序列号就会出现跳跃，轻则影响顺序交付，重则触发OutOfOrderSequenceException。
第三个细节是事务机制下不要随意在业务代码里手动指定分区。事务Producer发送消息时，如果业务里既指定了分区号，又用了自定义分区器，会造成混乱。我的建议是二选一：要么统一走分区器，要么统一在ProducerRecord里显式指定分区号，别混着来。
5.3 不是所有场景都需要自定义分区器
这篇文章讲了怎么手写分区器，但我也想泼一盆冷水。如果你的需求只是“不同key均匀分布”，默认分区器已经做得足够好，不需要自己写。如果你的需求是“同一个key必须进同一个分区”，默认分区器也能满足，它本来就是按key哈希取模。
真正需要自定义分区器的场景，通常是这几类：
- 按业务维度强制聚合，比如同一城市、同一用户、同一租户必须进同一分区。
- 需要兼容历史数据，比如之前已经按某个规则分了区，新老代码要平滑切换。
- 有冷热数据之分，想把高优先级数据单独导到某个分区，方便下游特殊处理。
- 想绕过默认的粘性策略，用轮询或自定义均匀算法替代。
换句话说，先想清楚为什么默认的不够用，再决定要不要动手写。
我在实际项目里测试过多个版本的分区器，一个很实在的体会是：分区器的代码量通常不大，难点全在对调用链的理解和对并发模型的认识上。很多人写出的功能正确，但上线后被并发问题或性能问题反噬。把partition()看成一条热搜路径，把configure()看成唯一的一次初始化机会，把onNewBatch()看成事件回调而不是业务入口，这个思维框架建立起来，你写分区器就不容易跑偏。
最后再分享一个小技巧：自定义分区器一定要单测。Cluster对象可以用Kafka测试工具包构造，或者直接用mock数据构造一个包含几个PartitionInfo的实例，测一下不同key的分区结果是否符合预期。分区规则这种代码，越早发现问题代价越小，别等上了生产、数据已经铺开再来改。
