python - 在 python 中使用来自不同容器的 Kafka 消息

Question

下面是我的 docker-compose 文件:

   version: "3"
    services:
      app:
        image: app
        restart: always
        ports:
          - "8001:8081"
        depends_on:
          - kafka
          - zookeeper
          - consumer
        environment:
          - KAFKA_HOST = kafka

  zookeeper:
    image: "wurstmeister/zookeeper:latest"
    ports:
        - "2181:2181"
    hostname: zookeeper   

  kafka:
    image: "wurstmeister/kafka:2.12-2.2.0"
    ports:
      - "9092:9092"
    hostname: kafka
    links: 
     - zookeeper:zookeeper
    environment:
      KAFKA_CREATE_TOPICS: "Topic01:2:2" #TOPIC:PARTITON:REPLICATION
      KAFKA_ZOOKEEPER_CONNECTION_TIMEOUT_MS: "60000"
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
      KAFKA_LISTENERS: 'PLAINTEXT://:9092'
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka:9092'

    volumes:
      - /var/run/docker.sock:/var/run/docker.sock

  consumer:
    image: consumer:latest
    build:
      context: ./consumer
    ports:
      - "8283:8283"

文本流是使用 app 的容器生成的图像并能够生成消息。

通过在kafka容器内部导航验证:

docker exec -it <CONTAINER ID of Kafka Image> /bin/bash

当我在容器中手动运行以下脚本时

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic Topic01 --from-beginning

我可以看到正在生成的文本 block 。

但是当我尝试从图像的容器中读取相同的数据时 consumer , 它让我空白,

尝试过的方法:

docker exec在消费者容器内，运行以下 python 代码:

from kafka import KafkaConsumer
consumer = KafkaConsumer('Topic01', bootstrap_servers='kafka:9092')
for messages in consumer:
        print(messages)

它什么都不打印。

是我的docker-compose文件有误还是python代码有误？

Answer 1

根据 Kafka 的监听器文档:

Listener List - Comma-separated list of URIs we will listen on and the listener names. If the listener name is not a security protocol, listener.security.protocol.map must also be set. Specify hostname as 0.0.0.0 to bind to all interfaces. Leave hostname empty to bind to default interface. Examples of legal listener lists: PLAINTEXT://myhost:9092,SSL://:9091 CLIENT://0.0.0.0:9092,REPLICATION://localhost:9093

您正在将 KAFKA_LISTENERS 设置为 PLAINTEXT://:9092，因此它绑定(bind)到默认接口(interface)，这可能无法从 Kafka 容器的外部访问(虽然同时它与使用 localhost 的控制台消费者一起工作良好)。尝试将 KAFKA_LISTENERS 指定为 PLAINTEXT://0.0.0.0:9092，以检查您的消费者是否开始消费消息。