java - Kafka Java Consumer SDK 长拉取而不使用 while

Question

我尝试使用 Kafka Java SDK 来实现消费者，但是我看到的大多数消费者示例都使用 while(true) 循环，并在循环内调用 consume 方法来获取一条消息。

while (true) {
            final ConsumerRecords<Long, String> consumerRecords =
                    consumer.poll(1000);
            if (consumerRecords.count()==0) {
                noRecordsCount++;
                if (noRecordsCount > giveUp) break;
                else continue;
            }
            consumerRecords.forEach(record -> {
                System.out.printf("Consumer Record:(%d, %s, %d, %d)\n",
                        record.key(), record.value(),
                        record.partition(), record.offset());
            });
            consumer.commitAsync();
        }

我想知道是否有任何优雅的方法可以在不使用 while 循环的情况下处理这个问题，这类似于以下 RabbitMQ 实现:

Consumer consumer = new DefaultConsumer(channel) {
      @Override
      public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
          throws IOException {
        String message = new String(body, "UTF-8");
        System.out.println(" [x] Received '" + message + "'");
      }
    };
    channel.basicConsume(QUEUE_NAME, true, consumer);

Answer 1

您可以尝试使用 Spring-kafka，它具有 @KafkaListener 注释并使方法监听主题，了解更多信息 here

因为在apache-kafka中没有优雅的方法来使方法作为主题的监听器，因为消费者需要在一定的时间间隔内轮询记录，需要循环中的代码

@KafkaListener(topics = "topicName", group = "foo")
public void listen(String message) {
System.out.println("Received Messasge in group foo: " + message);
}