kafka镜像制造器2自定义分区器

要自定义分区器，可以继承org.apache.kafka.clients.producer.Partitioner类，并实现其中的partition()方法。以下是一个示例：

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.record.InvalidRecordException;
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();

        if (keyBytes == null || !(key instanceof String)) {
            throw new InvalidRecordException("Invalid key");
        }

        // 自定义分区逻辑
        String keyString = (String) key;
        int partition = Math.abs(keyString.hashCode() % numPartitions);

        return partition;
    }

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

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

然后，在创建Kafka生产者时，使用自定义分区器：

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

import java.util.Properties;

public class KafkaProducerExample {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.example.CustomPartitioner");

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

        // 发送消息到自定义分区
        producer.send(new ProducerRecord<>("my-topic", "my-key", "my-value"));

        producer.close();
    }
}

在上述示例中，CustomPartitioner类继承了Partitioner接口，并实现了partition()方法来自定义分区逻辑。在创建Kafka生产者时，通过ProducerConfig.PARTITIONER_CLASS_CONFIG配置项指定使用自定义分区器。

请注意，自定义分区器的类名应该是完整的类路径。在示例中，CustomPartitioner类的完整类路径为com.example.CustomPartitioner。