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到StickyKafka的生产者分区策略经历过几次重要变化理解这段演进能帮你判断什么时候该自己动手。最早是轮询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 MapString, Integer businessLineMap; private SetString hotKeys; private Random random new Random(); Override public void configure(MapString, ? 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) { ListPartitionInfo 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而不是hashCodeKafka内置的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按Topic逐个分配每个Topic的分区按范围切给消费者实现简单同Topic分区连续消费者数不能整除分区数时前面的消费者多拿分区数远大于消费者数RoundRobin所有Topic的分区拉平后轮询分配整体均衡性好消费者订阅不同Topic时可能分配不均所有消费者订阅相同TopicSticky尽量保持原有分配均衡时最小化迁移减少Rebalance时的分区迁移实现复杂对Rebalance敏感的在线服务Range策略的问题在于假设2个消费者订阅同一个TopicTopic有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) { MapString, Subscription subscriptions groupSubscription.groupSubscription(); // 解析每个成员的机房和权重信息 MapString, String memberRack new HashMap(); MapString, Integer memberWeight new HashMap(); for (Map.EntryString, 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 SetString allTopics new HashSet(); for (Subscription sub : subscriptions.values()) { allTopics.addAll(sub.topics()); } // 按机房分组Topic-Partition MapString, ListTopicPartition 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())); } } // 分配优先把分区分给同机房的成员 MapString, ListTopicPartition assignment new HashMap(); for (String member : subscriptions.keySet()) { assignment.put(member, new ArrayList()); } for (Map.EntryString, ListTopicPartition entry : rackPartitions.entrySet()) { String rack entry.getKey(); ListTopicPartition partitions entry.getValue(); // 找出该机房的成员按权重排序 ListString 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; } } // 构建返回结果 MapString, Assignment result new HashMap(); for (Map.EntryString, ListTopicPartition 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:typeBrokerTopicMetrics,nameMessagesInPerSec,topicorder-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.ConsumerCoordinatorDEBUG如果发现同一个消费者每次拿到的分区都不一样那就是分配策略的确定性问题。修复方案是引入分配结果缓存在assign方法里先检查是否有上次的分配结果如果有且成员没变化直接返回缓存结果。只有当成员变化或分区数变化时才重新计算。另一个常见问题是分配策略的复杂度太高。我见过一个策略里做了复杂的图计算每次Rebalance要跑好几秒导致Rebalance超时。分配算法的复杂度应该控制在O(n log n)以内n是分区数。如果确实需要复杂计算考虑离线算好结果运行时只做查表。4.4 分区数与分区器的匹配原则分区器的效果和分区数强相关。分区数太少再好的分区器也施展不开分区数太多元数据开销和Rebalance成本都会上升。经验值是分区数 max(消费者线程数, 目标吞吐 / 单分区吞吐)。单分区吞吐在SSD上大约10-50MB/sHDD上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策略导致部分消费者过载。后来养成了一个习惯——分区器和分配策略的变更必须成对评估任何一方的改动都要检查另一方的兼容性。这个教训分享出来希望能帮你少走一段弯路。