Kafka python消费者在启动时读取所有消息

Question

我正在使用以下代码从主题中读取消息。我面临两个问题。 每当我启动消费者时，它正在读取队列中的所有消息？ 如何只读取未读消息？

from kafka import KafkaConsumer


consumer = KafkaConsumer('my-topic',
                         group_id='my-group',
                         bootstrap_servers=['localhost:9092'])
for message in consumer:
    consumer.commit() 
    # message value and key are raw bytes -- decode if necessary!
    # e.g., for unicode: `message.value.decode('utf-8')`
    print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,
                                          message.offset, message.key,
                                          message.value))

Answer 1

正如@Kenji 所说，您必须使用consumer.commit() 提交偏移量。如果您不想手动提交，您可以通过将enable_auto_commit=True 传递给您的KafkaConsumer 来启用自动提交。您可能还想调整auto_commit_interval_ms，这是每次自动提交之间的时间间隔（以毫秒为单位）。见这里：http://kafka-python.readthedocs.org/en/master/apidoc/KafkaConsumer.html。