Kafka Producer运行一段时间后断开，报node -1 disconnected错误求助

问题根源 #

你的代码当前每次发送消息都新建KafkaProducer实例，虽能正常工作但资源开销极大；尝试复用实例时出现"node -1 disconnected"错误，本质是Producer元数据过期无法获取有效Broker节点，或是复用的实例未正确处理连接异常与元数据刷新逻辑。

核心解决方案 #

1. 实现KafkaProducer单例复用 #

KafkaProducer是线程安全的，官方明确推荐全局复用一个实例，而非每次发送新建。需在应用启动时完成Producer初始化，后续发送操作复用该实例。

修改代码示例：

// 全局单例Producer实例
private KafkaProducer<String, String> kafkaProducer;

// 应用初始化阶段调用（如类构造方法、Spring的@PostConstruct方法）
public void initProducer() {
    if (kafkaProducer == null) {
        kafkaProducer = new KafkaProducer<>(configProperties);
        // 注册JVM关闭钩子，确保应用退出时Producer优雅关闭，释放资源
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            if (kafkaProducer != null) {
                kafkaProducer.close(Duration.ofSeconds(10));
            }
        }));
    }
}

public void sendMessage(String topic, String message) {
    L.debug(String.format("Sending %s: %s", topic, message));
    ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, message);
    // 复用已初始化的Producer实例
    kafkaProducer.send(producerRecord, (metadata, exception) -> {
        if (exception != null) {
            L.error("消息发送失败", exception);
            // 针对连接异常触发Producer重建逻辑
            handleProducerConnectionException(exception);
        } else {
            L.debug(String.format("消息发送成功，offset: %d", metadata.offset()));
        }
    });
    kafkaProducer.flush();
}

// 异常处理与Producer重建逻辑
private void handleProducerConnectionException(Exception exception) {
    // 判断是否为节点连接类异常
    if (exception instanceof KafkaException && exception.getMessage().contains("node -1 disconnected")) {
        // 关闭旧Producer实例
        if (kafkaProducer != null) {
            try {
                kafkaProducer.close(Duration.ofSeconds(5));
            } catch (Exception e) {
                L.error("关闭旧Producer实例失败", e);
            }
        }
        // 重建Producer实例
        kafkaProducer = new KafkaProducer<>(configProperties);
        L.info("已重建KafkaProducer实例");
    }
}

2. 调整关键配置参数 #

在configProperties中添加或修改以下参数，帮助Producer自动处理连接与元数据刷新：

• metadata.max.age.ms: 缩短元数据过期时间，默认300000ms（5分钟），建议设置为30000（30秒），让Producer主动刷新Broker集群拓扑
• connections.max.idle.ms: 空闲连接超时时间，默认540000ms（9分钟），可设置为180000（3分钟），及时释放无用连接
• reconnect.backoff.ms/reconnect.backoff.max.ms: 设置重连退避时间，默认50ms/1000ms，可调整为100ms/2000ms，确保断开后自动重试连接
• acks: 建议设置为1或all，确保消息发送确认，避免因无确认导致的元数据不同步

示例配置代码：

Properties configProperties = new Properties();
configProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "your-broker-address-list");
configProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
configProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
// 新增优化配置
configProperties.put(ProducerConfig.METADATA_MAX_AGE_MS_CONFIG, 30000);
configProperties.put(ProducerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG, 180000);
configProperties.put(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG, 100);
configProperties.put(ProducerConfig.RECONNECT_BACKOFF_MAX_MS_CONFIG, 2000);
configProperties.put(ProducerConfig.ACKS_CONFIG, "1");

3. 禁止频繁创建/销毁Producer #

每次新建KafkaProducer会建立新的TCP连接、线程池与元数据缓存，既浪费客户端资源，也会导致Broker端连接数过载。复用单例是官方推荐的最优实践。

关于"node -1 disconnected"的说明 #

该错误表示Producer的本地元数据中无可用Broker节点，常见原因：

• Producer长时间未刷新元数据，Broker集群拓扑变化（节点下线、地址变更）后无法感知
• 网络波动导致连接断开，且未触发自动重连
• 初始化时bootstrap.servers配置错误，或Broker拒绝连接

通过上述单例复用+配置调整+异常处理的方案，即可解决该问题。

内容的提问来源于stack exchange，提问作者Deepak Kaushik