Lock Condition.notify 抛出 java.lang.IllegalMonitorStateException

Question

Alfredo Morales

Asked: 2025-04-22 05:57:14 +0800 CST2025-04-22 05:57:14 +0800 CST 2025-04-22 05:57:14 +0800 CST

Kafka 消费者总是尝试 127.0.0.1:9092

772

我有一张 Kafka 在容器中运行的图像：

 service-kafka:
    container_name: pepito-kafka-server
    hostname: pepito
    image: apache/kafka:3.7.1
    networks: 
      - pepito-network
    ports:
        - 8484:9092
    extra_hosts:
      - "docker.internal:127.0.0.1"
    healthcheck:
      test: /opt/kafka/bin/kafka-cluster.sh cluster-id --bootstrap-server pepito:9092 || exit 1
      interval: 1s
      timeout: 60s
      retries: 60

正如你所见，我有一个用于 Kafka 的自定义端口，

现在，我尝试从我的 springboot 应用程序中像这样使用它：

@Configuration
public class KafkaTopicConfig {

    @Value(value = "${spring.kafka.bootstrap-servers}")
    private String bootstrapAddress;

    @Bean
    KafkaAdmin kafkaAdmin() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
        return new KafkaAdmin(configs);
    }
}

和，

@Configuration
public class KafkaProducerConfig {

    @Value(value = "${spring.kafka.bootstrap-servers}")
    private String bootstrapAddress;

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

    @Bean
    KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

和，

@EnableKafka
@Configuration
public class KafkaConsumerConfig {

    @Value(value = "${pepito.group-id}")
    private String groupId;

    @Value(value = "${spring.kafka.bootstrap-servers}")
    private String bootstrapAddress;

    @Bean
    ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return new DefaultKafkaConsumerFactory<>(props);
    }

    @Bean
    ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {

        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }

    @KafkaListener(topics = "${spring.application.name}", groupId = "${pepito.group-id}")
    public void listenGroupFoo(String message) {
        System.out.println("Received Message in group " + groupId + ": " + message);
    }

}

我有一个生产者和一个消费者，因为这个应用程序的想法是发送、接收和处理消息，

现在，我收到这些错误：

[36morg.apache.kafka.clients.Metadata       [0;39m [2m:[0;39m [Consumer clientId=consumer-pepito.app-1, groupId=pepito.app] Cluster ID: 5L6g3nShT-eMCtK--X85sw
[36mo.a.k.c.c.internals.ConsumerCoordinator [0;39m [2m:[0;39m [Consumer clientId=consumer-pepito.app-1, groupId=pepito.app] Discovered group coordinator localhost:9092 (id: 2147483546 rack: null)
[36mo.a.k.c.c.internals.ConsumerCoordinator [0;39m [2m:[0;39m [Consumer clientId=consumer-pepito.app-1, groupId=pepito.app] (Re-)joining group
[36morg.apache.kafka.clients.NetworkClient  [0;39m [2m:[0;39m [Consumer clientId=consumer-pepito.app-1, groupId=pepito.app] Node 2147453646 disconnected.
[2m2025-04-21T15:43:36.068-06:00[0;39m [33m WARN[0;39m [35m4088[0;39m [2m--- [pepito-app] [ntainer#0-0-C-1] [0;39m[36morg.apache.kafka.clients.NetworkClient  [0;39m [2m:[0;39m [Consumer clientId=consumer-pepito.app-1, groupId=pepito.app] Connection to node 2147483636 (localhost/127.0.0.1:9092) could not be established. Node may not be available.
[36mo.a.k.c.c.internals.ConsumerCoordinator [0;39m [2m:[0;39m [Consumer clientId=consumer-pepito.app-1, groupId=pepito.app] Group coordinator localhost:9092 (id: 2147483646 rack: null) is unavailable or invalid due to cause: null. isDisconnected: true. Rediscovery will be attempted.
[36mo.a.k.c.c.internals.ConsumerCoordinator [0;39m [2m:[0;39m [Consumer clientId=consumer-pepito.app-1, groupId=pepito.app] Request joining group due to: rebalance failed due to 'null' (DisconnectException)
[36morg.apache.kafka.clients.NetworkClient  [0;39m [2m:[0;39m [Consumer clientId=consumer-pepito.app-1, groupId=pepito.app] Node 1 disconnected.
[2m2025-04-21T15:43:36.087-06:00[0;39m [33m WARN[0;39m [35m4088[0;39m [2m--- [pepito-app] [ntainer#0-0-C-1] [0;39m[36morg.apache.kafka.clients.NetworkClient  [0;39m [2m:[0;39m [Consumer clientId=consumer-pepito.app-1, groupId=pepito.app] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Node may not be available.
[36morg.apache.kafka.clients.NetworkClient  [0;39m [2m:[0;39m [Consumer clientId=consumer-pepito.app-1, groupId=pepito.app] Node 1 disconnected.
[2m2025-04-21T15:43:36.199-06:00[0;39m [33m WARN[0;39m [35m4088[0;39m [2m--- [pepito-app] [ntainer#0-0-C-1] [0;39m[36morg.apache.kafka.clients.NetworkClient  [0;39m [2m:[0;39m [Consumer clientId=consumer-pepito.app-1, groupId=pepito.app] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Node may not be available.
[36morg.apache.kafka.clients.NetworkClient  [0;39m [2m:[0;39m [Consumer clientId=consumer-pepito.app-1, groupId=pepito.app] Node 1 disconnected.
[2m2025-04-21T15:43:36.302-06:00[0;39m [33m WARN[0;39m [35m4088[0;39m [2m--- [pepito-app] [ntainer#0-0-C-1] [0;39m[36morg.apache.kafka.clients.NetworkClient  [0;39m [2m:[0;39m [Consumer clientId=consumer-pepito.app-1, groupId=pepito.app] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Node may not be available.
[36morg.apache.kafka.clients.NetworkClient  [0;39m [2m:[0;39m [Consumer clientId=consumer-pepito.app-1, groupId=pepito.app] Node 1 disconnected.
[2m2025-04-21T15:43:36.548-06:00[0;39m [33m WARN[0;39m [35m4088[0;39m [2m--- [pepito-app] [ntainer#0-0-C-1] [0;39m[36morg.apache.kafka.clients.NetworkClient  [0;39m [2m:[0;39m [Consumer clientId=consumer-pepito.app-1, groupId=pepito.app] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Node may not be available.
[36morg.apache.kafka.clients.NetworkClient  [0;39m [2m:[0;39m [Consumer clientId=consumer-pepito.app-1, groupId=pepito.app] Node 1 disconnected.
[2m2025-04-21T15:43:36.998-06:00[0;39m [33m WARN[0;39m [35m4088[0;39m [2m--- [pepito-app] [ntainer#0-0-C-1] [0;39m[36morg.apache.kafka.clients.NetworkClient  [0;39m [2m:[0;39m [Consumer clientId=consumer-pepito.app-1, groupId=pepito.app] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Node may not be available.
[36morg.apache.kafka.clients.NetworkClient  [0;39m [2m:[0;39m [Consumer clientId=consumer-pepito.app-1, groupId=pepito.app] Node 1 disconnected.
[2m2025-04-21T15:43:37.800-06:00[0;39m [33m WARN[0;39m [35m4088[0;39m [2m--- [pepito-app] [ntainer#0-0-C-1] [0;39m[36morg.apache.kafka.clients.NetworkClient  [0;39m [2m:[0;39m [Consumer clientId=consumer-pepito.app-1, groupId=pepito.app] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Node may not be available.

它试图连接到 127.0.0.1:9092，这是 springboot 的默认主机/端口，

Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Node may not be available.

并且我的 application.properties 中有

spring:
  kafka:
    bootstrap-servers: "pepito:8484" # in my hosts pepito is pointing to where docker is running

我的配置中还有什么地方需要指定只连接到 pepito:8484，而不能连接其他 IP/端口？另外，我没有修改 Kafka 容器镜像，需要在那里设置吗？欢迎任何帮助。

1 个回答

Voted

aran · Answer 1 · 2025-04-22T16:59:59+08:00

所以基本上，你的 Spring Boot 应用程序就像是，“嘿 Kafka，我正在敲pepito:8484，我们聊天吧！”

卡夫卡打开门说： “太棒了，很高兴见到你！不过如果你想继续聊天，以后可以通过localhost:9092联系我。”

而你的可怜的应用程序就像是，“呃......好吧，我想......”
然后它尝试调用localhost:9092，但显然，家里没有人......连接错误。

在您的service-kafka容器中，添加环境变量ADVERTISED_LISTENERS如下：

service-kafka:
  container_name: pepito-kafka-server
  hostname: pepito
  image: apache/kafka:3.7.1
  networks: 
    - pepito-network
  ports:
    - 8484:9092
  environment:
    KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
    KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://pepito:8484 #here
  extra_hosts:
    - "docker.internal:127.0.0.1"
  healthcheck:
    test: /opt/kafka/bin/kafka-cluster.sh cluster-id --bootstrap-server pepito:9092 || exit 1
    interval: 1s
    timeout: 60s
    retries: 60

这一次，kafka 会告诉应用程序“没关系，现在我们可以交谈了”。

Kafka 消费者总是尝试 127.0.0.1:9092

重新格式化数字，在固定位置插入分隔符

为什么 C++20 概念会导致循环约束错误，而老式的 SFINAE 不会？

VScode 自动卸载扩展的问题（Material 主题）

Vue 3：创建时出错“预期标识符但发现‘导入’”[重复]

具有指定基础类型但没有枚举器的“枚举类”的用途是什么？

如何修复未手动导入的模块的 MODULE_NOT_FOUND 错误？

`(表达式，左值) = 右值` 在 C 或 C++ 中是有效的赋值吗？为什么有些编译器会接受/拒绝它？

在 C++ 中，一个不执行任何操作的空程序需要 204KB 的堆，但在 C 中则不需要

PowerBI 目前与 BigQuery 不兼容：Simba 驱动程序与 Windows 更新有关

AdMob：MobileAds.initialize() - 对于某些设备，“java.lang.Integer 无法转换为 java.lang.String”

Kafka 消费者总是尝试 127.0.0.1:9092

1 个回答

相关问题