Kafka制作人将消息发布到单个分区

Question

在我的本地系统中，我已经启动了一个单独的Kafka实例，旁边还有动物园管理员。Zookeper和kafka服务器都运行在默认端口上。

我创建了一个主题“test”，复制因子为1，因为我只有一个kafka实例正在运行。

同时，我还创建了两个分区。

但是当我使用java kafka-client jar创建一个生产者时，即使我对消息使用不同的键，生产者也会将所有消息推送到同一个分区，因为所有消息都是在同一个使用者上接收的。

分区也不是静态的，它在每次运行我的生产者时都在不断变化。

我已经尝试了相同的场景，其中一个生产者从命令提示符启动，配置与我使用java代码提供给kafka-client producer的配置完全相同。命令提示符生成器似乎运行良好，但代码生成器正在将所有消息推送到同一个分区。

public class KafkaProducerParallel {


public static void main(String[] args) throws InterruptedException, 
ExecutionException {

    Properties properties = new Properties();
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 
"localhost:9092");
    properties.put(ProducerConfig.CLIENT_ID_CONFIG, "parallelism- 
 producer");
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, 
 StringSerializer.class);
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, 
LongSerializer.class);


    Producer<String, Long> parallelProducer = new KafkaProducer<> 
(properties);

    for(long i=0;i<100;i++) {

        ProducerRecord<String, Long> producerRecord;

        if(i<50) {
            producerRecord = new ProducerRecord<String, 
 Long>("second-topic", "Amoeba", i);
        }else {
            producerRecord = new ProducerRecord<String, 
 Long>("second-topic", "Bacteria", i);
        }

        RecordMetadata recordMetadata = 
  parallelProducer.send(producerRecord).get();

        System.out.printf("Sent record : with key %s and value 
 %d to partition %s", producerRecord.key(), producerRecord.value(), 
 recordMetadata.partition());
        System.out.println();
    }

    parallelProducer.close();


}

}

根据文档，kafka broker通过使用密钥（生成密钥的散列）来决定将特定消息放入哪个分区。我在一段时间后更改了我的记录的键，但消息仍然每次都要到同一个分区。

代码的控制台输出示例：

  Sent record : with key Amoeba and value 0 to partition 1
  Sent record : with key Amoeba and value 1 to partition 1
  Sent record : with key Amoeba and value 2 to partition 1
  Sent record : with key Amoeba and value 3 to partition 1
  Sent record : with key Amoeba and value 4 to partition 1
  Sent record : with key Amoeba and value 5 to partition 1
  Sent record : with key Amoeba and value 6 to partition 1
  Sent record : with key Amoeba and value 7 to partition 1
  Sent record : with key Amoeba and value 8 to partition 1
  Sent record : with key Amoeba and value 9 to partition 1
  Sent record : with key Amoeba and value 10 to partition 1
  Sent record : with key Amoeba and value 11 to partition 1
  Sent record : with key Amoeba and value 12 to partition 1
  Sent record : with key Amoeba and value 13 to partition 1

 Sent record : with key Bacteria and value 87 to partition 1
 Sent record : with key Bacteria and value 88 to partition 1
 Sent record : with key Bacteria and value 89 to partition 1
 Sent record : with key Bacteria and value 90 to partition 1
 Sent record : with key Bacteria and value 91 to partition 1
 Sent record : with key Bacteria and value 92 to partition 1
 Sent record : with key Bacteria and value 93 to partition 1
 Sent record : with key Bacteria and value 94 to partition 1
 Sent record : with key Bacteria and value 95 to partition 1
 Sent record : with key Bacteria and value 96 to partition 1
 Sent record : with key Bacteria and value 97 to partition 1
 Sent record : with key Bacteria and value 98 to partition 1
 Sent record : with key Bacteria and value 99 to partition 1

Answer 1

一切按预期进行。

在kafkaproducer（用来确定分区）使用的特定case Partitioner，为两个键计算相同的分区:amoeba和bacteria。默认情况下，KafkaProducer使用org.apache.kafka.clients.producer.internals.DefaultPartitioner。

建议：改变键或增加分区数。