kafka生产者-如何知道哪个键会分配到哪个分区？

在Kafka生产者中，键（key）的分配到分区（partition）是通过Kafka的分区器（Partitioner）来完成的。默认情况下，Kafka使用的是基于键的哈希分区器（DefaultPartitioner），该分区器会根据键的哈希值将消息分配到不同的分区。

要知道哪个键会分配到哪个分区，可以通过自定义分区器来实现。下面是一个示例代码，展示了如何自定义分区器来知道键的分区：

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.utils.Utils;

import java.util.List;
import java.util.Map;

public class CustomPartitioner implements Partitioner {

    @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();

        // 根据键的哈希值计算分区索引
        int partitionIndex = Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;

        return partitionIndex;
    }

    @Override
    public void close() {
        // 关闭资源
    }

    @Override
    public void configure(Map<String, ?> configs) {
        // 配置分区器
    }
}

在上述代码中，我们实现了Partitioner接口，并重写了partition方法。在该方法中，我们使用murmur2哈希算法来计算键的哈希值，并将其取正（使用toPositive方法），然后对分区数取模得到分区索引。

接下来，我们需要在生产者代码中指定使用自定义分区器。示例如下：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class ProducerExample {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost: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.CustomPartitioner");

        Producer<String, String> producer = new KafkaProducer<>(props);

        String topic = "my-topic";
        String key = "my-key";
        String value = "my-value";

        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);

        producer.send(record);

        producer.close();
    }
}

在上述代码中，我们通过props.put("partitioner.class", "com.example.CustomPartitioner")来指定使用自定义分区器。

通过以上代码示例，我们可以自定义分区器来知道键会分配到哪个分区。