我对 python 中使用 kafka-python 模块的一些代码有疑问,我真的不知道为什么它不起作用。在我像快速入门页面建议的那样初始化 Zookeeper&Kafka 服务器并在我的计算机上创建主题后,我尝试使用 Termux 在我的智能手机上运行以下代码:
from json import dumps, loads
from kafka import KafkaProducer, KafkaConsumer
class CommunicationLayer:
def __init__(self, node_id):
self.output_publisher = KafkaProducer(bootstrap_servers=['192.168.1.103:9092'], value_serializer=lambda x:
dumps(x).encode('utf-8'))
self.window_receiver = KafkaConsumer("node_{}".format(node_id), bootstrap_servers=['192.168.1.103:9092'] , value_deserializer=lambda x: loads(x.decode('utf-8')))
...
(code that doesn't really help)
...
(IP指向运行kafka服务器的机器)。一旦我运行此代码,就会收到 NoBrokerAvailable 错误,据我所知,只有在您尚未初始化服务器时才会出现该错误。我错过了什么吗?
我尝试在我的智能手机上运行以下代码
注意:Kafka 客户端不适合在网络连接不可靠的环境中运行。
IP指向运行kafka服务器的机器
ZooKeeper和 Kafka是否在这台机器上对外公开,而不仅仅是在本地监听?
同样,您需要将advertised.listeners
Kafka 属性中的属性设置为该外部地址,以便任何客户端能够在引导连接后实际连接。
本文收集自互联网,转载请注明来源。
如有侵权,请联系[email protected] 删除。
我来说两句