WSL2 Ubuntu环境下Python Kafka本地生产消息无法消费问题咨询

问题原因 #

1. KafkaProducer的send方法是异步调用，默认不会立即发送消息，消息会暂存在本地缓冲区，你示例的生产者脚本执行完send后会直接退出，消息根本没有发送到Kafka broker。
2. 你的生产者代码直接传递字符串类型的消息值，kafka-python默认要求消息值为bytes类型，未做类型转换/序列化配置的情况下发送会触发异常，但由于脚本退出过快你可能看不到报错。
3. 你的消费者代码中bootstrap_servers配置项里localhost和9092之间多了一个空格，会导致地址解析异常。
4. KafkaConsumer默认从最新的偏移量开始消费，如果你先执行生产者再启动消费者，没有配置auto_offset_reset参数的话，不会消费启动前发送的历史消息。

修复后的代码 #

生产者代码 producer.py

from kafka import KafkaProducer
import logging

# 开启日志，方便排查连接、发送异常
logging.basicConfig(level=logging.INFO)

producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    # 配置序列化器，自动将字符串转为utf-8编码的bytes
    value_serializer=lambda v: str(v).encode('utf-8')
)
# 发送后等待broker返回确认结果，超时时间设为10秒
try:
    send_future = producer.send('sample', value='Hello, World!')
    send_result = send_future.get(timeout=10)
    print(f"消息发送成功，偏移量：{send_result.offset}")
except Exception as e:
    print(f"消息发送失败：{e}")
finally:
    # 强制刷写缓冲区并关闭连接
    producer.flush()
    producer.close()

消费者代码 consumer.py

from kafka import KafkaConsumer
import logging

logging.basicConfig(level=logging.INFO)

consumer = KafkaConsumer(
    'sample',
    # 移除地址中多余的空格
    bootstrap_servers=['localhost:9092'],
    # 配置从最早的未消费消息开始读取
    auto_offset_reset='earliest',
    # 配置反序列化器，自动将bytes转为字符串
    value_deserializer=lambda x: x.decode('utf-8'),
    # 配置消费者组id，便于记录消费偏移量
    group_id='test-group'
)

for message in consumer:
    print(f"收到消息：{message.value}")

验证与后续注意事项 #

• 测试时先运行消费者脚本，再运行生产者脚本，即可看到消费者正常输出消息。
• 后续连接远程Kafka时，需要确认远程Kafka服务的advertised.listeners配置为你WSL2环境能正常访问的公网/内网地址，而不是localhost或者Kafka节点的内部地址，避免出现NoBrokersAvailable错误。

内容的提问来源于stack exchange，提问作者Tristan Tran