如何使用kafka-python在同WiFi下两台Ubuntu设备间发送自定义payload

核心问题 #

你提供的代码出错原因主要有两点：

• 未给KafkaProducer配置value_serializer，Kafka默认仅支持字节类型的消息内容，直接传入字典对象会触发类型错误
• 未配置发送确认逻辑，可能存在消息滞留在本地缓冲区，未实际发送到Kafka broker的问题

可直接运行的代码示例 #

生产者代码 #

from time import sleep
from json import dumps
from kafka import KafkaProducer

producer = KafkaProducer(
    bootstrap_servers=['172.20.10.7:9092'],
    # 新增序列化配置：将字典转成JSON字符串后编码为UTF-8字节流
    value_serializer=lambda x: dumps(x).encode('utf-8')
)

for e in range(100):
    data = {'number': e}
    future = producer.send('demo01', value=data)
    # 等待发送结果确认，避免消息丢失
    try:
        metadata = future.get(timeout=10)
        print(f"发送成功：topic={metadata.topic}，分区={metadata.partition}，偏移量={metadata.offset}")
    except Exception as err:
        print(f"发送失败：{err}")
    sleep(1)

producer.flush()
producer.close()

消费者代码 #

from kafka import KafkaConsumer
from json import loads

consumer = KafkaConsumer(
    'demo01',
    bootstrap_servers=['172.20.10.7:9092'],
    auto_offset_reset='earliest',
    enable_auto_commit=True,
    auto_commit_interval_ms=1000,
    group_id='my-group',
    value_deserializer=lambda x: loads(x.decode('utf-8'))
)

try:
    for message in consumer:
        print(f"收到消息：{message.value}，偏移量={message.offset}")
except KeyboardInterrupt:
    print("停止消费")
finally:
    consumer.close()

附加排查项 #

如果运行代码仍有问题，按以下顺序检查配置：

• 确认172.20.10.7上Kafka的server.properties配置中，listeners和advertised.listeners没有绑定127.0.0.1，允许外部设备访问
• 确认9092端口没有被防火墙拦截，可在生产者设备上执行telnet 172.20.10.7 9092测试连通性
• 确认demo01 topic已手动创建，或者Kafka开启了auto.create.topics.enable=true配置
• 两台设备的kafka-python版本保持一致，推荐安装稳定版：pip install kafka-python==2.0.2

内容的提问来源于stack exchange，提问作者Edward Chang