如何關閉卡夫卡消費者一旦消費完所有消息？

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（）。它沒有被記錄爲靜態方法。