0

我正在尝试部署 2 个 Spring Boot 应用程序（kafka Producer 和 Consumer）。当我将 Producer 部署到 docker 时一切正常，但是当我部署时，我的 Consumer 不起作用，因为没有与 kafka 容器的连接。

日志告诉我这个错误

2019-11-17 05:32:22.644  WARN 1 --- [main] o.a.k.c.NetworkClient: [Consumer clientId=consumer-1, groupId=exampleGroup] Connection to node -1 could not be established. Broker may not be available.

我的 docker-compose.yml 是

version: '3'

services:

  zookeeper:
    image: wurstmeister/zookeeper
    container_name: zookeeper
    restart: always
    ports:
      - 2181:2181

  kafka:
    image: wurstmeister/kafka
    container_name: kafka
    restart: always
    ports:
      - 9092:9092
    depends_on:
      - zookeeper
    links:
      - zookeeper:zookeeper
    environment:
      KAFKA_ADVERTISED_HOST_NAME: localhost
      KAFKA_ADVERTISED_PORT: 9092
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_CREATE_TOPICS: "topic1:1:1"

在我的 KafkaConfig 课程中：

@EnableKafka
@Configuration
public class KafkaConfig {

    @Bean
    public ConsumerFactory<String, String> consumerFactory(){
        Map<String, Object> config = new HashMap<>();

        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConstants.KAFKA_BROKERS);
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "exampleGroup");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
      //  config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, KafkaConstants.ENABLE_AUTO_COMMIT_CONFIG);
        config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, KafkaConstants.OFFSET_RESET_EARLIER);
       // config.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, KafkaConstants.SESSION_TIMEOUT_MS);

        return new DefaultKafkaConsumerFactory<>(config);
    }

和常量类

public class KafkaConstants {

    public static String KAFKA_BROKERS = "localhost:9092";
    public static Integer MESSAGE_COUNT=1000;
    public static String TOPIC_NAME="demo";
    public static String GROUP_ID_CONFIG="exampleGroup";
    public static Integer MAX_NO_MESSAGE_FOUND_COUNT=100;
    public static String OFFSET_RESET_LATEST="latest";
    public static String OFFSET_RESET_EARLIER="earliest";
    public static Integer MAX_POLL_RECORDS=1;
    public static Integer SESSION_TIMEOUT_MS = 180000;
    public static Integer REQUEST_TIMEOUT_MS_CONFIG = 181000;
    public static String ENABLE_AUTO_COMMIT_CONFIG = "false";
    public static Integer AUTO_COMMIT_INTERVAL_MS_CONFIG = 8000;
}

当我在我的计算机上安装 zookepper 和 kafka 并使用 intellij 运行这 2 个 spring boot 应用程序时工作正常。问题是当我部署到本地 docker 时。

你能帮我么？

更新

更新我的 docker-compose：

version: '3'

services:

  zookeeper:
    image: wurstmeister/zookeeper
    container_name: zookeeper
    restart: always
    ports:
      - 2181:2181

  kafka:
    image: wurstmeister/kafka
    container_name: kafka
    restart: always
    ports:
      - 9092:9092
    depends_on:
      - zookeeper
    links:
      - zookeeper:zookeeper
    environment:
      KAFKA_ADVERTISED_HOST_NAME: kafka
      KAFKA_ADVERTISED_PORT: 9092
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_CREATE_TOPICS: "ACC_GROUP_CREATE:1:1"

  consumer:
    image: micro1
    container_name: micro1
    depends_on:
      - kafka
    restart: always
    ports:
      - 8088:8088
    depends_on:
      - kafka
    links:
      - kafka:kafka

  producer:
    image: micro2
    container_name: micro2
    depends_on:
      - kafka
    restart: always
    ports:
      - 8087:8087
    depends_on:
      - kafka
    links:
      - kafka:kafka

工作正常！基于@hqt 的响应，但我不知道为什么需要添加这些消费者/生产者行

4

2 回答 2

1

由于KAFKA_ADVERTISED_HOST_NAME属性的问题。这是解释为什么 Kafka 需要广告地址的文档。

关键是当你运行一个客户端时，你传递给它的代理就是它要去的地方，并从中获取关于集群中代理的元数据。它将连接到以读取/写入数据的实际主机和 IP 是基于代理 在初始连接中传回的数据——即使它只是一个节点并且返回的代理与连接到的代理相同。

当您设置KAFKA_ADVERTISED_HOST_NAME为本地主机时：

  • 您的应用程序在“Intellij”上运行，这意味着在主机环境中运行。该主机创建 Kafka 的容器，因此来自 localhost:9092 的访问将指向 Kafka 的容器。
  • 当您的应用在容器内运行时，localhost:9092 表示容器本身。所以毫无意义。（这个容器甚至没有任何监听 9092 端口的进程）

在容器环境中运行 Web 应用程序时，将KAFKA_ADVERTISED_HOST_NAME属性更新为将起作用。kafka请注意，您的 Web 应用程序和 kafka 容器都必须在 docker 的网络上。

这是使用 Wurstmeister 的镜像运行 Kafka 集群的建议 docker-compose。

version: "2"
services:
  zookeeper:
    image: wurstmeister/zookeeper
    container_name: zookeeper
    ports:
      - 2181:2181

  kafka:
    image: wurstmeister/kafka
    container_name: kafka
    ports:
      - 9092:9092
    depends_on:
      - zookeeper
    environment:
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_HOST_NAME: kafka
      KAFKA_ADVERTISED_PORT: 9092
      KAFKA_CREATE_TOPICS: "topic1:1:1"

   web_app:
    # your definition of the web_app goes  here

然后，您可以连接到kafka:9092容器环境内地址上的 Kafka 代理。

于 2019-11-17T20:41:18.227 回答
0

这是一个常见问题，您需要阅读和理解的权威文档是https://www.confluent.io/blog/kafka-listeners-explained

我正在复制它的 tl;dr: 以供参考：

您需要将 Advertisementd.listeners（或 >KAFKA_ADVERTISED_LISTENERS 如果您使用 Docker >images）设置为外部地址（主机/IP），以便客户端 > 可以正确连接到它。否则，他们将尝试连接 > 到内部主机地址——如果无法访问，那么就会出现问题。”

于 2019-11-17T21:54:55.043 回答