如何消费Python生产的Kafka数据？消费者运行无输出如何解决

1 核心原因分析 #

你遇到的问题绝大多数是因为生产者异步发送逻辑未处理、消费者配置不匹配两个原因导致，以下是逐步骤修复方案：

2 第一步：修复生产者代码 #

KafkaProducer.send() 是异步方法，你脚本循环结束后会直接退出，内存中还未刷入Kafka的消息会直接丢失。需要在循环结束后增加强制刷写和资源释放逻辑：

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

producer = KafkaProducer(
    value_serializer = lambda x:dumps(x).encode('utf-8'),
    bootstrap_servers = ["localhost:9092"]
)
for i in range(1,100):
    producer.send('test', value = {"hello" : i})
    sleep(0.001)
# 新增以下两行
producer.flush()
producer.close()

3 第二步：修复消费者代码 #

你当前配置group_id=None会导致偏移量提交逻辑异常，auto_offset_reset='earliest'无法正常生效，同时建议补充和生产者匹配的反序列化逻辑，方便直接查看消息内容：

from json import loads
from kafka import KafkaConsumer

consumer = KafkaConsumer(
    "test",
    bootstrap_servers = ["localhost:9092"],
    auto_offset_reset = 'earliest',
    enable_auto_commit = True,
    # 修改group_id为任意非空字符串即可
    group_id = 'test-group-1',
    # 新增和生产者匹配的反序列化逻辑
    value_deserializer = lambda x: loads(x.decode('utf-8'))
)

for message in consumer:
    # 可以直接打印value查看内容，也可以打印完整message查看元数据
    print(message.value)
    # print(message)

4 第三步：前置依赖检查 #

运行代码前先确认以下环境配置正常：

• Kafka服务已经正常启动，9092端口没有被防火墙拦截
• 已经创建test主题，可以通过Kafka自带命令验证：

  # 查看所有主题，确认test存在
  kafka-topics.sh --list --bootstrap-server localhost:9092
  # 如果不存在则创建主题
  kafka-topics.sh --create --topic test --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
  

5 运行顺序 #

先启动消费者脚本，再启动生产者脚本，即可看到消费者控制台输出对应的消息内容。

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