如何实现Kafka Producer基于消费者而非分区的消息公平分配？

首先明确说：Kafka 本身并没有提供直接让生产者“基于消费者分配消息”的能力——这是因为 Kafka 的核心设计是围绕「分区」来做消息路由和消费负载均衡的，生产者只负责把消息发送到分区，而消费组的协调器会负责把分区分配给消费者，两者是解耦的。

你遇到的问题本质是消息没有均匀分布到所有20个分区，导致部分消费者分配的分区没有消息，负载不均。我们可以从调整消息的分区分布入手，来实现消费负载的公平分配：

1. 先排查默认分区器的行为问题 #

你提到使用 DefaultPartitioner（kafka-client 2.3.1）发送10条消息后，只有partition-0到9有消息，这其实不符合默认分区器的预期——当生产者不指定消息key时，DefaultPartitioner 会采用轮询所有可用分区的策略来发送消息。出现这种情况大概率是因为生产者初始化时没有获取到全部20个分区的元数据。

你可以尝试调整生产者的以下配置：

• metadata.max.age.ms：默认是5分钟，调小这个值（比如设为1000），让生产者更频繁地更新集群元数据，确保能感知到所有20个分区都是可用的。
• 先通过kafka-topics.sh --describe --topic <你的topic名>检查所有分区的ISR状态，确保没有不可用的分区（如果acks设置为all，不可用分区会被默认分区器排除）。

当生产者能正确感知到所有20个分区后，无key的消息会轮询发送到所有20个分区。10条消息的话，会均匀分布在10个不同的分区（每个分区1条），而你的消费组每个消费者分配4个分区，这样每个消费者会处理1-2条消息，基本达到负载均衡的效果。

2. 自定义分区器实现更精准的均匀分布 #

如果默认分区器的轮询策略还不能满足你的需求（比如需要强制消息均匀分布到所有分区），可以自定义一个分区器：

public class UniformPartitioner implements Partitioner {
    private AtomicInteger counter = new AtomicInteger(0);
    private int totalPartitions;

    @Override
    public void configure(Map<String, ?> configs) {
        // 可以从配置中传入总分区数，或者后续从元数据获取
    }

    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        totalPartitions = partitions.size();
        // 无key时，用原子计数器实现严格轮询所有分区
        if (keyBytes == null) {
            return counter.getAndIncrement() % totalPartitions;
        }
        // 有key时，沿用默认的哈希策略保证同key消息进入同一分区
        return Utils.toPositive(Utils.murmur2(keyBytes)) % totalPartitions;
    }

    @Override
    public void close() {}
}

然后在生产者配置中指定这个分区器：

partitioner.class=com.your.package.UniformPartitioner

这样所有无key的消息会均匀分布到20个分区，消费组的每个消费者分配的4个分区会分摊到10条消息，最终每个消费者处理2条左右的消息，实现你想要的负载公平。

3. 为什么不能直接基于消费者分配消息？ #

最后补充一下：Kafka 不支持这种模式的核心原因是消费者的状态是动态变化的——消费者可能随时上下线、消费组可能发生重平衡，这些信息只有消费组协调器知道，生产者无法实时获取到最新的消费者-分区分配关系。如果生产者硬要绑定消费者，会导致系统耦合度极高，并且无法应对消费者动态变化的场景，违背了Kafka的解耦设计原则。

内容的提问来源于stack exchange，提问作者Mahdi Ezzaouia