如何关闭卡夫卡消费者一旦消费完所有消息？

Question

我有以下程序来消耗所有来到Kafka的消息。

from kafka import KafkaConsumer 

consumer = KafkaConsumer('my_test_topic', 
         group_id='my-group', 
         bootstrap_servers=['my_kafka:9092']) 
for message in consumer: 
    consumer.commit() 
    print ("%s key=%s value=%s" % (message.topic,message.key, 
              message.value)) 
KafkaConsumer.close() 

使用上面的程序，我能够消耗所有来到卡夫卡的消息。但是，一旦所有消息都消耗完了，我想关闭不在发生的卡夫卡消费。我需要帮助。

Answer 1

如果我向KafkaConsumer对象提供consumer_timeout_ms参数，我现在可以关闭卡夫卡消费者。它接受以毫秒为单位的超时值。 以下是代码片段。

from kafka import KafkaConsumer 

consumer = KafkaConsumer('my_test_topic', 
         group_id='my-group', 
         bootstrap_servers=['my_kafka:9092'], 
         consumer_timeout_ms=1000) 
for message in consumer: 
    consumer.commit() 
    print ("%s key=%s value=%s" % (message.topic,message.key, 
              message.value)) 
KafkaConsumer.close() 

在上面的代码中，如果消费者在1秒内没有看到任何消息，它将关闭会话。

Answer 2

它看起来像你想要consumer.close（）而不是KafkaConsumer.close（）。它没有被记录为静态方法。