永远运行kafka consumer（新的consumer API）

Question

我在Apache Kafka上构建了一个排队系统。应用程序将为特定的Kafka主题生成消息，在使用者端，我必须使用为该主题生成的所有记录。
我使用新的Java使用者API编写了consumer。代码看起来像

  Properties props = new Properties();  
                     props.put("bootstrap.servers", kafkaBrokerIp+":9092");  
                     props.put("group.id",groupId);  
                     props.put("enable.auto.commit", "true");
                     props.put("session.timeout.ms", "30000");
                     props.put("auto.offset.reset", "earliest");
        props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer(props);
                     consumer.subscribe(Arrays.asList("consumertest"));  
                     while (true) {  
                         ConsumerRecords<String, String> records = consumer.poll(100);  
                         for (ConsumerRecord<String, String> record : records){  
                             System.out.println("Data recieved : "+record.value());  
                             }  
                     }

这里我需要永远运行消费者，这样生产者推入kafka主题的任何记录都应该立即消费和处理。
所以我的困惑是，使用无限while循环（像示例代码中那样）消费数据是正确的方法吗？

Answer 1

虽然使用无限循环是可以的，但在Kafka消费者文档中可以找到一种稍微优雅的方法，如下所示：

public class KafkaConsumerRunner implements Runnable {
    private final AtomicBoolean closed = new AtomicBoolean(false);
    private final KafkaConsumer consumer;

    public void run() {
        try {
            consumer.subscribe(Arrays.asList("topic"));
            while (!closed.get()) {
                ConsumerRecords records = consumer.poll(10000);
                // Handle new records
            }
        } catch (WakeupException e) {
            // Ignore exception if closing
            if (!closed.get()) throw e;
        } finally {
           consumer.close();
        }
    }

    // Shutdown hook which can be called from a separate thread
    public void shutdown() {
        closed.set(true);
        consumer.wakeup();
    }
}

这使您可以选择使用钩子进行优雅的关机。