使用Python从融合的Kafka主题中消费数据并退出

Question

我正在尝试编写一个python代码来使用来自Confluent Kafka主题的数据，并作为测试项目的一部分执行数据验证。我能够读取数据，但是消费过程处于无限循环中，如果循环读取所有消息，则寻找退出的决策点。

参见下面的示例代码

conf = {'bootstrap.servers': "server:port", #
    'group.id': str(uuid.uuid1),
    'auto.offset.reset': 'earliest',
    'security.protocol': 'SASL_SSL',
    'sasl.mechanism': 'PLAIN',        
    'session.timeout.ms': 6000
    }
consumer = Consumer(conf) 
consumer.subscribe([topic], on_assign=on_assign)
try:
while True:
    msg=consumer.poll(timeout=2.0)

    # print(msg)

    if msg is None:
        print('msg is None')
        continue

    if msg.error().code() == KafkaError._PARTITION_EOF:
        print('End of partition reached {0}/{1}'
              .format(msg.topic(), msg.partition()))
        print( msg.error().code())
    else:
        print(msg.value())           

except KeyboardInterrupt:
   print("Interrupted through kb")

finally:
   consumer.close()

请建议决定是否阅读所有消息的最佳方法，以便我可以退出循环并关闭消费者。

Answer 1

根据定义，Apache Kafka 主题是无限的事件流。流没有“结束”，只有您可以选择定义的人为结束。

您需要在您的应用程序逻辑中定义这一点。例如，如果您在超过< code >的时间内没有收到信息