Python中Kafka代码运行差异：交互Shell正常，文件执行失败

这是个非常典型的confluent-kafka使用误区，我来帮你拆解清楚为什么会出现这种差异：

为什么两种运行方式结果不同？ #

confluent-kafka的Producer采用异步发送的设计：当你调用produce()时，消息只是被加入到本地的发送队列中，真正的网络发送操作是由Producer内部的后台线程来完成的。

• 在Python交互Shell中：执行完produce()后，Shell会话会保持活跃，后台线程有足够的时间把队列里的消息发送到Kafka服务器，所以消费者能收到。
• 在脚本文件中：脚本执行完produce()语句后会立即退出整个程序，此时后台线程还没来得及把消息发送出去就被终止了，自然消费者收不到任何内容。

解决办法 #

有几种简单的方式可以让脚本中的Producer完成消息发送：

1. 调用flush()（最推荐） #

flush()方法会阻塞当前线程，直到所有待发送的消息都完成处理（成功发送到Kafka或发送失败）。修改后的代码如下：

from confluent_kafka import Producer
p = Producer({'bootstrap.servers': 'localhost:9092'})
p.produce('mytopic', key='hello', value='world')
p.flush()  # 等待所有消息发送完成

2. 使用poll()等待发送 #

poll()方法会让Producer处理后台的发送事件，你可以指定一个超时时间，给后台线程留出发送消息的窗口：

from confluent_kafka import Producer
p = Producer({'bootstrap.servers': 'localhost:9092'})
p.produce('mytopic', key='hello', value='world')
p.poll(1)  # 等待1秒，处理发送任务

3. 自定义回调+等待（适合需要监控发送状态的场景） #

如果你需要确认消息是否成功发送，可以给produce()添加回调函数，然后等待回调触发：

from confluent_kafka import Producer
import time

# 用来记录发送状态
delivered = False

def on_delivery(err, msg):
    global delivered
    if err:
        print(f"发送失败: {err}")
    else:
        print(f"消息已投递到 {msg.topic()} 分区 {msg.partition()}")
        delivered = True

p = Producer({'bootstrap.servers': 'localhost:9092'})
p.produce('mytopic', key='hello', value='world', callback=on_delivery)

# 循环等待直到回调触发
while not delivered:
    p.poll(0.1)
    time.sleep(0.1)

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