Python 生产者可以通过 shell 发送，但不能通过 .py

Question

我有一个运行并经过测试的 Kafka 集群，我正在尝试使用 Python 脚本向代理发送消息。这在我使用 Python3 shell 并调用生产者方法时有效，但是当我将这些相同的命令放入 python 文件并执行它时 - 脚本似乎挂起。

我正在为消费者和生产者使用 kafka-python 库。当我使用 Python3 shell 时，我可以使用 Kafka GUI 工具 2.0.4 看到消息出现在主题中我在 python 代码中尝试了各种循环和语句，但似乎没有什么能使它 'run' 完成。

>>>from kafka import KafkaProducer
>>>producer = KafkaProducer(bootstrap_servers='BOOTSTRAP_SRV:9092')
>>>producer.send('MyTopic', b'Has this worked?')
>>>>>><kafka.producer.future.FutureRecordMetadata object at 0x7f7af9ece048>

这有效并且字节出现在代理主题数据中。

当我将与上面相同的代码放入 python .py 文件并使用 Python3 执行时，它完成了，但没有数据发送到 Kafka 代理。也没有显示错误。

from kafka import KafkaProducer
producer = KafkaProducer(bootstrap_servers='BOOTSTRAP_SRV:9092')
producer.send('MyTopic', b'Some Data to Check')

Answer 1

如您所见，它 returns 是一个未来。

Kafka 客户端将批处理记录，它们不会立即一次发送一条记录，要做到这一点，您需要等待或刷新生产者缓冲区，以便它在应用程序退出之前发送.换句话说，交互式终端将生产者数据保存在内存中，运行在后台，而另一种方式丢弃该数据

As the docs, show

future = producer.send(...)

try:
    record_metadata = future.get(timeout=10)
except KafkaError:
    # Decide what to do if produce request failed...
    log.exception()
    pass

或者如果您不关心元数据或抢占未来，则只需输入 producer.flush()。

Python 生产者可以通过 shell 发送，但不能通过 .py

Python Producer can send via shell, but not .py

python

python-3.x

apache-kafka

kafka-python