★ FEATURED ARTICLE 1. 消息路由的底层逻辑与设计动机消息中间件在分布式系统里扮演的角色说白了就是一根“数据总线”。所有上游业务把事件丢进来下游业务按需取走。但这里有个容易被忽视的问题消息到底被丢到了哪个分区很多人用默认配置跑了很久直到某天发现某个分区磁盘爆了、某个消费者闲得发慌才回过头来查路由策略。Kafka 的分区机制本质上是一套“路由表”。生产者决定消息写进哪个分区消费者决定从哪些分区读数据。默认情况下Kafka 提供了几种开箱即用的策略比如生产者端的轮询、随机、按 key 哈希消费者端的 Range、RoundRobin、Sticky 等。这些策略覆盖了大部分通用场景但一旦业务有了特殊诉求默认策略就不够用了。什么算特殊诉求举几个我实际遇到过的例子。某业务要求同一批订单数据必须严格按时间顺序进入同一个分区但订单 ID 是随机生成的 UUID按 key 哈希完全无法保证顺序。另一个场景是下游消费者集群里有两组机器一组配置高、一组配置低希望高配机器多承担一些分区低配机器少分一些但默认的消费者分区分配策略是均匀分配的做不到按权重倾斜。还有一个典型场景是灰度发布新版本消费者只希望订阅特定标记的消息老版本继续处理其余消息这需要消费者端能按自定义规则过滤分区。这些需求靠配置文件是解决不了的必须自己写代码实现Partitioner接口或ConsumerPartitionAssignor接口。标题里提到的“自定义生产者分区器”和“自定义消费者分区器”正是解决这类问题的两把钥匙。前者控制消息写入的路由规则后者控制消费者组内部分区的分配规则。两者配合使用才能实现端到端的精细化流量控制。这篇文章适合已经用过 Kafka 基础功能、但对分区机制只停留在“知道有这回事”阶段的开发者。我会从接口设计、实现细节、参数计算、踩坑经验几个维度展开尽量把每个决策背后的“为什么”讲清楚。代码示例用 Java 编写因为 Kafka 原生客户端就是 Java 的其他语言的客户端原理相通照着思路翻译即可。2. 生产者分区器把消息送到该去的地方2.1 Partitioner 接口的核心方法拆解Kafka 生产者端的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) {} }partition方法的返回值就是目标分区的编号从 0 开始。参数里包含了 topic 名称、key 对象、key 的字节数组、value 对象、value 的字节数组以及集群的元数据Cluster。Cluster这个参数很关键它能告诉你当前 topic 有多少个分区、每个分区的 leader 在哪个节点上、有没有不可用的分区。很多自定义分区器只用了 key 和分区数忽略了Cluster提供的丰富信息导致实现出来的逻辑不够健壮。onNewBatch是较新版本加入的默认方法当生产者准备创建新批次时会被调用。如果你需要根据批次状态做动态调整可以重写它。close方法用于释放资源一般不需要特别处理。实现自定义分区器时有一个原则必须遵守同一个 key 必须始终映射到同一个分区除非你明确知道自己在做什么。这是 Kafka 保证分区内有序性的基础。如果同一个 key 的消息被分散到不同分区消费者端就无法按 key 维度的顺序处理了。2.2 按业务权重分配分区的完整实现假设有一个场景某类消息需要按照来源系统的权重写入不同分区权重高的系统占用更多分区权重低的系统占用较少分区。这种需求在默认的哈希策略下无法实现因为哈希是均匀散列的。先定义权重配置。我习惯把权重放在一个Map里key 是业务标识value 是权重值public class WeightedPartitioner implements Partitioner { private final MapString, Integer weightMap new HashMap(); private final AtomicInteger counter new AtomicInteger(0); Override public void configure(MapString, ? configs) { weightMap.put(systemA, 3); weightMap.put(systemB, 2); weightMap.put(systemC, 1); } Override public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { int totalPartitions cluster.partitionCountForTopic(topic); String bizKey key null ? default : key.toString(); int weight weightMap.getOrDefault(bizKey, 1); // 计算该业务可用的分区范围 int totalWeight weightMap.values().stream() .mapToInt(Integer::intValue).sum(); int partitionsForBiz Math.max(1, (int) Math.floor((double) weight / totalWeight * totalPartitions)); // 用取模方式在业务专属范围内轮询 int idx Math.abs(counter.getAndIncrement() % partitionsForBiz); return idx; } }这段代码有几个细节值得展开。第一totalPartitions是从Cluster动态获取的而不是写死的这样 topic 扩容分区后不需要改代码。第二partitionsForBiz用Math.max(1, ...)保证每个业务至少有一个分区可用避免权重过小导致计算出 0 个分区。第三用AtomicInteger做轮询计数器保证多线程环境下计数准确。但这个实现有个隐患不同业务计算出的分区范围可能重叠。比如 systemA 算出 3 个分区systemB 算出 2 个分区如果都从 0 开始取模它们会争抢 0 和 1 号分区。正确的做法是给每个业务分配一个固定的起始偏移量private final MapString, Integer offsetMap new HashMap(); // 在 configure 中初始化 offsetMap.put(systemA, 0); offsetMap.put(systemB, 3); offsetMap.put(systemC, 5);然后在partition方法里用offsetMap.get(bizKey) idx作为最终分区号。这样各业务的分区范围互不重叠权重分配才真正生效。2.3 分区器参数配置与生产端集成写完分区器类之后需要在生产者配置中指定它Properties props new Properties(); props.put(bootstrap.servers, broker1:9092,broker2:9092); props.put(key.serializer, org.apache.kafka.common.serialization.StringSerializer); props.put(value.serializer, org.apache.kafka.common.serialization.StringSerializer); props.put(partitioner.class, com.example.WeightedPartitioner);注意partitioner.class配置项的全限定类名必须准确否则生产者启动时会直接抛ClassNotFoundException。另外如果自定义分区器需要读取额外的配置参数可以通过configure方法的configs参数获取这些参数来自生产者的Properties中所有以特定前缀开头的配置项。有一个容易踩的坑分区数变更后的兼容性。如果 topic 从 6 个分区扩容到 12 个分区原来按 6 取模的逻辑会全部失效同一个 key 可能被路由到完全不同的分区。所以自定义分区器里涉及分区数计算的逻辑一定要考虑扩容场景。我的做法是在分区器里维护一个版本号扩容后通过配置中心推送新版本号触发分区器重新计算映射关系同时下游消费者做好双读兼容。提示自定义分区器上线前务必在测试环境验证分区分布是否符合预期。可以用kafka-producer-perf-test工具配合自定义分区器压测观察各分区的消息量是否与权重匹配。3. 消费者分区分配器让每个消费者各司其职3.1 ConsumerPartitionAssignor 接口解析消费者端的自定义分区分配器要实现ConsumerPartitionAssignor接口。这个接口比生产者的Partitioner复杂不少因为它涉及消费者组内多个成员之间的协调。核心方法是assignpublic interface ConsumerPartitionAssignor { GroupAssignment assign(Cluster metadata, GroupSubscription groupSubscription); void onAssignment(Assignment assignment, ConsumerGroupMetadata metadata); String name(); default ListRebalanceProtocol supportedProtocols() { ... } }assign方法接收集群元数据和组订阅信息返回一个GroupAssignment里面包含了每个消费者成员应该分配哪些分区。groupSubscription参数里包含了所有成员的订阅关系你可以从中获取每个成员订阅了哪些 topic、有哪些自定义的用户数据。onAssignment方法在分配结果下发到消费者后调用可以用来做一些初始化工作比如根据分配到的分区调整本地缓存。name()方法返回分配器的名称这个名称会出现在消费者组的协调日志里方便排查问题。3.2 按消费者权重分配分区的实现假设消费者组里有三个成员配置分别是高配、中配、低配希望分区分配比例大约是 3:2:1。默认的 Range 或 RoundRobin 策略做不到这一点需要自己实现。思路是这样的先收集所有成员及其权重然后计算每个成员应得的分区数量最后按顺序把分区分配下去。public class WeightedAssignor implements ConsumerPartitionAssignor { Override public GroupAssignment assign(Cluster metadata, GroupSubscription groupSubscription) { MapString, Subscription subscriptions groupSubscription.groupSubscription(); // 收集所有 topic 的分区 SetString allTopics new HashSet(); for (Subscription sub : subscriptions.values()) { allTopics.addAll(sub.topics()); } ListTopicPartition allPartitions new ArrayList(); for (String topic : allTopics) { for (PartitionInfo p : metadata.partitionsForTopic(topic)) { allPartitions.add(new TopicPartition(topic, p.partition())); } } // 解析每个成员的权重 MapString, Integer memberWeights new HashMap(); int totalWeight 0; for (Map.EntryString, Subscription entry : subscriptions.entrySet()) { int weight parseWeight(entry.getValue().userData()); memberWeights.put(entry.getKey(), weight); totalWeight weight; } // 按权重计算每个成员应得的分区数 MapString, ListTopicPartition assignment new HashMap(); int assigned 0; ListString members new ArrayList(subscriptions.keySet()); for (int i 0; i members.size(); i) { String member members.get(i); int weight memberWeights.get(member); int count; if (i members.size() - 1) { count allPartitions.size() - assigned; } else { count (int) Math.round( (double) weight / totalWeight * allPartitions.size()); } assignment.put(member, new ArrayList(allPartitions.subList(assigned, Math.min(assigned count, allPartitions.size())))); assigned count; } return new GroupAssignment(assignment); } }这段代码里parseWeight方法从订阅的userData中解析权重。消费者在加入组时可以通过ConsumerConfig的group.instance.id或者自定义的userData传递权重信息。最后一个成员直接拿走剩余所有分区避免因为四舍五入导致分区分配不完整。3.3 分配策略的触发时机与再平衡处理自定义分配器写完之后需要在消费者配置中指定props.put(partition.assignment.strategy, com.example.WeightedAssignor);这里有个关键点分配策略的触发时机。消费者组发生以下事件时会触发再平衡新成员加入、成员主动离开、成员心跳超时、订阅的 topic 分区数变化。每次再平衡都会重新调用assign方法所以你的分配逻辑必须是幂等的不能依赖上一次的分配结果。再平衡期间所有消费者都会暂停消费直到新的分配方案下发完成。如果分配逻辑太复杂、耗时太长会导致再平衡时间过长影响整体吞吐。我的经验是分配计算尽量控制在毫秒级不要在里面做远程调用或复杂查询。还有一个坑粘性分配。Kafka 从 2.4 版本开始支持 Sticky 分配策略目的是在再平衡时尽量保留原有的分配关系减少分区迁移带来的开销。如果你自定义的分配器每次都重新计算可能会破坏粘性导致大量分区在成员之间来回迁移。解决办法是在assign方法里参考上一次的分配结果尽量让分区留在原成员上。可以通过Subscription的userData传递上一次的分配信息或者利用 Kafka 提供的AbstractStickyAssignor作为基类来扩展。注意自定义消费者分配器需要所有消费者成员都使用相同的分配器类。如果组内成员配置不一致协调者会抛出InconsistentGroupProtocolException导致消费者无法加入组。4. 端到端联调与性能验证4.1 生产者与消费者分区器的协同工作生产者分区器和消费者分配器虽然是两个独立的组件但在实际业务中往往需要配合使用。举个例子生产者按业务权重把消息写入不同分区消费者端也希望按业务权重分配分区让高配机器处理高权重业务的分区。这时候两边的逻辑需要对齐。对齐的关键在于分区编号的语义。生产者端决定了 0-2 号分区属于 systemA3-4 号分区属于 systemB5 号分区属于 systemC。消费者端在分配时需要知道这个映射关系才能把对应分区分配给合适的消费者。我的做法是在消费者分配器的userData里携带业务标识分配时根据业务标识匹配分区范围。联调时可以用一个简单的验证方法生产者发送一批带业务标识的消息消费者端记录每个成员实际处理的消息数量最后统计各成员的处理量比例是否与权重一致。如果不一致检查两边的分区范围定义是否对齐。4.2 分区分布验证与监控指标上线之后监控是必不可少的。Kafka 自带了一些关键指标配合自定义分区器使用时需要特别关注指标名称含义关注点records-per-second各分区每秒写入消息数是否与权重匹配bytes-in-per-second各分区每秒写入字节数是否存在数据倾斜consumer-lag消费者滞后消息数高权重消费者是否过载rebalance-rate再平衡频率是否频繁触发再平衡assigned-partitions每个消费者分配到的分区数是否与权重一致如果发现某个分区的bytes-in-per-second远高于其他分区说明生产者的权重分配没有生效需要检查分区器的partition方法返回值是否落在预期范围内。如果consumer-lag持续增长说明消费者处理能力不足可能需要调整权重比例或增加消费者实例。4.3 压测方案与容量评估自定义分区器上线前压测是必须的。我通常用以下步骤做容量评估第一步确定单分区的写入上限。用kafka-producer-perf-test向单个分区持续写入观察吞吐量何时达到瓶颈。这个值取决于磁盘类型、网络带宽、消息大小等因素通常在几 MB/s 到几十 MB/s 之间。第二步确定单消费者的处理上限。写一个简单的消费程序记录每秒能处理多少条消息。这个值取决于业务逻辑的复杂度简单的反序列化和落库可能达到几万 TPS复杂的计算逻辑可能只有几百 TPS。第三步根据权重比例计算所需的分区数和消费者数。假设总写入量是 100 MB/s单分区上限是 20 MB/s那么至少需要 5 个分区。如果权重是 3:2:1可以分配 3 个分区给高权重业务、2 个给中权重、1 个给低权重总共 6 个分区。第四步验证再平衡场景。模拟消费者上下线观察再平衡期间的消息积压情况和恢复时间。如果再平衡时间超过 30 秒需要优化分配逻辑或调整session.timeout.ms和heartbeat.interval.ms参数。5. 常见问题与排查技巧实录5.1 分区器不生效的排查路径自定义分区器写完、配置也加了但消息还是均匀分布在各分区这是最常见的问题。排查路径如下先确认partitioner.class配置是否真的被加载了。可以在分区器的构造函数里打一行日志启动生产者时观察日志是否输出。如果没有输出说明配置没生效检查配置项的 key 是否拼写正确、类名是否全限定、类是否在 classpath 中。如果日志输出了但分区结果不对检查partition方法的返回值。可以在方法里加日志打印topic、key、计算出的分区号。如果返回值始终是同一个值可能是取模运算的分母为 0 或者计数器没有正确递增。还有一种情况是消息的 key 为 null。Kafka 默认对 null key 使用轮询策略但如果你自定义了分区器null key 也会走你的逻辑。如果代码里没有处理 null key 的分支可能会抛NullPointerException导致生产者发送失败。5.2 消费者再平衡失败的典型原因消费者分配器导致的再平衡失败通常有以下几个原因一是分配结果不完整。GroupAssignment里必须包含所有订阅了 topic 的成员且每个成员分配到的分区不能重复。如果某个分区被分配给了两个成员协调者会拒绝这次分配触发新一轮再平衡。二是成员权重解析失败。如果userData的格式不一致比如有的成员传了 JSON、有的传了纯文本parseWeight方法可能抛异常导致整个分配过程失败。解决办法是在解析时做容错处理解析失败时使用默认权重。三是分配器类不一致。前面提到过组内所有成员必须使用相同的分配器类。如果某个成员用了默认的 Range 策略另一个用了自定义策略协调者会抛出InconsistentGroupProtocolException。排查时检查所有消费者的partition.assignment.strategy配置是否一致。5.3 分区倾斜的应急处理方案即使做了权重分配线上仍可能出现分区倾斜。应急处理方案分短期和长期两种。短期方案是手动触发再平衡。通过调整消费者的group.instance.id或者重启部分消费者实例强制触发再平衡让分配器重新计算分配方案。如果倾斜是因为某个消费者处理能力下降导致的重启后可能恢复正常。长期方案是引入动态权重调整。在消费者端定期上报处理延迟和吞吐量分配器根据这些指标动态调整权重。比如某个消费者延迟持续偏高下次再平衡时自动降低其权重减少分配到的分区数。这个方案实现复杂度较高但能有效应对消费者性能波动。问题现象可能原因排查方法解决方案消息均匀分布分区器未加载检查构造函数日志确认配置项和类路径同一 key 分散分区数变更对比扩容前后分区数引入版本号兼容再平衡超时分配逻辑耗时打印 assign 耗时优化计算逻辑分区重复分配分配结果有误检查 GroupAssignment确保分区不重叠消费者无法加入组分配器不一致检查所有成员配置统一分配器类提示自定义分区器和分配器的代码建议单独抽成模块方便在多个项目间复用。同时做好单元测试模拟不同分区数、不同成员数量的场景确保边界条件处理正确。6. 个人实操经验与后续扩展方向我在实际项目里用自定义分区器解决过几个棘手问题踩过的坑比想象中多。最深刻的一次是分区扩容后没有做兼容处理导致同一个订单的消息被路由到了不同分区下游按订单维度做聚合时出现了数据错乱。后来在分区器里加了分区数版本号扩容时通过配置中心推送新版本消费者端做双读兼容才彻底解决。另一个经验是自定义分配器的逻辑尽量简单。我见过有团队在assign方法里查数据库获取权重结果再平衡时数据库连接池被打满整个消费者组卡死。分配器应该只做纯计算所有外部数据提前通过userData传进来。后续如果要做扩展可以考虑几个方向。一是把权重配置做成动态可调的通过配置中心实时推送不用重启消费者就能调整分配比例。二是引入机器学习预测根据历史消费延迟预测未来的负载提前调整权重。三是把分区器和分配器的逻辑做成插件化不同业务线可以注册自己的实现通过 SPI 机制加载。这些扩展方向听起来复杂但核心思路没变生产端控制写入路由消费端控制读取分配两端对齐分区语义就能实现精细化的流量治理。把基础的自定义分区器和分配器写扎实后面的扩展都是水到渠成的事。
阅读完成 · 觉得有帮助？
