连接到在 Docker 中运行的 Kafka

Question

我在本地计算机上设置了一个单节点 Kafka Docker 容器，如 Confluence 文档（步骤 2-3）。

此外，我还公开了 Zookeeper 的端口 2181 和 Kafka 的端口 9092，以便我能够从本地计算机上运行的客户端连接到它们：

$ docker run -d \
    -p 2181:2181 \
    --net=confluent \
    --name=zookeeper \
    -e ZOOKEEPER_CLIENT_PORT=2181 \
    confluentinc/cp-zookeeper:4.1.0

$ docker run -d \
    --net=confluent \
    --name=kafka \
    -p 9092:9092 \
    -e KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 \
    -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092 \
    -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
    confluentinc/cp-kafka:4.1.0

问题： 当我尝试从主机，连接失败，因为它无法解析地址：kafka:9092。

这是我的 Java 代码：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("client.id", "KafkaExampleProducer");
props.put("key.serializer", LongSerializer.class.getName());
props.put("value.serializer", StringSerializer.class.getName());
KafkaProducer<Long, String> producer = new KafkaProducer<>(props);
ProducerRecord<Long, String> record = new ProducerRecord<>("foo", 1L, "Test 1");
producer.send(record).get();
producer.flush();

异常：

java.io.IOException: Can't resolve address: kafka:9092
    at org.apache.kafka.common.network.Selector.doConnect(Selector.java:235) ~[kafka-clients-2.0.0.jar:na]
    at org.apache.kafka.common.network.Selector.connect(Selector.java:214) ~[kafka-clients-2.0.0.jar:na]
    at org.apache.kafka.clients.NetworkClient.initiateConnect(NetworkClient.java:864) [kafka-clients-2.0.0.jar:na]
    at org.apache.kafka.clients.NetworkClient.ready(NetworkClient.java:265) [kafka-clients-2.0.0.jar:na]
    at org.apache.kafka.clients.producer.internals.Sender.sendProducerData(Sender.java:266) [kafka-clients-2.0.0.jar:na]
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:238) [kafka-clients-2.0.0.jar:na]
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:176) [kafka-clients-2.0.0.jar:na]
    at java.lang.Thread.run(Thread.java:748) [na:1.8.0_144]
Caused by: java.nio.channels.UnresolvedAddressException: null
    at sun.nio.ch.Net.checkAddress(Net.java:101) ~[na:1.8.0_144]
    at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:622) ~[na:1.8.0_144]
    at org.apache.kafka.common.network.Selector.doConnect(Selector.java:233) ~[kafka-clients-2.0.0.jar:na]
    ... 7 common frames omitted

问题：如何连接到在 Docker 中运行的 Kafka？我的代码是从主机运行的，而不是 Docker。

注意：我知道理论上我可以尝试 DNS 设置和 /etc/hosts 但这只是一种解决方法 - 它不应该是这样的。

这里也有类似的问题，但它是基于ches/kafka图像。我使用基于 confluenceinc 的图像，但它不一样。

Answer 1

tl;dr - 从容器到主机的简单端口转发将不起作用...主机文件（例如/etc/hosts *NIX 系统）不应修改以解决 Kafka 网络问题，因为此解决方案不可移植。

1) 您想要连接到哪个确切的 IP/主机名 + 端口？确保该值在经纪人。如果您看到诸如Connection to node -1 (localhost/127.0.0.1:9092)之类的错误，则意味着您的应用容器尝试连接到自身。您的应用程序容器是否也在运行 Kafka 代理进程？可能不会。

2) 确保作为 bootstrap.servers 一部分列出的服务器实际上是可解析的。例如 ping IP/主机名，使用 netcat 检查端口...如果您的客户端位于容器中，则需要从容器中执行此操作，不仅仅是您的主机。如果容器没有立即崩溃以访问其 shell，请使用 docker exec 。

3) 如果从主机而不是另一个容器运行进程，要验证主机上的端口是否正确映射，请确保 docker ps 表明 kafka 容器是从 0.0.0.0 映射的:<主机端口> ->/tcp。如果尝试从 Docker 网络外部运行客户端，则端口必须匹配。两个容器之间不需要端口转发；使用链接/docker网络

---

下面的答案使用 confluenceinc docker 镜像来解决所提出的问题，不是 wurstmeister/kafka。如果您设置了 KAFKA_ADVERTISED_HOST_NAME 变量，请将其删除（这是一个已弃用的属性）。相同的变量适用于 apache/kafka 图像。

以下部分尝试汇总使用另一个图像所需的所有详细信息。对于其他常用的 Kafka 镜像，都是在容器中运行的 Apache Kafka。
您只依赖于它的配置方式。以及哪些变量导致了这种情况。

apache/kafka

现在有官方图片了！在下面查找示例 Compose 文件的链接。

香肠/卡夫卡

自 2023 年 10 月起，DockerHub 中不再存在此内容。无论如何，2022 年之后就不再维护了。

请参阅有关监听器配置的自述文件部分，另外阅读他们的连接维基。

bitnami/kafka

如果您想要一个小容器，请尝试这些。这些图像比 Confluence 小得多，并且比 wurstmeister 维护得更好。 参考他们的监听器配置的自述文件。

debezium/kafka

此处提到了相关文档此处。

注意：已弃用公布的主机和端口设置。广告听众涵盖了两者。与 Confluence 容器类似，Debezium 可以使用 KAFKA_ 前缀的代理设置来更新其属性。

其他

• ubuntu/kafka 要求您通过 Docker 映像参数添加 --overrideadvertising.listeners=kafka:9092...我发现它比环境变量的可移植性差，所以不推荐
• spotify/kafka 已弃用且已过时。
• fast-data-dev 或 lensesio/box 非常适合一体化解决方案，具有架构注册表、Kafka Connect 等，但如果您 <仅仅想要卡夫卡。另外，这是一种在一个容器中运行多个服务的 Docker 反模式
• 您自己的 Dockerfile - 为什么？这些其他的东西是否不完整？从拉取请求开始，而不是从头开始。

如需补充阅读、功能齐全 docker-compose 和网络图，请参阅@rmoff 的此博客

回答

Confluence 快速入门 (Docker) 文档 假设所有生产和消费请求都将位于 Docker 网络内。

您可以通过在自己的容器中运行 Kafka 客户端代码（使用 Docker 网桥）来解决连接到 kafka:9092 的问题，但否则您需要添加更多环境变量来公开容器在外部，同时仍然在 Docker 网络中工作。

首先添加 PLAINTEXT_HOST:PLAINTEXT 协议映射，它将侦听器协议映射到 Kafka 协议

键：KAFKA_LISTENER_SECURITY_PROTOCOL_MAP
值：PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT

然后在不同的端口上设置两个通告的侦听器。 （kafka 这里指的是 docker 容器名称；它也可能被命名为 broker，因此请仔细检查您的服务 + 主机名）。

键：KAFKA_ADVERTISED_LISTENERS
值：PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092

注意此处的协议与上面协议映射设置的左侧值匹配

运行时容器，添加 -p 29092:29092 作为主机端口映射，并通告 PLAINTEXT_HOST 侦听器。

---

所以...（使用上述设置）

如果某些东西仍然不起作用，可以将KAFKA_LISTENERS设置为包含://0.0.0.0:< /code> 其中两个选项都与公布的设置和 Docker 转发的端口匹配




客户端位于同一台计算机上，而不是在容器中

广告本地主机和关联的端口将允许您在容器外部进行连接，正如您所期望的那样。

换句话说，当在 Docker 网络外部运行任何 Kafka 客户端（包括您可能在本地安装的 CLI 工具）时，请使用 localhost:29092 作为引导服务器（需要 Docker 端口转发） )

另一台计算机上的客户端

如果尝试从外部服务器连接，您需要公布主机的外部主机名/IP（例如 192.168.xy）以及/在本地主机的位置。
简单地通过端口转发来通告 localhost 是行不通的，因为 Kafka 协议仍将继续通告您已配置的侦听器。

如果不在同一本地网络中，此设置需要 Docker 端口转发和路由器端口转发（以及防火墙/安全组更改），例如，您的容器在云中运行并且您想要与其交互从您的本地计算机。

容器中的客户端（或另一个代理）位于同一主机上

这是最不容易出错的配置；您可以直接使用 DNS 服务名称。

在 Docker 网络中运行应用程序时，请使用 Docker 服务名称，例如 kafka:9092（请参阅上面广告的 PLAINTEXT 侦听器配置）作为引导服务器，就像任何其他 Docker 服务通信一样（不需要任何端口转发）

---

如果您使用单独的 docker run 命令或 Compose 文件，则需要定义共享网络手动使用撰写networks 部分或 docker network --create

---

查看使用 Kraft 的完整 Confluence 堆栈的示例 Compose 文件 或 更最小的一个（使用 Zookeeper ) 对于单个经纪人。

对于 apache/kafka 映像，Kafka Github 存储库。

如果使用多个代理，那么他们需要使用唯一的主机名+广告侦听器。 查看示例

相关问题

从 Docker 连接到主机上的 Kafka (ksqlDB)

附录

对于任何对 Kubernetes 部署感兴趣的人：

• 访问 Kafka
• Operator（推荐）：https://operatorhub.io/?keyword=Kafka
• Helm Artifact Hub：https://artifacthub.io/packages/search?ts_query_web=kafka&sort=stars&page=1

Answer 2

当你第一次连接到一个kafka节点时，它会返回所有的kafka节点和连接的url。然后你的应用程序将尝试直接连接到每个kafka。

问题始终是 kafka 会给你什么 url ？这就是为什么有 KAFKA_ADVERTISED_LISTENERS ，kafka 将使用它来告诉世界如何访问它。

现在，对于您的用例，有多个小问题需要考虑：

假设您设置了 plaintext://kafka:9092 ，

• 如果您的 docker compose 中有一个使用 kafka 的应用程序，那么这是可以的。该应用程序将从 kafka 获取可通过 docker 网络解析的 kafka URL。
• 如果您尝试从主系统或不在同一 Docker 网络中的另一个容器进行连接，则会失败，因为无法解析 kafka 名称。

==>为了解决这个问题，你需要有一个特定的 DNS 服务器，比如服务发现服务器，但这对于小东西来说是个大麻烦。或者您手动将 kafka 名称设置为每个 /etc/hosts 中的容器 ip

如果您设置 plaintext://localhost:9092

• 这将如果你有端口映射（启动kafka时为-p 9092:9092），那么在你的系统上就可以了。
• 如果你从容器上的应用程序（是否相同的docker网络）进行测试，这将会失败（localhost是容器本身而不是kafka）一）

==>如果您有此问题并希望在另一个容器中使用 kafka 客户端，解决此问题的一种方法是共享两个容器的网络（相同的 IP）

最后一个选项：在名称中设置 IP：plaintext://xyza :9092 （kafka 广告的 url 不能为 0.0.0.0，如文档 https://kafka.apache.org/documentation/#brokerconfigs_advertished.listeners ）

这将是每个人都可以...但是你怎么能得到 xyza 名字呢？

唯一的方法是在启动容器时硬编码此 IP：docker run .... --net confluence --ip 10.xyz ...。请注意，您需要将 ip 调整为 confluence 子网中的一个有效 ip。

Answer 3

在 Zookeeper

1. docker 容器运行之前 --name Zookeeper -p 2181:2181 Zookeeper

在 kafka

2. docker 容器运行之后 --name kafka -p 9092:9092 -e KAFKA_ZOOKEEPER_CONNECT=192.168.8.128:2181 -e confluenceinc/cp-kafka

KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://ip_address_of_your_computer_but_not_localhost!!!:9092 -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 kafka 消费者和生产者配置中的

@Bean
public ProducerFactory<String, String> producerFactory() {
    Map<String, Object> configProps = new HashMap<>();
    configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.8.128:9092");
    configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    return new DefaultKafkaProducerFactory<>(configProps);
}

@Bean
public ConsumerFactory<String, String> consumerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.8.128:9092");
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "group_id");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    return new DefaultKafkaConsumerFactory<>(props);
}

我按照这些规定运行我的项目。祝你好运，伙计。

Answer 4

这允许我在 M1 Mac 上的 Kafka 应用程序中访问 localhost:9092

Key: KAFKA_ADVERTISED_LISTENERS
Value: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092

以及端口转发：

ports
   - "9092:9092"

最后，再次，对于我的设置，我必须以这种方式设置侦听器密钥

Key: KAFKA_LISTENERS
Value: PLAINTEXT://0.0.0.0:29092,PLAINTEXT_HOST://0.0.0.0:9092

Answer 5

解决此问题的最简单方法是使用 -h 选项向代理添加自定义主机名

docker run -d \
    --net=confluent \
    --name=kafka \
    -h broker-1 \
    -p 9092:9092 \
    -e KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 \
    -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092 \
    -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
    confluentinc/cp-kafka:4.1.0

，然后编辑 /etc/hosts

127.0.0.1   broker-1

并使用：

props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker-1:9092");

Answer 6

      KAFKA_BROKER_ID: 1
      KAFKA_ADVERTISED_HOST_NAME: kafka:9092
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

此配置工作正常，

请确保从 docker 内部

从外部容器 localhost:9092 连接 kafka:29092

完整工作 docker compose 配置

    version: "3.3"

    services:
      zookeeper:
        image: confluentinc/cp-zookeeper:6.2.0
        container_name: zookeeper
        networks:
          - broker-kafka
        ports:
          - "2181:2181"
        environment:
          ZOOKEEPER_CLIENT_PORT: 2181
          ZOOKEEPER_TICK_TIME: 2000
          ALLOW_ANONYMOUS_LOGIN: yes
        volumes:
          - ./bitnami/zookeeper:/bitnami/zookeeper
    
      kafka:
        image: confluentinc/cp-kafka:6.2.0
        container_name: kafka
        networks:
          - broker-kafka
        depends_on:
          - zookeeper
        ports:
          - "9092:9092"
        expose:
          - "9092"
        environment:
          KAFKA_BROKER_ID: 1
          KAFKA_ADVERTISED_HOST_NAME: kafka:9092
          KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
          KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
          KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
          KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
          KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
          # KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"
          # KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
          # KAFKA_CONFLUENT_LICENSE_TOPIC_REPLICATION_FACTOR: 1
          KAFKA_CONFLUENT_BALANCER_TOPIC_REPLICATION_FACTOR: 1
          # KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
          KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
        volumes:
          - ./bitnami/kafka:/bitnami/kafka
    
      kafdrop:
        image: obsidiandynamics/kafdrop
        container_name: kafdrop
        ports:
          - "9000:9000"
        expose:
          - "9000"
        networks:
          - broker-kafka
        environment:
          KAFKA_BROKERCONNECT: "PLAINTEXT://kafka:29092"
          JVM_OPTS: "-Xms16M -Xmx48M -Xss180K -XX:-TieredCompilation -XX:+UseStringDeduplication -noverify"
          SPRING_PROFILES_ACTIVE: "dev"
        depends_on:
          - kafka
          - zookeeper
    
      consumer:
        container_name: consumer
        build:
          context: ./consumer
          dockerfile: Dockerfile
        environment:
          - KAFKA_TOPIC_NAME=app
          - KAFKA_SERVER=kafka
          - KAFKA_PORT=29092
        ports:
          - 8001:8001
        restart: "always"
        depends_on:
          - zookeeper
          - kafka
          - publisher
          - kafdrop
        networks:
          - broker-kafka
    
      publisher:
        container_name: publisher
        build:
          context: ./producer
          dockerfile: Dockerfile
        environment:
          - KAFKA_TOPIC_NAME=app
          - KAFKA_SERVER=kafka
          - KAFKA_PORT=29092
        ports:
          - 8000:8000
        restart: "always"
        depends_on:
          - zookeeper
          - kafka
          - kafdrop
        networks:
          - broker-kafka
        volumes:
          - ./testproducer:/producer
    
    networks:
      broker-kafka:
        driver: bridge