java - Kafka消费者没有回应

Question

我是卡夫卡世界的新手，我正在尝试使用生产者和消费者的现有示例创建简单的 java Kafka 应用程序。

生产者给出结果:

Sent:Message 0
Sent:Message 1
Sent:Message 2
Sent:Message 3
.............

但是在Consumer端，代码执行正确，没有错误，但控制台上没有显示结果 Consumer console

制作人:

public class ProducerTest {
public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", "192.168.33.10:9092");
props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
Producer<String, String> producer = null;
try {
  producer = new KafkaProducer<>(props);
  for (int i = 0; i < 100; i++) {
    String msg = "Message " + i;
    producer.send(new ProducerRecord<String, String>("HelloKafkaTopic", msg));
    System.out.println("Sent:" + msg);
  }
} catch (Exception e) {
  e.printStackTrace();

} finally {
  producer.close();
}

}

}

消费者:

 public class ConsumerTest {

public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", "192.168.33.10:9092");
props.put("group.id", "group-1");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("auto.offset.reset", "smallest");
props.put("session.timeout.ms", "30000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);
kafkaConsumer.subscribe(Arrays.asList("HelloKafkaTopic"));
while (true) {
  ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
  for (ConsumerRecord<String, String> record : records) {
    System.out.println("Partition: " + record.partition() + " Offset: " + record.offset()
        + " Value: " + record.value() + " ThreadID: " + Thread.currentThread().getId());
  }
}

 }  }

有什么建议，先谢谢了

Answer 1

一些建议:
1.运行kafka-console-consumer脚本 --from-beginning验证您是否可以看到生成的消息。
2. 更改props.put("auto.offset.reset", "smallest");至props.put("auto.offset.reset", "earliest");
3.调用poll时不要使用这么小的超时时间。相反，使用相对较大的超时，例如 3000。即使使用 Integer.MAX_VALUE 也是合理的。