使用 Java kafka 客户端进行自定义分区

Question

我能够用 Java 编写一个 kafka 示例应用程序。它有 3 个主题，并且 pub/sub 运行良好。但无法将这些主题分配到不同的分区。

我的消费者

public class Consumers extends Thread {
    private static final List<String> TOPIC_LIST = Arrays.asList("topic1", "topic2", "topic3");
    private static final List<TopicPartition> PARTITION_LIST = 
Arrays.asList(new TopicPartition(TOPIC_LIST.get(0), 1), new TopicPartition(TOPIC_LIST.get(1), 2));

    private void message() {
        Properties consumerProperties = KafkaProperties.getConsumerProperties();
        org.apache.kafka.clients.consumer.KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProperties);
        consumer.assign(PARTITION_LIST);
        Logger.debug("Kafka IP : " + consumerProperties.getProperty("bootstrap.servers"));
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(100);
                for (ConsumerRecord<String, String> record : records) {
                    process(record.topic(), record.value());
                }
            }
        } catch (Exception e) {
            Logger.error("error while consuming : " + e.getMessage());
            e.printStackTrace();
        } finally {
            consumer.close();
        }
    }

    private void process(String topic, String value) {
        KafkaProcessor.process(topic, value);
    }

    @Override
    public void run() {
        message();
    }
}

我的自定义分区

public class CustomPartitioner implements Partitioner {
private static Map<String, Integer> partitionMap;

@Override
public void configure(Map<String, ?> configs) {
    System.out.println("Inside CustomPartitioner.configure " + configs);
    partitionMap = new HashMap<>();
    for (Map.Entry<String, ?> entry : configs.entrySet()) {
        if (entry.getKey().startsWith("partitions.")) {
            String keyName = entry.getKey();
            String value = (String) entry.getValue();
            int partitionId = Integer.parseInt(keyName.substring(11));
            partitionMap.put(value, partitionId);
        }
    }
}

@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
    List partitions = cluster.availablePartitionsForTopic(topic);
    String valueStr = (String) value;
    String name = ((String) value).split(":")[0];
    if (partitionMap.containsKey(name)) {
        //If the country is mapped to particular partition return it
        return partitionMap.get(name);
    } else {
        //If no country is mapped to particular partition distribute between remaining partitions
        int noOfPartitions = cluster.topics().size();
        return value.hashCode() % noOfPartitions + partitionMap.size();
    }
}

public void close() {
}}

我的制作人

public void producer(String topic, String message) {
    Producer<String, String> producer = new KafkaProducer<>(KafkaProperties.getProducerProperties());
    try {
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, null, message);
        producer.send(producerRecord);
        producer.close();
    } catch (Exception e) {
        Logger.error("kafka message publish error: ", e);
    }
}

当我运行这段代码时。我收到以下警告。

[warn] o.a.k.c.p.ProducerConfig - The configuration partitions.2 = partition2 was supplied but isn't a known config.
[warn] o.a.k.c.p.ProducerConfig - The configuration partitions.1 = partition1 was supplied but isn't a known config.

我的生产者属性如下，

 properties.put("bootstrap.servers", "127.0.0.1:9092");
 properties.put("acks", "all");
 properties.put("retries", 0);
 properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, CustomPartitioner.class.getCanonicalName());
 properties.put("partitions.1", "partition1");
 properties.put("partitions.2", "partition2");
 properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
 properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

使用这些属性和代码，我无法发送或接收消息。我该如何解决这个问题。

Answer 1

创建分区是通过主题配置来配置的，而不是生产者配置。对于您想要的现有主题：

bin/kafka-topics.sh --zookeeper <ZK_HOST> --alter --topic <TOPIC_NAME> --partitions <NUM_PARTITIONS>

您的警告非常明确：partitions.1 和 partitions.2 不是有效属性，因此 Kafka 让您知道它不知道如何处理它们。您应该删除这两行。

更重要的是，您不需要CustomPartitioner。 Kafka 通过密钥的哈希值自动跨分区分发数据。因此，如果您将国家/地区作为键添加到您的记录中：

ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, message.split(":")[0], message);

那么你可以保证来自同一个国家的所有数据都将进入同一个分区，并且你可以删除你的整个CustomPartitioner类。同时删除consumer.assign(PARTITION_LIST);；再次，Kafka 会为您管理。