Spring应用集成Kafka Producer生产消息Broker断连无法自动重连如何解决

问题根因 #

你当前配置未显式声明Kafka生产者的重连、元数据刷新相关参数，默认参数无法达到预期的5分钟自动重连效果，需补充以下配置：

需新增的核心配置项 #

• retries：设置消息发送失败后的重试次数，建议设置为≥10，若需保证消息严格顺序，需同步配置max.in.flight.requests.per.connection=1避免消息乱序
• retry.backoff.ms：两次重试之间的等待间隔，默认100ms，可根据业务需求调整，例如设置为1000（1秒）
• connections.max.idle.ms：连接最大空闲时间，默认值为540000（9分钟），超过该时长的空闲连接会被主动关闭，下次发送消息时会自动重建连接，若期望5分钟触发重连可设置为300000
• metadata.max.age.ms：集群元数据最大有效期，默认值为300000（5分钟），到期后生产者会主动拉取最新的集群节点信息，感知宕机Broker的恢复状态，和上述参数配合即可实现5分钟维度的自动重连
• delivery.timeout.ms：消息发送的总超时时间，需大于 重试间隔*重试次数+消息攒批等待时间 的总和，避免重试未完成就提前抛出超时异常，可设置为120000（2分钟）

调整后的完整配置代码 #

@Bean
public KafkaProducer<String, String> buildProducer() {
    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServerUrl);
    props.put(ProducerConfig.CLIENT_ID_CONFIG, clientId);
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    // 新增重连相关配置
    props.put(ProducerConfig.RETRIES_CONFIG, 10);
    props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 1000);
    props.put(ProducerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG, 300000);
    props.put(ProducerConfig.METADATA_MAX_AGE_CONFIG, 300000);
    props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120000);
    // 按需开启严格顺序保证
    // props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1);
    
    return new KafkaProducer<>(props);
}

补充说明 #

KafkaProducer本身是线程安全的，你当前使用单例Bean注入的方式没有问题，无需调整实例化逻辑。如果是多Broker集群，单节点宕机后生产者会自动将流量切到可用Broker，待宕机节点恢复后，通过定时的元数据刷新即可自动重建连接。

内容的提问来源于stack exchange，提问作者jagannathan rajagopalan