使用Python自定义KafkaProducer时触发断言错误求助

我来帮你排查这个断言错误的问题，先拆解下你的代码里存在的几个关键问题：

• 数据格式不兼容：kafka-python库的KafkaProducer.send()方法，默认要求value参数是字节类型，你直接传入字符串"text"会触发序列化相关的断言失败。解决这个的方式有两种：

  1. 直接把字符串转成字节，比如value=b"text"
  2. 初始化Producer时指定value_serializer，让它自动帮你完成字符串到字节的转换

• 端口配置大概率错误：你写的7092不是Kafka的默认监听端口，默认端口是9092，除非你特意修改过Kafka的配置文件，否则这个端口根本连不上服务，这也会引发断言类的错误。

• 冗余的循环代码：for x in range(10)已经会自动让x从0迭代到9，你在循环里加的x=x+1完全多余，不仅没用还容易让人误解逻辑，建议删掉。

• 导入语句格式错误：你的两行导入写在同一行里了，Python会报语法错误，正确的写法应该分开换行。

下面是修正后的完整代码：

from kafka import KafkaProducer
from kafka.errors import KafkaError

# 初始化Producer，指定正确端口和序列化器
producer = KafkaProducer(
    bootstrap_servers=['127.0.0.1:9092'],
    value_serializer=lambda x: x.encode('utf-8')  # 自动将字符串编码为UTF-8字节
)

# 循环发送消息
for _ in range(10):
    topic = "kafkatopic"
    producer.send(topic=topic, value="text")
    # 调用flush确保消息被立即发送，避免后台异步缓存导致的异常
    producer.flush()

# 捕获并处理发送过程中的Kafka错误
try:
    producer.flush()  # 等待所有未完成的消息发送完毕
except KafkaError as e:
    print(f"消息发送失败: {str(e)}")

额外提醒：

• 确保你的Kafka服务已经正常启动，并且能通过127.0.0.1:9092访问到
• 如果你的Kafka没有开启自动创建主题的配置，需要先手动创建kafkatopic这个主题，否则会报错

内容的提问来源于stack exchange，提问作者qing zhangqing