Confluent.Kafka如何基于MetriceType分区键实现消息分区并保证顺序

你遇到的所有消息分区始终为0的问题，通常由以下几个原因导致：

• 你创建的目标Topic实际分区数为1。Kafka默认创建Topic的分区数由Broker端num.partitions参数控制，默认值通常为1，此时无论怎么设置消息Key，所有消息只能写入唯一的0号分区。
• 调用生产者Produce/ProduceAsync方法时，显式指定了Partition参数为0，显式指定的分区优先级会高于Key和自定义分区器逻辑。
• 消息Key存在为空的情况，若Key为null，Confluent.Kafka默认会使用粘滞分区策略，大批量消息可能会集中写入单个分区。

方案1：使用默认分区器实现按MetriceType分区（最简单） #

不需要自定义分区器就能满足你的需求：

1. 首先根据业务中MetriceType的数量、吞吐量提前规划分区数，通过AdminClient创建对应分区数的Topic，分区数创建后也可后续扩容，仅扩容不会重新分配历史数据。
2. 保持你现有的将MetriceType作为消息Key的逻辑，Confluent.Kafka默认的分区器会对Key做Murmur2哈希后对分区数取模，保证同一个Key的所有消息一定会写入同一个分区，天然满足同MetriceType的消息顺序性要求。
   对应验证代码示例：

// 生产者配置示例
var producerConfig = new ProducerConfig
{
    BootstrapServers = "your-kafka-broker:9092",
    // 显式指定分区器为默认的Murmur2哈希，和Java客户端默认策略对齐
    Partitioner = Partitioner.Murmur2Random
};

using var producer = new ProducerBuilder<string, string>(producerConfig).Build();

// 构造消息
var metrice = new Metrice { MetriceType = "cpu_usage", MetriceValue = 0.75 };
var message = FormatMessage(metrice.MetriceType, JsonSerializer.Serialize(metrice));

// 发送消息，不要显式指定Partition
var deliveryResult = await producer.ProduceAsync("your-topic-name", message);
// 打印分区号验证
Console.WriteLine($"消息写入分区：{deliveryResult.Partition.Value}");

方案2：自定义分区器实现专属分区逻辑 #

如果需要实现指定MetriceType写入固定分区的强绑定逻辑，Confluent.Kafka的.NET版本支持自定义分区器，实现方式如下：

// 自定义分区器实现
public class CustomMetricePartitioner : IPartitioner
{
    // 你可以提前配置MetriceType和分区的映射关系
    private readonly Dictionary<string, int> _typePartitionMap = new Dictionary<string, int>
    {
        {"cpu_usage", 0},
        {"memory_usage", 1},
        {"disk_usage", 2}
    };
    private readonly int _defaultPartition = 3;

    public Partition Partition(string topic, int partitionCount, ReadOnlySpan<byte> key, bool keyIsNull, ReadOnlySpan<byte> value, bool valueIsNull, bool isTransactional)
    {
        if (keyIsNull) return _defaultPartition;
        var keyStr = Encoding.UTF8.GetString(key);
        return _typePartitionMap.TryGetValue(keyStr, out var partition) 
            ? new Partition(partition) 
            : new Partition(_defaultPartition);
    }

    public void Dispose() {}
}

// 生产者配置时指定自定义分区器
var producerConfig = new ProducerConfig
{
    BootstrapServers = "your-kafka-broker:9092",
};
using var producer = new ProducerBuilder<string, string>(producerConfig)
    .SetPartitioner("your-topic-name", new CustomMetricePartitioner())
    .Build();

注意：使用自定义分区器时需要保证Topic的分区数大于你映射配置的最大分区号，否则会出现写入错误

如果不想提前规划分区数，也可以基于Kafka的主题路由逻辑实现：将不同MetriceType的消息写入不同的Topic，消费时按Topic消费即可保证顺序，这种方案更适合MetriceType数量较少且固定的场景。

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