Kafka生产者发消息但消费者无法消费（Python+AWS EC2环境）

针对你遇到的kafka-python生产者发送消息后无法被接收的问题，按以下步骤排查解决：

1. 确保消息实际发送完成 #

kafka-python的send()方法是异步执行的，仅返回Future对象不代表消息已成功写入Kafka。需要显式触发消息提交：

修改生产者代码，在send()后添加flush()或等待Future结果：

# 方式1：强制刷新发送缓存
producer.send('demo_test', value='{"hello":"World"}')
producer.flush()  # 确保消息发送完成

# 方式2：等待发送结果并捕获异常（更推荐，能排查发送失败原因）
future = producer.send('demo_test', value='{"hello":"World"}')
try:
    record_metadata = future.get(timeout=10)
    print(f"消息发送到分区 {record_metadata.partition}，偏移量 {record_metadata.offset}")
except Exception as e:
    print(f"发送失败：{str(e)}")

2. 检查Kafka监听配置（关键） #

Kraft模式下，需要确保Kafka的监听地址配置正确，让外部客户端能正确连接并接收消息：

1. 找到Kraft的配置文件（通常是kraft/server.properties）
2. 修改以下配置：

# 允许所有IP访问
listeners=PLAINTEXT://0.0.0.0:9092
# 对外暴露EC2公网IP，让客户端能正确获取Kafka地址
advertised.listeners=PLAINTEXT://50.19.162.182:9092

3. 重启Kraft节点：

# 停止Kraft
bin/kafka-server-stop.sh
# 启动Kraft
bin/kafka-server-start.sh -daemon kraft/server.properties

3. 修正序列化/反序列化逻辑（避免解析问题） #

你当前的生产者将JSON字符串再次用json.dumps()处理，会导致消息内容被双重转义，消费者解析后得到的是字符串而非JSON对象。建议直接传入字典：

生产者代码修改：

producer.send('demo_test', value={"hello":"World"})  # 直接传字典，由serializer转为JSON字符串

消费者代码保持不变即可正确解析为字典。

4. 优化消费者配置（确保能接收历史消息） #

如果消费者是新的消费组，默认不会接收之前发送的消息，需添加auto_offset_reset='earliest'配置：

consumer = KafkaConsumer(
    'demo_test',
    bootstrap_servers=['50.19.162.182:9092'],
    value_deserializer=lambda v: json.loads(v.decode('utf-8')),
    group_id='test_consumer_group',  # 显式指定消费组ID
    auto_offset_reset='earliest'  # 从最早的偏移量开始消费
)

5. 确认EC2安全组配置 #

确保EC2安全组的入站规则允许Jupyter所在IP访问9092端口（TCP协议），虽然你能创建生产者实例，但如果规则限制可能导致消息无法传输。

内容的提问来源于stack exchange，提问作者Archit Mishra