MacBook Docker环境下Kafka Python生产消费功能异常求助

1. 修正Kafka监听配置（最常见原因） #

官方Kafka镜像默认监听配置仅适配容器内部通信，主机客户端连接时会拿到容器内部地址，导致无法正常交互。需在Docker Compose的Kafka服务中添加以下环境变量：

services:
  kafka:
    image: apache/kafka:latest
    environment:
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,PLAINTEXT_HOST://0.0.0.0:29092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1  # 单节点部署需设置为1
    ports:
      - "9092:29092"  # 主机端口映射到容器的PLAINTEXT_HOST监听端口
    volumes:
      - ./kafka-data:/var/lib/kafka/data

配置说明：

• KAFKA_ADVERTISED_LISTENERS明确告知外部客户端（主机）通过localhost:9092访问Kafka
• 容器内部服务间通信使用kafka:9092（对应Docker Compose服务名）

2. 调整Python客户端脚本配置 #

确保脚本中bootstrap_servers指向主机的localhost:9092，并添加异常捕获机制避免静默失败：

生产者脚本（带结果验证）

from kafka import KafkaProducer
import json

try:
    producer = KafkaProducer(
        bootstrap_servers=['localhost:9092'],
        value_serializer=lambda v: json.dumps(v).encode('utf-8'),
        acks='all',  # 等待所有副本确认，确保消息发送状态可追踪
        retries=3
    )
    future = producer.send('test_topic', {'content': 'test message'})
    # 等待发送结果，超时抛出异常
    result = future.get(timeout=10)
    print(f"消息发送成功: 分区={result.partition()}, 偏移量={result.offset()}")
except Exception as e:
    print(f"消息发送失败: {str(e)}")
finally:
    producer.close()

消费者脚本（确保能读取历史消息）

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'test_topic',
    bootstrap_servers=['localhost:9092'],
    auto_offset_reset='earliest',  # 从Topic最早的消息开始消费
    group_id='test_consumer_group',
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

print("等待接收消息...")
for message in consumer:
    print(f"收到消息: {message.value}")

3. 验证网络连通性 #

在主机终端执行以下命令，确认Kafka端口是否正常开放：

nc -zv localhost 9092

如果输出Connection to localhost port 9092 [tcp/XmlIpcRegSvc] succeeded!则端口正常；若失败，检查Docker容器是否正常运行，以及端口映射配置是否正确。

4. 检查Topic可用性 #

进入Kafka容器，执行命令查看Topic状态：

docker exec -it <kafka-container-name> kafka-topics.sh --describe --topic test_topic --bootstrap-server localhost:9092

重点查看ReplicationFactor和ISR字段，确保ISR列表包含至少一个副本（单节点部署时ReplicationFactor需设为1）。

5. 验证库版本兼容性 #

若上述步骤无效，可尝试更换Python Kafka客户端库测试，比如使用confluent-kafka：

pip install confluent-kafka

测试脚本：

from confluent_kafka import Producer, Consumer, KafkaError

# 生产者测试
p = Producer({'bootstrap.servers': 'localhost:9092'})
def delivery_cb(err, msg):
    if err:
        print(f"发送失败: {err}")
    else:
        print(f"发送成功: {msg.topic()} [{msg.partition()}]")

p.produce('test_topic', value='confluent test message', callback=delivery_cb)
p.flush()

# 消费者测试
c = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'confluent_test_group',
    'auto.offset.reset': 'earliest'
})
c.subscribe(['test_topic'])
print("等待消息...")
while True:
    msg = c.poll(1.0)
    if msg is None:
        continue
    if msg.error():
        if msg.error().code() == KafkaError._PARTITION_EOF:
            continue
        else:
            print(msg.error())
            break
    print(f"收到消息: {msg.value().decode('utf-8')}")
c.close()

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