Docker+Kafka+Python环境问题：消息无法收发及连接异常求助

一、先确认Docker Compose的Kafka配置 #

Kafka的网络监听配置是核心问题，以下是经过验证的docker-compose.yml示例，替换你当前的配置：

version: '3.8'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.4.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - "2181:2181"

  kafka:
    image: confluentinc/cp-kafka:7.4.0
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
      - "29092:29092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

关键说明：KAFKA_ADVERTISED_LISTENERS分别配置了容器内部（kafka:29092）和外部主机（localhost:9092）的访问地址，确保内外都能正常连接。

二、Ubuntu下kafka-console-producer卡住的解决 #

1. 先创建目标主题（如果还没创建）：

kafka-topics.sh --create --topic test_topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

2. 执行正确的生产者命令：

kafka-console-producer.sh --broker-list localhost:9092 --topic test_topic

3. 排查异常：
   • 用docker-compose ps确认zookeeper、kafka容器状态为Up
   • 用docker-compose logs kafka查看日志，搜索是否有连接zookeeper失败的报错

三、Windows下相关问题解决 #

1. 注意：Kafka不是HTTP服务，不能用浏览器访问localhost
   你遇到的ERR_EMPTY_RESPONSE是因为错误用浏览器访问Kafka端口，正确的方式是用代码或命令行工具连接。

2. 验证Python生产者/消费者代码
   生产者代码（确保调用flush()发送缓冲区消息）：

import json
from kafka import KafkaProducer

def json_serializer(data):
    return json.dumps(data).encode('utf-8')

producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=json_serializer
)

# 读取JSON文件并发送
with open('your_data.json', 'r') as f:
    data = json.load(f)
producer.send('test_topic', data)
producer.flush()  # 必须执行，确保消息发送完成
producer.close()

消费者代码：

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'test_topic',
    bootstrap_servers=['localhost:9092'],
    auto_offset_reset='earliest',  # 读取最早未消费的消息
    value_deserializer=lambda x: json.loads(x.decode('utf-8'))
)

for message in consumer:
    print(message.value)

3. 前置依赖检查：确保已安装kafka-python库：

pip install kafka-python

四、通用排查步骤 #

1. 测试端口连通性：
   • Ubuntu：nc -zv localhost 9092
   • Windows（PowerShell）：Test-NetConnection localhost -Port 9092
2. 查看已创建的主题：

kafka-topics.sh --list --bootstrap-server localhost:9092

3. 重启Docker服务：
   • Ubuntu：sudo systemctl restart docker
   • Windows：在Docker Desktop中点击重启按钮

内容的提问来源于stack exchange，提问作者dr-whyy