Kafka python消费者没有从主题中读取消息答案

【问题标题】：Kafka python consumer not reading the message from topicsKafka python消费者没有从主题中读取消息
【发布时间】：2021-10-14 17:07:15
【问题描述】：

我正在尝试阅读指定主题中可用的消息，如下所示：

def consumer():
    consumer = KafkaConsumer('TOPIC_NAME',
                             enable_auto_commit=False,
                             value_deserializer=lambda m: json.loads(m.decode('ascii')),
                             bootstrap_servers=['xx.x.xx.xx:port'],
                             auto_offset_reset='earliest',
                             api_version=(20, 2, 1),
                             consumer_timeout_ms=1000
                             )

    #consumer.subscribe(['TOPIC_NAME'])
    for _ in range(10):
        msg = consumer.poll(0.05)
        if msg:
            dat = {
                'msg_value': msg.value(),  # This is the actual content of the message
                'msg_headers': msg.headers(),  # Headers of the message
                'msg_key': msg.key(),  # Message Key
                'msg_partition': msg.partition(),  # Partition id from which the message was extracted
                'msg_topic': msg.topic(),  # Topic in which Producer posted the message to
            }
            print(dat)
        print('No Msg found')

我没有收到任何错误，但它没有读取任何消息，只是打印 No Msg Found。当我检查该主题时，那里有消息。谁能帮我知道我是否在这里遗漏了什么。

【问题讨论】：

您应该将主题名称放在此处consumer.subscribe(['TOPIC_NAME']) 或者您可以删除此行，因为您在定义消费者时已经包含要订阅的主题作为参数。
@zweack 还是同样的问题
您是否尝试过指定组 ID？然后描述该组以确保消费者确实在处理数据？
@OneCricketeer 感谢您的回复。已经试过了，还是不行。
您是否尝试过其他客户端，例如 confluent 中的客户端？

标签： python apache-kafka kafka-consumer-api

【解决方案1】：

对于 Kafka 消费者，我认为更标准的实现是：

for message in consumer:
    print(msg)

如果你坚持使用consumer.poll()，请不要将time_out设置为0.05这么小的数字，使用while true持续轮询。

【讨论】：