Python Kafka用户在消息到达时不读取消息

Question

我刚开始学Kafka，Kafka-蟒蛇。在下面的代码中，我试图在消息到达时读取它们。但出于某种原因，消费者似乎要等到一定数量的消息积累后才能获取它们。

我最初以为是因为正在批量出版的制片人。当我运行“kafka-console-consumer--bootstrap-servers--topic”时，我可以看到发布后收到的每一条消息（就像在consumer控制台上看到的那样）

def run():
    success_consumer = KafkaConsumer('success_logs',
                                     bootstrap_servers=KAFKA_BROKER_URL,
                                     group_id=None,
                                     fetch_min_bytes=1,
                                     fetch_max_bytes=10,
                                     enable_auto_commit=True)
    #dummy poll
    success_consumer.poll()
    for msg in success_consumer:
        print(msg)

    success_consumer.close()

有人能指出用KafKaconsumer改变了什么配置吗？为什么它不能读取像“Kafka-控制台-消费者”这样的消息？

Answer 1

KafkaConsumer类还有一个FETCH_MAX_WAIT_MS参数。您应该将其设置为0

success_consumer = KafkaConsumer(...,fetch_max_wait_ms=0)