python之无法在 docker 化环境中从 Flask 连接到 Kafka_编程设计

我正在尝试构建一个以 Kafka 作为界面的 Flask 应用程序。我使用了 Python 连接器，kafka-python和 Kafka 的 Docker 镜像，spotify/kafkaproxy .

下面是 docker-compose 文件。

version: '3.3' 
services: 
  kafka: 
    image: spotify/kafkaproxy 
    container_name: kafka_dev 
    ports: 
      - '9092:9092' 
      - '2181:2181' 
    environment: 
      - ADVERTISED_HOST=0.0.0.0 
      - ADVERTISED_PORT=9092 
      - CONSUMER_THREADS=1 
      - TOPICS=PROFILE_CREATED,IMG_RATED 
      - ZK_CONNECT=kafka7zookeeper:2181/root/path 
  flaskapp: 
    build: ./flask-app 
    container_name: flask_dev 
    ports: 
      - '9000:5000' 
    volumes: 
      - ./flask-app:/app 
    depends_on: 
      - kafka

下面是我用来连接到 kafka 的 Python 片段。在这里，我使用 Kafka 容器的别名 kafka 进行连接，因为 Docker 会负责将别名映射到它的 IP 地址。

from kafka import KafkaConsumer, KafkaProducer 
 
TOPICS = ['PROFILE_CREATED', 'IMG_RATED'] 
BOOTSTRAP_SERVERS = ['kafka:9092'] 
 
consumer = KafkaConsumer(TOPICS, bootstrap_servers=BOOTSTRAP_SERVERS)

我遇到了 NoBrokersAvailable 错误。由此，我了解到Flask应用程序找不到Kafka服务器。

Traceback (most recent call last): 
  File "./app.py", line 11, in <module> 
    consumer = KafkaConsumer("PROFILE_CREATED", bootstrap_servers=BOOTSTRAP_SERVERS) 
  File "/usr/local/lib/python3.6/site-packages/kafka/consumer/group.py", line 340, in __init__ 
    self._client = KafkaClient(metrics=self._metrics, **self.config) 
  File "/usr/local/lib/python3.6/site-packages/kafka/client_async.py", line 219, in __init__ 
    self.config['api_version'] = self.check_version(timeout=check_timeout) 
  File "/usr/local/lib/python3.6/site-packages/kafka/client_async.py", line 819, in check_version 
    raise Errors.NoBrokersAvailable() 
kafka.errors.NoBrokersAvailable: NoBrokersAvailable

其他观察结果:

我能够从 Flask 容器运行 ping kafka 并从 Kafka 容器获取数据包。
当我在本地运行 Flask 应用程序时，尝试通过设置 BOOTSTRAP_SERVERS = ['localhost:9092'] 连接到 Kafka 容器，它工作正常。

请您参考如下方法:

更新

如 cricket_007 所述，假设您正在使用下面提供的 docker-compose，您应该使用 kafka:29092 从另一个容器连接到 Kafka。所以你的代码看起来像这样:

from kafka import KafkaConsumer, KafkaProducer 
 
TOPICS = ['PROFILE_CREATED', 'IMG_RATED'] 
BOOTSTRAP_SERVERS = ['kafka:29092'] 
 
consumer = KafkaConsumer(TOPICS, bootstrap_servers=BOOTSTRAP_SERVERS)

结束更新

我建议您使用来自 Confluent Inc 的 Kafka 图像，他们有各种使用 docker-compose 的示例设置，可以随时使用，并且他们总是在更新它们。

试试这个:

--- 
version: '2' 
services: 
zookeeper: 
    image: confluentinc/cp-zookeeper:latest 
    environment: 
    ZOOKEEPER_CLIENT_PORT: 2181 
    ZOOKEEPER_TICK_TIME: 2000 
 
kafka: 
    image: confluentinc/cp-kafka:latest 
    depends_on: 
    - zookeeper 
    ports: 
    - 9092:9092 
    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 
 
flaskapp: 
    build: ./flask-app 
    container_name: flask_dev 
    ports: 
    - '9000:5000' 
    volumes: 
    - ./flask-app:/app

我用了这个docker-compose.yml并在顶部添加您的服务请注意:

The config used here exposes port 9092 for external connections to the broker i.e. those from outside the docker network. This could be from the host machine running docker, or maybe further afield if you've got a more complicated setup. If the latter is true, you will need to change the value 'localhost' in KAFKA_ADVERTISED_LISTENERS to one that is resolvable to the docker host from those remote clients

请务必查看其他示例，它们可能对您有用，尤其是在迁移到生产环境时:https://github.com/confluentinc/cp-docker-images/tree/5.0.1-post/examples

同样值得检查:

看来你需要指定 api_version 来避免这个错误。有关更多详细信息，请查看 here .

Version 1.3.5 of this library (which is latest on pypy) only lists certain API versions 0.8.0 to 0.10.1. So unless you explicitly specify api_version to be (0, 10, 1) the client library's attempt to discover the version will cause a NoBrokersAvailable error.

producer = KafkaProducer( 
    bootstrap_servers=URL, 
    client_id=CLIENT_ID, 
    value_serializer=JsonSerializer.serialize, 
    api_version=(0, 10, 1) 
)

这应该有效，有趣的是设置 api_version 是根据这个意外地解决了这个问题:

When you set api_version the client will not attempt to probe brokers for version information. So it is the probe operation that is failing. One large difference between the version probe connections and the general connections is that the former only attempts to connect on a single interface per connection (per broker), where as the latter -- general operation -- will cycle through all interfaces continually until a connection succeeds. #1411 fixes this by switching the version probe logic to attempt a connection on all found interfaces.

描述了实际问题here