1. 从一个真实的数据倾斜事故说起

去年帮一个做实时风控的团队排查问题，他们的Kafka集群有12个分区，但监控面板上有个诡异现象：8个分区几乎空转，另外4个分区的写入速率是其他分区的十几倍，消费端Lag持续飙升。排查了一圈，Broker没问题，网络没问题，最后定位到生产者的分区策略上——他们用的是默认的粘性分区器，而消息Key的分布恰好集中在少数几个值上，导致大量消息被路由到同一批分区。

这个案例很典型。很多人用Kafka的时候，对分区器的关注度远远不够，觉得默认的就行了。但只要你涉及有序性保证、数据倾斜治理、多租户隔离、灰度路由这些场景，默认分区器基本都不够用。Kafka给了我们两个口子：生产者端自定义分区器和消费者端自定义分区分配策略。前者决定消息写到哪个分区，后者决定消费者读哪些分区。两者配合好了，才能把整个链路的吞吐和均衡都控制住。

这篇文章我会把这两块拆开讲透。从接口设计、实现细节、参数配置到踩坑经验，都会覆盖到。适合已经用过Kafka、但想深入控制分区行为的开发者，也适合正在被数据倾斜困扰的运维同学。

2. 生产者分区器：消息到底该落在哪个分区

2.1 Partitioner接口的设计意图

Kafka生产者的分区逻辑封装在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(); }

partition方法是核心，返回值就是目标分区号。参数里给了topic、key、序列化后的keyBytes、value、valueBytes以及集群元数据Cluster。Cluster里可以拿到分区数量、可用分区列表等信息，这是你做自定义路由时最重要的数据来源。

这里有个设计细节值得注意：key和keyBytes都传进来了。为什么要冗余？因为keyBytes是序列化后的字节数组，如果你需要做哈希计算，直接用keyBytes比先序列化key再算要高效得多。Kafka默认的DefaultPartitioner就是拿keyBytes做murmur2哈希的。

Configurable接口让你能通过configure方法拿到生产者的配置项，这意味着你可以在分区器里读取自定义参数，比如"哪些Key应该走哪些分区"的映射表路径。Closeable则用于释放资源，比如你加载了外部配置文件，关闭时需要清理。

2.2 默认分区器的演进：从Round-Robin到Sticky

Kafka的生产者分区策略经历过几次重要变化，理解这段演进能帮你判断什么时候该自己动手。

最早是轮询（Round-Robin）：不管有没有Key，消息依次分配到各个分区。这种方式均衡性最好，但有个致命问题——相同Key的消息会被打散到不同分区，破坏了Kafka最核心的有序性保证。

后来改成Key哈希 + 无Key轮询：有Key的消息按murmur2哈希取模选分区，无Key的消息轮询。这解决了有序性问题，但无Key场景下轮询会导致频繁的批次切换，影响吞吐。

2.4版本之后引入了粘性分区器（Sticky Partitioner）：无Key的消息会先"粘"在一个分区上，攒够一个批次再换下一个分区。这样既保证了均衡，又减少了批次切换开销。有Key的消息依然走哈希。

但粘性分区器有个前提：它假设你的Key分布是均匀的。如果你的Key天然集中（比如用用户ID做Key，但大客户占了80%流量），哈希结果就会倾斜。这时候默认分区器就无能为力了，必须自定义。

2.3 自定义分区器的完整实现

下面是一个生产级的自定义分区器示例，实现了"按业务线路由 + 热点Key打散"的逻辑：

public class BusinessLinePartitioner implements Partitioner { private Map<String, Integer> businessLineMap; private Set<String> hotKeys; private Random random = new Random(); @Override public void configure(Map<String, ?> configs) { // 从配置中读取业务线到分区的映射 String mapping = (String) configs.get("business.line.mapping"); businessLineMap = parseMapping(mapping); // 读取热点Key列表 String hotKeyStr = (String) configs.get("hot.keys"); hotKeys = new HashSet<>(Arrays.asList(hotKeyStr.split(","))); } @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 (key == null) { // 无Key消息，用线程本地随机数打散 return random.nextInt(numPartitions); } String keyStr = key.toString(); // 热点Key：打散到所有分区，牺牲有序性换均衡 if (hotKeys.contains(keyStr)) { return random.nextInt(numPartitions); } // 普通Key：按业务线前缀路由到指定分区段 String businessLine = extractBusinessLine(keyStr); Integer basePartition = businessLineMap.get(businessLine); if (basePartition != null) { // 在该业务线的分区段内做哈希 int segmentSize = numPartitions / businessLineMap.size(); int offset = Math.abs(murmur2(keyBytes)) % segmentSize; return basePartition * segmentSize + offset; } // 兜底：默认哈希 return Math.abs(murmur2(keyBytes)) % numPartitions; } private int murmur2(byte[] data) { // Kafka内置的murmur2实现 final int seed = 0x9747b28c; final int m = 0x5bd1e995; final int r = 24; int h = seed ^ data.length; int length = data.length; int i = 0; while (length >= 4) { int k = (data[i] & 0xff) | ((data[i + 1] & 0xff) << 8) | ((data[i + 2] & 0xff) << 16) | ((data[i + 3] & 0xff) << 24); k *= m; k ^= k >>> r; k *= m; h *= m; h ^= k; i += 4; length -= 4; } // 处理剩余字节... h ^= h >>> 13; h *= m; h ^= h >>> 15; return h; } @Override public void close() { // 清理资源 } }

这个实现里有几个关键决策点值得展开说。

为什么热点Key要打散？因为热点Key如果固定落在一个分区，那个分区就会成为瓶颈。打散之后，虽然相同Key的消息可能落到不同分区，破坏了严格有序性，但换来了整体吞吐的提升。这是一个典型的权衡——如果你的业务对热点Key的有序性要求极高，那就不能打散，得考虑其他方案（比如给热点Key单独建Topic）。

为什么用murmur2而不是hashCode？Kafka内置的murmur2分布更均匀，而且和默认分区器保持一致，方便迁移。Java的hashCode()在某些数据集上分布很差，容易产生碰撞。

分区段划分的边界处理：numPartitions / businessLineMap.size()这个除法在分区数不能被业务线数整除时会丢余数。生产环境建议把分区数设置成业务线数的整数倍，或者用更精细的分配算法。

2.4 配置与部署的实操细节

自定义分区器写完之后，需要在生产者配置里指定：

props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.example.BusinessLinePartitioner"); props.put("business.line.mapping", "order:0,payment:1,risk:2"); props.put("hot.keys", "user_10086,user_10010");

这里有个容易踩的坑：自定义配置项必须通过configure方法读取，不能直接注入。因为分区器实例是Kafka内部通过反射创建的，不走Spring容器。如果你在分区器里用了@Autowired，启动时会直接NPE。

另一个坑是分区器实例的线程安全。Kafka生产者是线程安全的，多个线程共享同一个分区器实例。所以你的partition方法必须是线程安全的。上面代码里用了Random，它在多线程下虽然线程安全但性能有争用，高并发场景建议换成ThreadLocalRandom。

还有一点：分区器变更后，历史消息不会重新分布。如果你从默认分区器切换到自定义分区器，新消息会按新规则走，但老消息还在原来的分区里。消费端如果依赖分区来做有序处理，需要考虑到这个过渡期。

3. 消费者分区分配策略：谁读哪些分区

3.1 ConsumerPartitionAssignor接口解析

生产者端决定消息写哪里，消费者端决定谁来读。Kafka的消费者分区分配策略通过org.apache.kafka.clients.consumer.ConsumerPartitionAssignor接口实现：

public interface ConsumerPartitionAssignor { String name(); GroupAssignment assign(Cluster metadata, GroupSubscription groupSubscription); void onAssignment(Assignment assignment, ConsumerGroupMetadata metadata); default void onLeaderElected(...) {} }

assign方法是核心，由消费组的Leader消费者调用。它接收集群元数据和所有成员的订阅信息，返回每个成员应该分配到的分区列表。注意：只有Leader执行分配逻辑，其他成员只是接收结果。所以你的分配算法只需要在Leader上跑，不用担心分布式一致性问题。

GroupSubscription里包含了每个成员的订阅Topic列表和用户自定义数据（通过Subscription.userData()传递）。这个userData是个很有用的扩展点——你可以让每个成员上报自己的元信息（比如机器规格、所在机房），Leader根据这些信息做亲和性分配。

3.2 三种内置策略的适用边界

Kafka内置了三种分配策略，理解它们的差异是自定义的基础：

Range策略的问题在于：假设2个消费者订阅同一个Topic，Topic有3个分区，那么消费者A拿分区0和1，消费者B拿分区2。A的负载是B的两倍。如果消费者订阅了多个Topic，这种不均衡会累积。

RoundRobin解决了单Topic的不均衡，但如果消费者订阅的Topic不同，可能出现某个消费者被分到大量分区而另一个很少的情况。

Sticky是2.4版本后的默认策略，它在均衡性和稳定性之间做了折中。但Sticky的"粘性"是有限度的——当成员变化时，它还是会触发重新分配，只是尽量少迁移。

3.3 自定义分配策略的实现：机房亲和 + 负载感知

下面实现一个"同机房优先 + 按成员权重分配"的策略，适用于多机房部署的场景：

public class RackAwareAssignor implements ConsumerPartitionAssignor { @Override public String name() { return "rack-aware"; } @Override public GroupAssignment assign(Cluster metadata, GroupSubscription groupSubscription) { Map<String, Subscription> subscriptions = groupSubscription.groupSubscription(); // 解析每个成员的机房和权重信息 Map<String, String> memberRack = new HashMap<>(); Map<String, Integer> memberWeight = new HashMap<>(); for (Map.Entry<String, Subscription> entry : subscriptions.entrySet()) { ByteBuffer userData = entry.getValue().userData(); if (userData != null) { String rack = readString(userData); int weight = userData.getInt(); memberRack.put(entry.getKey(), rack); memberWeight.put(entry.getKey(), weight); } } // 收集所有需要分配的Topic-Partition Set<String> allTopics = new HashSet<>(); for (Subscription sub : subscriptions.values()) { allTopics.addAll(sub.topics()); } // 按机房分组Topic-Partition Map<String, List<TopicPartition>> rackPartitions = new HashMap<>(); for (String topic : allTopics) { for (PartitionInfo p : metadata.partitionsForTopic(topic)) { String rack = p.rack() != null ? p.rack() : "default"; rackPartitions.computeIfAbsent(rack, k -> new ArrayList<>()) .add(new TopicPartition(topic, p.partition())); } } // 分配：优先把分区分给同机房的成员 Map<String, List<TopicPartition>> assignment = new HashMap<>(); for (String member : subscriptions.keySet()) { assignment.put(member, new ArrayList<>()); } for (Map.Entry<String, List<TopicPartition>> entry : rackPartitions.entrySet()) { String rack = entry.getKey(); List<TopicPartition> partitions = entry.getValue(); // 找出该机房的成员，按权重排序 List<String> rackMembers = memberRack.entrySet().stream() .filter(e -> e.getValue().equals(rack)) .map(Map.Entry::getKey) .sorted((a, b) -> memberWeight.get(b) - memberWeight.get(a)) .collect(Collectors.toList()); if (rackMembers.isEmpty()) { // 没有同机房成员，分配给所有成员轮询 rackMembers = new ArrayList<>(subscriptions.keySet()); } // 按权重比例分配 int totalWeight = rackMembers.stream() .mapToInt(memberWeight::get).sum(); int idx = 0; for (TopicPartition tp : partitions) { String member = rackMembers.get(idx % rackMembers.size()); assignment.get(member).add(tp); idx++; } } // 构建返回结果 Map<String, Assignment> result = new HashMap<>(); for (Map.Entry<String, List<TopicPartition>> e : assignment.entrySet()) { result.put(e.getKey(), new Assignment(e.getValue())); } return new GroupAssignment(result); } @Override public void onAssignment(Assignment assignment, ConsumerGroupMetadata metadata) { // 可以在这里做分配后的初始化，比如预热缓存 } }

这个实现的核心思路是：先按机房分组，再在机房内按权重分配。这样跨机房流量最小化，同时高配机器能承担更多分区。

有几个实现细节需要特别注意。

userData的序列化格式：Subscription.userData()返回的是ByteBuffer，你需要自己定义序列化协议。上面代码里假设了"字符串+整数"的格式，实际使用时需要和消费者端的Subscription构造逻辑对齐。建议用简单的自定义协议，不要引入复杂的序列化框架，因为这段代码在Rebalance时会频繁执行。

分区分配的确定性：assign方法可能被多次调用（比如Leader切换），每次调用应该产生相同的结果。所以代码里用了排序来保证顺序稳定。如果你用了HashMap的遍历顺序，不同JVM实例可能产生不同结果，导致分配震荡。

权重信息的来源：成员权重可以从配置读取，也可以根据机器规格动态计算。我见过一个团队用CPU核数作为权重，效果不错。但要注意，权重差异不宜过大，否则低配机器可能分不到分区，造成资源浪费。

3.4 消费者端的配置与协作

自定义分配策略需要在消费者端配置：

props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "com.example.RackAwareAssignor");

消费者在加入组时，会通过Subscription.userData()上报自己的元信息：

ConsumerPartitionAssignor.Subscription subscription = new ConsumerPartitionAssignor.Subscription( Collections.singletonList("order-topic"), buildUserData("rack-a", 8) // 机房A，权重8 );

这里有个关键点：所有消费者必须使用相同的分配策略。如果组内成员配置不一致，会导致协议不兼容，消费者无法加入组。Kafka在JoinGroup阶段会校验协议名称，不匹配的成员会被拒绝。

另一个坑是Rebalance的触发频率。自定义分配策略如果实现不当，可能导致频繁Rebalance。比如你的assign方法每次返回的结果差异很大，消费者就会不断重新分配。建议在assign里加入稳定性优化：优先保持上次的分配结果，只在必要时调整。

4. 生产环境中的分区器调优与故障排查

4.1 数据倾斜的量化诊断方法

数据倾斜不是靠感觉判断的，需要量化。最直接的方法是看每个分区的消息速率和字节速率：

kafka-run-class.sh kafka.tools.JmxTool \ --object-name kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec,topic=order-topic \ --jmx-url service:jmx:rmi:///jndi/rmi://broker1:9999/jmxrmi

但更实用的是用Kafka自带的kafka-log-dirs工具看各分区的日志大小：

kafka-log-dirs.sh --bootstrap-server broker1:9092 \ --topic-list order-topic --describe

输出里会列出每个分区的size和offsetLag。如果最大分区和最小分区的size差距超过3倍，基本可以判定有倾斜。

诊断出倾斜后，要区分是Key分布问题还是分区器问题。方法很简单：统计一段时间内各Key的消息量，看Top 10 Key占比。如果Top 10占了50%以上，那是Key分布问题，自定义分区器只能缓解不能根治，需要考虑Key设计或拆分Topic。如果Key分布均匀但分区不均，那就是分区器的问题。

4.2 分区器热更新的安全操作

生产环境修改分区器不能直接重启，需要平滑过渡。推荐的做法是双写+灰度切换：

第一步，新老分区器同时存在，通过配置开关控制走哪个。新分区器先只对1%的流量生效，观察目标分区的写入速率和消费Lag。

第二步，逐步扩大灰度比例，每次扩大后观察至少30分钟。重点关注消费端是否有分区Lag异常升高——这通常意味着新分区器把消息打到了消费能力不足的分区。

第三步，全量切换后，保留老分区器代码至少一个版本周期，以便快速回滚。

这里有个细节：分区器切换会导致相同Key的消息落到不同分区。如果你的消费端依赖分区有序性，切换期间会出现乱序。解决方案是在切换点打一个特殊标记消息，消费端遇到标记后清空本地状态。

4.3 消费者分配策略导致的Rebalance风暴

自定义分配策略最容易引发的问题是Rebalance风暴。典型症状是消费者频繁进出组，日志里大量Rebalance记录，消费Lag忽高忽低。

根因通常是分配结果不稳定。比如你的策略里用了Math.random()或者依赖了非确定性的数据源，每次assign结果都不同，消费者就会不断重新分配。

排查方法是打开消费者的DEBUG日志，观察每次Rebalance后的分配结果：

log4j.logger.org.apache.kafka.clients.consumer.internals.ConsumerCoordinator=DEBUG

如果发现同一个消费者每次拿到的分区都不一样，那就是分配策略的确定性问题。

修复方案是引入分配结果缓存：在assign方法里，先检查是否有上次的分配结果，如果有且成员没变化，直接返回缓存结果。只有当成员变化或分区数变化时才重新计算。

另一个常见问题是分配策略的复杂度太高。我见过一个策略里做了复杂的图计算，每次Rebalance要跑好几秒，导致Rebalance超时。分配算法的复杂度应该控制在O(n log n)以内，n是分区数。如果确实需要复杂计算，考虑离线算好结果，运行时只做查表。

4.4 分区数与分区器的匹配原则

分区器的效果和分区数强相关。分区数太少，再好的分区器也施展不开；分区数太多，元数据开销和Rebalance成本都会上升。

经验值是：分区数 = max(消费者线程数, 目标吞吐 / 单分区吞吐)。单分区吞吐在SSD上大约10-50MB/s，HDD上5-20MB/s。假设你的目标吞吐是500MB/s，用SSD，那至少需要10-50个分区。

但分区数不是越多越好。分区数超过1000后，Kafka的元数据同步和Controller的负担会明显增加。而且分区数多了之后，自定义分区器的哈希空间变大，如果Key的哈希值分布不够均匀，反而容易出现新的倾斜。

一个实用的做法是：先用默认分区器跑一段时间，收集Key的分布数据，再根据数据设计分区器。不要一上来就自定义，很多时候调整Key的设计比写分区器更有效。

5. 几个容易被忽略的边界场景

5.1 分区器与事务的交互

如果你用了Kafka事务，分区器的行为会有些微妙变化。事务要求同一个事务内的消息要么全成功要么全失败，而分区器决定了消息落在哪些分区。如果事务内消息被分到了多个分区，事务提交时需要协调多个分区的提交，开销更大。

更麻烦的是，如果分区器在事务执行过程中返回了不同的分区（比如依赖了外部状态），可能导致事务的部分消息写入失败。所以事务场景下，分区器必须是纯函数——相同的输入必须返回相同的输出，不能依赖随机数或外部可变状态。

5.2 消费者分配策略与静态成员的配合

Kafka 2.3引入了静态成员（Static Membership），通过group.instance.id配置。静态成员在重启时不会触发Rebalance，而是保持原有的分区分配。

但自定义分配策略需要感知静态成员。如果你的策略在成员重启时重新分配了分区，静态成员的优势就没了。实现时需要在assign方法里检查成员的group.instance.id，如果成员只是暂时离线，保留其原有分配。

5.3 多Topic订阅下的分配均衡

当一个消费者组订阅多个Topic时，分配策略需要跨Topic做均衡。Range策略是按Topic独立分配的，可能导致消费者A在Topic1上分到3个分区、Topic2上分到1个，而消费者B反过来。整体看是均衡的，但单个消费者内部的负载可能不均。

自定义策略时，建议把所有Topic的分区拉平成一个列表，统一做分配。这样能保证每个消费者的总分区数大致相等。但要注意，如果不同Topic的流量差异很大，按分区数均衡不等于按流量均衡。更精细的做法是给每个Topic配置权重，按加权分区数分配。

5.4 分区器变更后的消费端兼容

最后说一个实际运维中容易忽略的点：分区器变更后，消费端的offset管理。Kafka的offset是按分区存储的，分区器变更后，相同Key的消息可能落到新分区，而新分区的offset从当前最新位置开始。如果消费端依赖offset做断点续传，需要确保新分区的消费起点正确。

建议在分区器变更时，消费端先暂停消费，等所有在途消息处理完，记录当前各分区的offset，然后切换分区器，再从记录的offset开始消费。这个过程需要停机窗口，但能保证数据不丢不重。

如果业务不能停机，那就需要在消息里带上业务时间戳或序列号，消费端根据这些信息做去重和排序。这增加了消费端的复杂度，但换来了平滑切换的能力。

我在实际项目里踩过最深的坑是：自定义分区器上线后忘了同步更新消费端的分配策略，结果生产端把消息打散到了所有分区，消费端还在用Range策略，导致部分消费者过载。后来养成了一个习惯——分区器和分配策略的变更必须成对评估，任何一方的改动都要检查另一方的兼容性。这个教训分享出来，希望能帮你少走一段弯路。