如何关闭现有Kafka Producer及Spring Kafka各版本对应处理方案

适用场景 #

• 已配置setBootstrapServersSupplier()实现运行时动态调整Kafka集群地址，需要立刻终止旧连接、让新的生产请求使用新集群地址时使用
• Kafka集群侧地址发生变更，需要主动销毁所有已缓存的旧生产者实例、释放旧连接资源时使用
• 需要批量更新生产者全局配置，要求后续所有新生产请求使用新配置生成的生产者实例时使用

正确调用方式 #

注意：必须注入DefaultKafkaProducerFactory实现类，不能注入ProducerFactory接口，接口层未定义reset()方法
调用代码示例：

// 注入实现类
@Autowired
private DefaultKafkaProducerFactory<Object, Object> producerFactory;

public void switchKafkaCluster(List<String> newBootstrapServers) {
    // 先更新地址获取规则
    producerFactory.setBootstrapServersSupplier(() -> newBootstrapServers);
    // 执行reset销毁所有缓存的旧生产者实例
    producerFactory.reset();
}

调用注意事项 #

• reset()仅销毁空闲状态的生产者实例，正在处理消息发送逻辑的实例会等请求完成后再销毁，不会丢失正在处理的消息
• reset()执行完成后，下一次调用KafkaTemplate.send()时会自动调用supplier获取新地址，创建新的生产者实例处理请求

2.2.x版本未引入KafkaResourceFactory相关能力，没有内置的setBootstrapServersSupplier和reset()方法，可通过自定义代理类实现相同需求：

方案：封装动态代理ProducerFactory #

自己实现ProducerFactory接口，内部持有真实的DefaultKafkaProducerFactory实例，需要切换地址时销毁旧实例、创建新实例替换即可，代码示例：

@Component
public class DynamicKafkaProducerFactory implements ProducerFactory<Object, Object> {
    private volatile DefaultKafkaProducerFactory<Object, Object> delegate;

    // 初始化时加载默认配置
    public DynamicKafkaProducerFactory(@Value("${spring.kafka.bootstrap-servers}") String defaultBootstrap) {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, defaultBootstrap);
        // 其他生产者配置（序列化、重试、acks等）按需补充
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        this.delegate = new DefaultKafkaProducerFactory<>(configs);
    }

    // 切换集群地址时调用该方法
    public void updateBootstrapServers(List<String> newBootstrapServers) {
        // 复用原有其他配置，仅替换bootstrap地址
        Map<String, Object> newConfigs = new HashMap<>(delegate.getConfigurationProperties());
        newConfigs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, newBootstrapServers);
        // 创建新的ProducerFactory实例
        DefaultKafkaProducerFactory<Object, Object> newFactory = new DefaultKafkaProducerFactory<>(newConfigs);
        // 销毁旧实例释放连接资源
        this.delegate.destroy();
        // 替换为新实例
        this.delegate = newFactory;
    }

    // 代理所有ProducerFactory接口方法，全部调用内部delegate实例的对应方法
    @Override
    public Producer<Object, Object> createProducer() {
        return delegate.createProducer();
    }

    @Override
    public boolean isTransactional() {
        return delegate.isTransactional();
    }

    // 其余接口方法按上述规则补充实现即可
}

如果是消费者需要动态改地址，同样可以参考该逻辑封装KafkaListenerContainerFactory的代理类，需要切换地址时先停止旧的监听容器，用新配置创建新的容器再启动即可。如果是固定多集群切换场景，也可以提前初始化多个KafkaTemplate实例，动态路由到对应集群的Template发送消息，切换开销更低。

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