KAFKA中的几个消费者的循环不起作用

Question

我有5个独立的docker图像:1个用于kafka经纪人，1个动物园管理员，1个生产者和2个消费者。我通过生产者向主题发布消息。基本上，我希望消息将在循环算法中使用，因此，为此，我使用相同的< code>group.id定义了消费者，并将< code > partition . assignment . strategy 的配置添加为< code > org . Apache . Kafka . clients . consumer . roundrobinsignator ，

但我发现只有一个消费者收到了所有的消息。

我的生产者代码：

import java.util.List;
import java.util.Properties;

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


public class DiscoveryKafkaProducer{
   Producer<String, String> producer;

   public DiscoveryKafkaProducer(Properties configs) {
   producer = new KafkaProducer<String, String>(configs);   
}

   public void send(String topic, List<String> records) {
   for(String record: records){
       producer.send(new ProducerRecord<String, String>(topic, record));
   }
   producer.flush();
}

我的消费者代码:

public static void main(String[] args) {
    String server = "lshraga-ubuntu-sp-nac:9092";
    Properties consumerConfigs = new Properties();
    consumerConfigs.put("bootstrap.servers", server);
    consumerConfigs.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    consumerConfigs.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    consumerConfigs.put("group.id", "discovery");
    consumerConfigs.put("client.id", "discovery");
    consumerConfigs.put("partition.assignment.strategy", "org.apache.kafka.clients.consumer.RoundRobinAssignor");
    List<String> list = new ArrayList<String>();
    DiscoveryKafkaConsumer consumer1 = new DiscoveryKafkaConsumer(Collections.singletonList(topicName), consumerConfigs);
    
     try {
         while (true) {
            System.out.println("Start to consume");
            consumer1.poll(1000L);
         }
    } catch (InterruptedException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
    
 public class DiscoveryKafkaConsumer {

       Consumer<String, String> consumer;
       Integer id;
       public DiscoveryKafkaConsumer(List<String> topics, Properties configs) {
       consumer = new KafkaConsumer<String, String>(configs);
       consumer.subscribe(topics);
       }


   public DiscoveryKafkaConsumer(int i, List<String> topics, Properties configs) {
       consumer = new KafkaConsumer<String, String>(configs);
       consumer.subscribe(topics);
       this.id = i;
}

   public void poll(long timeout) throws InterruptedException {

       ConsumerRecords<String,String> records = consumer.poll(timeout);
       System.out.println("Hey!Consumer #" + id + "got records:" + records);
       Map<String, List<String>> results = new HashMap<String, List<String>>();
       records.forEach((cr) -> {
    

   System.out.println("cr.topic()=" + cr.topic());
       List<String> list = results.get(cr.topic()); 
       if(list == null) {
           list = new ArrayList<>(); 
           results.put(cr.topic(), list);
           }
            list.add(cr.value());
            System.out.println("list=" + list);
       });

}

我使用Kafka客户端版本0.11.0.0。

我需要添加/配置什么才能以轮循机制方式使用消息？

Answer 1

仍然不可能从Java增加分区数。有KIP-195和相关PR应该包含在下一个版本中：https://cwiki.apache.org/confluence/display/KAFKA/KIP-195：管理员客户端。.