docker-compose环境下kafka-python基础生产消费不生效问题

核心原因 #

kafka-python 库的 KafkaProducer.send() 是 异步非阻塞方法：调用后消息不会立刻发送到Broker，而是先存入客户端本地缓冲区，由后台IO线程批量异步发送。你的生产端代码调用send()后直接运行结束退出进程，缓冲区里的消息还没来得及发送就被回收，消费端自然收不到内容。

修复步骤 #

• 补全生产者逻辑，等待消息实际发送完成
  不要调用完send()就直接退出进程，需要显式等待发送结果，同时建议明确指定Broker端口、捕获发送异常：

  from kafka import KafkaProducer
  # 建议明确写全端口，避免默认端口适配问题
  producer = KafkaProducer(bootstrap_servers='broker:9092')
  # send方法返回future对象，调用get()阻塞等待发送结果，超时直接抛错
  send_result = producer.send('my-topic', b'my message!').get(timeout=10)
  print(f"消息已发送到分区{send_result.partition}，偏移量{send_result.offset}")
  producer.close()
  

  也可以在send后调用producer.flush()，会阻塞直到所有缓冲区消息都发送完成。

• 调整消费者配置，避免偏移量/配置问题
  消费者默认配置存在几个容易踩坑的点，测试阶段可以显式指定参数规避：

  from kafka import KafkaConsumer
  consumer = KafkaConsumer(
      'my-topic',
      bootstrap_servers='broker:9092',
      group_id='test-consumer-group', # 必须指定消费者组，否则偏移量无法正常提交
      auto_offset_reset='earliest', # 没有已提交偏移量时，从最早的消息开始消费
      consumer_timeout_ms=15000 # 测试场景加超时，避免无消息时永久阻塞
  )
  for msg in consumer:
      # 消息value是字节类型，解码后方便查看
      print(f"收到消息：{msg.value.decode('utf-8')}")
  

  注意：必须 先启动消费端程序，再运行生产端发送消息，否则如果消费端在消息发送之后才启动，默认latest策略下会跳过之前已存在的消息。

• 容器网络连通性校验
  由于服务基于Docker Compose部署，先在运行Python程序的容器内执行nc -zv broker 9092命令，确认可以正常连通Kafka服务端口，避免网络隔离导致的发送失败。如果调用send().get()时抛出连接异常，优先排查Docker网络、Kafka的advertised.listeners配置是否正确。

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