Spring应用能否通过单个bootstrap.servers连接两个Kafka集群？

不能用单个bootstrap.servers配置同时对接两个不同的Kafka集群。

原因解释 #

Kafka的bootstrap.servers参数作用是指定单个集群的初始连接节点，客户端会通过这些节点获取整个集群的元数据（比如所有broker地址、topic分区信息等）。如果把两个独立集群的broker地址混填到同一个bootstrap.servers里，客户端会错误地将它们视为同一个集群的节点，导致元数据解析混乱，最终出现连接失败、消息发送/消费异常等问题。

Spring应用中的正确配置方式 #

需要为每个集群单独配置一套独立的Kafka客户端参数，核心是使用两个不同的bootstrap.servers值。以Spring Boot为例，具体实现步骤如下：

1. 在配置文件中定义两套集群参数
   比如在application.yml中：

   # 集群1配置
   kafka-cluster1:
     bootstrap-servers: cluster1-broker1:9092,cluster1-broker2:9092
     key-serializer: org.apache.kafka.common.serialization.StringSerializer
     value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
   
   # 集群2配置
   kafka-cluster2:
     bootstrap-servers: cluster2-broker1:9092,cluster2-broker2:9092
     key-serializer: org.apache.kafka.common.serialization.StringSerializer
     value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
   

2. 编写配置类创建独立的客户端实例
   通过@ConfigurationProperties绑定不同的配置，生成各自的ProducerFactory和KafkaTemplate：

   @Configuration
   public class MultiClusterKafkaConfig {
   
       // 集群1的生产者配置
       @Bean
       @ConfigurationProperties("kafka-cluster1")
       public KafkaProperties cluster1KafkaProperties() {
           return new KafkaProperties();
       }
   
       @Bean
       public ProducerFactory<String, Object> cluster1ProducerFactory() {
           return cluster1KafkaProperties().buildProducerFactory();
       }
   
       @Bean
       public KafkaTemplate<String, Object> cluster1KafkaTemplate() {
           return new KafkaTemplate<>(cluster1ProducerFactory());
       }
   
       // 集群2的生产者配置
       @Bean
       @ConfigurationProperties("kafka-cluster2")
       public KafkaProperties cluster2KafkaProperties() {
           return new KafkaProperties();
       }
   
       @Bean
       public ProducerFactory<String, Object> cluster2ProducerFactory() {
           return cluster2KafkaProperties().buildProducerFactory();
       }
   
       @Bean
       public KafkaTemplate<String, Object> cluster2KafkaTemplate() {
           return new KafkaTemplate<>(cluster2ProducerFactory());
       }
   }
   

3. 业务代码中使用对应实例
   注入指定的KafkaTemplate来操作对应集群：

   @Service
   public class KafkaMessageService {
   
       private final KafkaTemplate<String, Object> cluster1KafkaTemplate;
       private final KafkaTemplate<String, Object> cluster2KafkaTemplate;
   
       public KafkaMessageService(KafkaTemplate<String, Object> cluster1KafkaTemplate,
                                 KafkaTemplate<String, Object> cluster2KafkaTemplate) {
           this.cluster1KafkaTemplate = cluster1KafkaTemplate;
           this.cluster2KafkaTemplate = cluster2KafkaTemplate;
       }
   
       public void sendToCluster1(String topic, Object message) {
           cluster1KafkaTemplate.send(topic, message);
       }
   
       public void sendToCluster2(String topic, Object message) {
           cluster2KafkaTemplate.send(topic, message);
       }
   }
   

如果是消费者场景，同样需要为每个集群创建独立的ConsumerFactory和ConcurrentKafkaListenerContainerFactory，并在@KafkaListener注解中通过containerFactory属性指定使用的容器工厂。

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