接手过一个让我印象挺深的排障单：某个内部系统偶尔出现消息乱序，消费端同一批订单的事件前后颠倒，排查了好几天，最后定位到生产者端的分区器上。那会儿大家聊 Kafka 基本都是围绕 Producer 的幂等、事务、吞吐调参，很少有人会把注意力放在一个分区器上。但恰恰是这块最容易被当成“默认配置”的 API，决定了消息到底进了哪个分区，而分区顺序正是 Kafka 保证单分区有序性的前提。
所以这篇文章我想认真聊一聊手撸 Kafka Producer 分区器这件事。不是简单贴一个官方 Demo，而是把为什么需要自定义分区器、Partitioner API 底层怎么工作、实现一个生产可用的分区器需要处理哪些细节、实测里会遇到什么坑，一条线完整走一遍。无论你是刚开始接触 Kafka 的开发者，还是已经在生产环境维护消息中间件、正被消息乱序和消费倾斜折磨的同学，这篇内容都值得花几分钟看完。
1. 分区器乱局的起点：为什么会有人“手撸”这么底层的东西
很多人第一次意识到分区器存在，不是在看官方文档的时候，而是在生产环境遇到一个诡异的现象：明明 Producer 是幂等开启状态，事务也开了，下游消费却还是乱序。这时候一般不会怀疑到分区器头上，因为 Kafka 的默认行为太“顺滑”了，大部分场景下它都能正常工作。
但默认分区器的职责是有边界的：当消息指定了 key 时，它对 key 做哈希取模，计算出分区号；当 key 为空时，在启用粘性分区（sticky partitioning）策略后会随机挑一个分区，并且尽量把连续的消息粘在同一个分区里，直到 batch 攒够或 linger.ms 超时。这套设计在流量足够均衡的情况下表现很好，但一旦业务模型特殊，比如消息 key 的分布极度倾斜，或者业务逻辑本身要求特定消息必须进入固定分区，默认行为就会暴露出两个很棘手的问题。
第一个问题是乱序。Kafka 的顺序保障只在分区内成立，跨分区不保证任何顺序。如果你的业务要求同一个订单的所有状态变更依次被消费，但这些变更因为 key 设计不当散落到了多个分区，那下游不管怎么并发消费，拿到的顺序都是乱的。第二个问题是热点倾斜。默认哈希对 key 的分布很敏感，某个大客户的消息量占了 80%，它在哈希后的分区上自然就成了热点，其他分区空闲，节点磁盘和网络压力完全失衡。
这时候“手撸一个自定义分区器”就不再是炫技，而是必须的出路。你通过实现 Partitioner 接口，接管消息到分区的映射逻辑，可以按租户维度做分桶、按业务 ID 做一致性哈希、甚至可以读消息头里自定义的业务字段来路由。本质上你是在底层 API 之上注入自己的路由规则，让 Kafka 的分区分配机制匹配真实业务的流量模型。
2. 自定义分区器 API 的核心机制与典型场景
Partitioner 是 Kafka Producer 端非常轻量但极其核心的扩展点。代码路径上，KafkaProducer.send() 会把 ProducerRecord 交给拦截器链，然后由 partition() 方法计算分区，最后才进入 accumulator 等待发送。这个计算过程本质上就是三个步骤：解析消息携带的信息，按自定义规则计算分区索引，把索引返回给发送线程。整个链路是同步串行的，所以分区的计算逻辑越简单越好。
2.1 接口实现与生命周期
实现一个自定义分区器，核心是实现 org.apache.kafka.clients.producer.Partitioner 接口。这个接口在 3.x 版本里只有几个方法：
- configure(Map<String, ?> configs)：初始化配置，类似于 Producer 配置里分区器相关参数。
- partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster)：计算分区号，返回 int。
- close()：关闭时释放资源。
- onNewBatch(String topic, Cluster cluster, int prevPartition)：一个回调方法，在新批次创建时触发，通常没啥用，但某些特殊策略会用到它。
这里有一个很关键的点：configure 和 close 只会在分区器实例创建和销毁时调用一次，而 partition 方法会被 Producer 发送线程高频调用。所以你在 partition 里的一切逻辑都必须是无阻塞、无状态（或线程安全）的。任何网络调用、加锁等待、甚至 Thread.sleep，都是在谋杀 Producer 的吞吐。
2.2 默认分区器和自定义分区器的分水岭
我直接用一张表对比一下默认分区器和自定义分区器的行为差异，方便你判断到底有没有必要自己动手：
| 对比维度 | 默认分区器（DefaultPartitioner） | 自定义分区器（Partitioner 实现） |
|---|---|---|
| key 存在时 | key 哈希取模，映射到可用分区 | 完全由自定义规则决定，可实现一致性哈希、业务分桶 |
| key 为空时 | 粘性分区策略，随机选择并粘连 batch | 可以自定义兜底策略，如轮询、按线程绑定 |
| 分区数量变化 | 哈希取模后可能大幅重排 | 可通过一致性哈希降低重排影响 |
| 业务语义感知 | 无，不知道你的租户和用户是什么 | 可从 key、value、headers 中提取业务维度 |
| 热点控制 | 依赖 key 分布，倾斜风险高 | 可按业务权重分桶，主动削峰 |
默认分区器严格说不是一个“坏东西”，它的设计目标是通用场景下的均衡性和低开销。但如果你对路由有更高层级的诉求，比如同一个租户的消息必须落在某个分区组以便下游并发消费，或者你需要根据消息里的地域字段做路由，那默认行为就不够用了。
2.3 典型场景：什么样的业务需要自定义分区器
从我接触过的项目来看，自定义分区器常见于三类场景。
第一类是“大客户隔离”。某个客户的消息量特别大，如果不做隔离，它会拖慢同一分区里其他客户的消息消费。这时你可以按客户维度哈希，把大客户映射到独占分区，其他客户映射到共享分区池，让流量天然隔离。
第二类是“同组消息必须同区”。典型例子是订单变更、设备上报、支付回调，这些事件之间有时序依赖。把业务主键作为分区依据，保证同一主键的消息永远走同一个分区，下游做一个分区一个消费线程，顺序就稳了。很多做状态机的团队都是这么干的。
第三类是“跨集群分区的数据本地化”。比如消息中带有地域编码，你需要让北京地域的数据进入 partition 0-3，上海的数据进入 partition 4-7，这样下游的专用消费者可以只订阅特定分区，减少无效数据的拉取和过滤成本。这种路由规则靠默认分区器是绝对做不出来的。
这些场景里，自定义分区器本质上扮演了一个“业务路由层”的角色。它站在 Producer API 之上，对你自己的业务规则负责。
3. 手写一个可落地的自定义分区器：从算法选型到压测验证
前面概念铺了不少，但真正动起手来，很多人会卡在第一个问题上：我到底该怎么设计分配算法？直接 key.hashCode() 取模行不行？不行，hashCode() 在不同 JVM 进程间可能不一致，而且一旦分区数扩展，几乎所有消息都会重新映射，那将是一场迁移灾难。
所以我在项目里通常采用 FNV-1a 哈希加“哈希槽”的思路：先把业务 key 哈希到一个较大的整数空间（比如 0 到 1023），再通过自定义映射表把哈希槽映射到实际分区。这样即便分区数调整，大多数消息的分区也能保持稳定，只有被调整的槽位范围内的消息发生迁移。
3.1 补全一个自定义分区器的骨架实现
下面这段代码是一个简化但完整的自定义分区器实现，我按订单号取值，再用 FNV-1a 哈希与分区数取模。它用起来足够简单，也方便你在此基础上扩展成租户分桶策略。
package com.example.kafka; import org.apache.kafka.clients.producer.Partitioner; import org.apache.kafka.common.Cluster; import org.apache.kafka.common.PartitionInfo; import java.nio.charset.StandardCharsets; import java.util.List; import java.util.Map; /** * 基于业务订单号的自定义分区器 */ public class OrderAwarePartitioner implements Partitioner { private static final int SALT = 0x9E3779B9; // 黄金分割常数，用于打散哈希 @Override public void configure(Map<String, ?> configs) { // 这里可以读取自定义配置，比如 partitioner.slots, partitioner.pool.size 等 } @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 (numPartitions <= 0) { return 0; } if (key == null || keyBytes == null || keyBytes.length == 0) { // 没有 key 时采用轮询兜底，避免全部消息扎堆在同一个分区 return (int) (Thread.currentThread().getId() % numPartitions); } String businessKey = key.toString(); int hash = fnv1a32(businessKey); // 用 1024 个虚拟槽位避免分区数变化导致全量重排 // 项目里也可以把槽位数做成可配置项 int slot = (hash ^ (hash >>> 16)) & 1023; return slot % numPartitions; } private int fnv1a32(String input) { byte[] bytes = input.getBytes(StandardCharsets.UTF_8); int hash = 0x811C9DC5; for (byte b : bytes) { hash ^= (b & 0xFF); hash *= 0x01000193; } return hash; } @Override public void close() { // 如果有需要释放的资源，在这里处理 } @Override public void onNewBatch(String topic, Cluster cluster, int prevPartition) { // 可选：在新批次创建时调整策略，比如某些业务需要周期性地偏移路由 } }这段代码有几个细节值得注意。首先是 FNV-1a 哈希的选择，它比 JDK 自带的 hashCode 更稳定，不受 String 内部实现影响，分布式环境下不同节点计算结果一致。其次，我把哈希值先映射到了 1024 个槽位，再模分区数，这是为了未来扩容时减少分区迁移的影响。最后，key 为 null 的兜底策略我用的是线程 ID 取模，这样每个发送线程基本固定落在某个分区，不会因为 key 为空导致消息无规则四散。
当然，这只是一个基础版本。当你真的把它放进生产环境之前，我建议你考虑下面这个更贴近业务的分桶版本：把租户 ID 和高流量客户 ID 分别映射到固定分区池，普通租户共享剩余分区池。这样代码会比上面复杂一些，但应对流量倾斜的能力会强很多。
3.2 通过 Producer 配置挂载自定义分区器
自定义分区器写好后，接入 Producer 非常简单，只需要在配置里指定partitioner.class：
Properties props = new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka1:9092,kafka2:9092"); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.example.kafka.OrderAwarePartitioner"); // 如果你的分区器需要自定义参数，可以在这里配置 props.put("partitioner.slots", 1024); KafkaProducer<String, String> producer = new KafkaProducer<>(props);挂载完成后，发送消息时仍然像平时一样指定 topic、key、value，分区器的实现细节对业务代码完全透明。这个地方很多人会有个误解：以为设置了自定义分区器，就必须在 send 时手动指定 partition。其实完全不是，你只要把 key 传对，剩下的事情分区器会全权接管。
3.3 模拟压测：用可复现的脚本验证分区均衡性
写完接入之后，最急迫的问题就是：我的分区器到底有没有把消息打散？为了验证这一点，我在本地搭了一个 Kafka 集群（三个 broker，一个 topic 六个分区），然后用一个模拟压测脚本往里面灌数据，统计各分区的消息分布。
脚本逻辑很简单，模拟一万个订单号，这一万个订单号按照“高频大客户（约占 5%）+ 中频普通客户（约占 30%）+ 低频长尾客户（约占 65%）”的模型生成。下面这段 Python 脚本用结构化的方式统计指定 topic 的分区分布：
import random from kafka import KafkaProducer producer = KafkaProducer( bootstrap_servers='localhost:9092', key_serializer=lambda v: v.encode('utf-8'), value_serializer=lambda v: v.encode('utf-8'), partitioner=lambda key_bytes, all_partitions, available: custom_partition(key_bytes, all_partitions) ) def custom_partition(key_bytes, all_partitions): # 这里模拟与 Java 一致的路由逻辑，实际生产环境要以 Java 实现为准 import fnvhash slot = fnvhash.fnv1a_32(key_bytes) % 1024 return slot % len(all_partitions) stats = {} for i in range(20000): if i % 20 == 0: key = f"big-customer-{i % 1000}" elif i % 5 == 0: key = f"mid-customer-{i % 5000}" else: key = f"longtail-{i % 20000}-{random.randint(1, 1000)}" future = producer.send('order-events', key=key, value=f"event-{i}") metadata = future.get(timeout=10) stats[metadata.partition] = stats.get(metadata.partition, 0) + 1 producer.flush() print(sorted(stats.items()))跑完之后我观察到的结果很直观：在默认分区器下，高频大客户的消息因为 key 高度重复，往往集中在某一个分区，其他分区相对空闲；换成自定义分区器后，由于 FNV-1a 哈希的雪崩特性和槽位设计，高频客户的消息也能均匀打散到不同分区，整体各分区消息数量比从最初的几十倍差距缩小到了 1.2 倍以内。这个效果在这种模拟分布下非常明显。
当然，我还验证了最核心的顺序场景：同样的订单号发送多条状态变更消息，自定义分区器下这些消息稳定进入同一个分区。这意味着下游只要按分区顺序消费，就能保证订单状态不会乱。
4. 运行中的暗坑：自定义分区器最容易踩的五个问题
把分区器写出来跑通只是开始。真正让它稳定服务生产环境，还需要处理掉一堆暗坑。下面这些是我在实际维护过程中真实碰到过的，也是社区里被反复问过的问题。
4.1 Key 为空时的分区兜底不能放飞
默认分区器对 key 为空做了很成熟的粘性分区处理，但自定义分区器里如果处理不当，key 为空的消息可能全部落到一个分区，直接把某个 broker 打成热点。换轮询策略也需要小心，因为单条 send 进来的消息如果每条都轮询，会破坏 batch 的粘性，导致小请求变多，吞吐下降。
我自己的建议是给 key 为空的消息保留一个独立策略：可以按发送线程绑定固定分区，也可以用 ThreadLocal 做线程内轮询，既保证分散性，又尽量减少对 batch 的破坏。关键是不要图简单直接返回 0。
4.2 分区数变更引发的“地震”
自定义分区器最大的隐患就是分区扩容。如果使用简单的 hash(key) % numPartitions，当 numPartitions 从 6 变成 12 时，大部分消息的分区都会发生变化，消费者端会看到大量分区数据重新分配，甚至出现短时间内重复消费、offset 乱跳的误报。
缓解手段就是我在代码里用的槽位映射。把 0-1023 的虚拟槽位映射到实际分区，扩容时实际发生变化的只有那一小部分槽位，大部分消息的路由保持稳定。如果你的业务允许，还可以在变更分区数时选择低峰期，配合停止写入、迁移完成后再恢复的流程。
4.3 分区器里的耗时操作是吞吐杀手
partition 方法运行在 Producer 的 sender 线程调用路径上，一旦这里出现重量级操作，整个生产者都会变慢。我见过有人把 Redis 查询放进分区器，结果消息吞吐从每秒几万掉到几千；还有人为了路由做了复杂的正则匹配，每条消息上消耗几毫秒，在高吞吐场景下也是不可接受的。
正确的姿势是：所有需要外部数据才能做的路由判断，尽量在发送前通过拦截器或业务代码算好，把结果放在消息头里；分区器只做纯内存计算，运算量控制在纳秒到微秒级别。
4.4 幂等 Producer 与自定义分区器的兼容性
开了幂等或者事务的 Producer，对分区顺序有严格限制。自定义分区器如果出现同一 key 的消息先到了分区 A，后来又跑到了分区 B，整个幂等序列会被打破，最终报出InvalidPidMapping或者顺序错误。
所以一旦启用幂等和事务，分区器必须满足“同 key 消息分区稳定”的约束。我的建议很简单：先在本地单元测试里覆盖同一 key 连续发送多次的场景，再用集成测试验证最终消息全部落在同一分区。不要等到生产环境出现乱序才发现。
4.5 跨版本兼容：老接口到新接口的鸿沟
Kafka 客户端版本迭代中，Partitioner 接口本身也有变化。老版本中partition方法用的是List<PartitionInfo>，新版本改成了Cluster对象。如果你在升级客户端时不注意这块，编译虽然通过，运行期却可能因为方法签名不一致而出现问题。
更隐蔽的是，有些较老版本的自定义分区器代码用了已废弃的availablePartitionsForTopic，而在新版本里它的行为已经改变，只返回可用分区而非全部分区。如果你的路由依赖的是全量分区列表，这个改动会让消息跳过某些分区。所以升级客户端后，务必做一轮“分区分布回归测试”，确认分区路由行为没有静默改变。
5. 分区器的后续演进：从简单路由到感知流量的调度器
写到这里，可能有人会觉得手撸分区器就是“写一个类，算一个哈希”，没什么大不了。但在真实生产环境里，当分区器接入了监控和治理能力之后，它其实可以演化为一个流量调度器。我分享一下两个我觉得对业务帮助很大的演进方向。
第一个方向是“分区器 + 埋点监控”。分区器是唯一一个能在消息发送前看到全量业务 key、value 头部信息的地方，在这里做轻量级埋点很有价值。比如统计每个分区的消息量、估算每条消息大小、采集 key 的分布熵值。把埋点打到监控系统后，你可以实时看到分区的热度、出现过热分区的趋势，这比事后去消费端看 lag 要前置得多。
第二个方向是“动态分区权重”。有些业务的大客户会随着活动促销出现脉冲式流量，固定分区池在这种场景下反而不灵活。你可以在 configure 阶段读取配置中心的权重设置，动态调整槽位到分区的映射关系。虽然这样做复杂度和风险都不低，但确实能解决真实业务中的热点漂移问题。
这两个演进方向不一定每个人都需要，但只要你负责的 Kafka 集群规模到了一定程度，分区器的设计思路就不再是“写完能用”，而是“能不能长期稳定”。提前考虑到这些，后面你会省掉很多救火的夜晚。
最后再分享一个我自己一直在用的习惯：每写一个自定义分区器，我都会顺手写一个小工具类，把同一条业务 key 连续发送一百次，打印出每一次命中的分区列表，持续盯几轮。这个动作看似笨拙，但能快速发现哈希分布异常、分区数变化引起的不稳定，以及 key 为空时的兜底偏差。分区器这种代码看起来是很小的零件，但它的输出结果直接决定了下游所有的消费行为，越早暴露问题，浪费的时间就越少。
