Spring Boot应用配置多Kafka消费者集群（各集群SSL配置独立）

完全可以通过application.yml/properties配置实现，Spring Kafka和Spring Cloud Stream都支持多集群独立SSL配置。你之前配置失效是因为混用了两套框架的配置体系，下面分别给出两种方案的实现方式：

方案一：Spring Kafka 多集群消费 #

Spring Kafka通过多消费者工厂+监听器容器工厂实现多集群连接，每个集群的SSL配置完全独立。

1. application.yml 配置 #

# 第一个Kafka集群配置
kafka-cluster-1:
  bootstrap-servers: cluster1-kafka:9093
  consumer:
    group-id: group-cluster1
    auto-offset-reset: earliest
    key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  ssl:
    trust-store-location: classpath:cluster1-truststore.jks
    trust-store-password: cluster1-trust-pass
    key-store-location: classpath:cluster1-keystore.jks
    key-store-password: cluster1-key-pass
    key-password: cluster1-key-pass

# 第二个Kafka集群配置
kafka-cluster-2:
  bootstrap-servers: cluster2-kafka:9093
  consumer:
    group-id: group-cluster2
    auto-offset-reset: earliest
    key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  ssl:
    trust-store-location: classpath:cluster2-truststore.jks
    trust-store-password: cluster2-trust-pass
    key-store-location: classpath:cluster2-keystore.jks
    key-store-password: cluster2-key-pass
    key-password: cluster2-key-pass

2. 配置类实现 #

创建两个独立的消费者工厂和监听器容器工厂，绑定各自的集群配置：

import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class MultiKafkaClusterConfig {

    // 集群1配置属性
    @Bean
    @ConfigurationProperties(prefix = "kafka-cluster-1")
    public Map<String, Object> cluster1KafkaProps() {
        return new HashMap<>();
    }

    // 集群1消费者工厂
    @Bean("cluster1ConsumerFactory")
    public ConsumerFactory<String, String> cluster1ConsumerFactory() {
        return new DefaultKafkaConsumerFactory<>(cluster1KafkaProps());
    }

    // 集群1监听器容器工厂
    @Bean("cluster1ListenerFactory")
    public ConcurrentKafkaListenerContainerFactory<String, String> cluster1ListenerFactory(
            @Qualifier("cluster1ConsumerFactory") ConsumerFactory<String, String> factory) {
        ConcurrentKafkaListenerContainerFactory<String, String> containerFactory = new ConcurrentKafkaListenerContainerFactory<>();
        containerFactory.setConsumerFactory(factory);
        return containerFactory;
    }

    // 集群2配置属性
    @Bean
    @ConfigurationProperties(prefix = "kafka-cluster-2")
    public Map<String, Object> cluster2KafkaProps() {
        return new HashMap<>();
    }

    // 集群2消费者工厂
    @Bean("cluster2ConsumerFactory")
    public ConsumerFactory<String, String> cluster2ConsumerFactory() {
        return new DefaultKafkaConsumerFactory<>(cluster2KafkaProps());
    }

    // 集群2监听器容器工厂
    @Bean("cluster2ListenerFactory")
    public ConcurrentKafkaListenerContainerFactory<String, String> cluster2ListenerFactory(
            @Qualifier("cluster2ConsumerFactory") ConsumerFactory<String, String> factory) {
        ConcurrentKafkaListenerContainerFactory<String, String> containerFactory = new ConcurrentKafkaListenerContainerFactory<>();
        containerFactory.setConsumerFactory(factory);
        return containerFactory;
    }
}

3. 消费者代码 #

在@KafkaListener中指定对应集群的监听器工厂：

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class MultiClusterConsumer {

    @KafkaListener(topics = "topic-cluster1", containerFactory = "cluster1ListenerFactory")
    public void consumeCluster1(String message) {
        System.out.println("Cluster1消息: " + message);
    }

    @KafkaListener(topics = "topic-cluster2", containerFactory = "cluster2ListenerFactory")
    public void consumeCluster2(String message) {
        System.out.println("Cluster2消息: " + message);
    }
}

方案二：Spring Cloud Stream 多集群消费 #

通过多绑定器配置实现，无需额外编写配置类，直接在yml中定义不同集群的绑定器即可。

1. application.yml 配置 #

spring:
  cloud:
    stream:
      # 定义两个绑定器，对应两个Kafka集群
      binders:
        kafka-cluster1:
          type: kafka
          environment:
            spring:
              kafka:
                bootstrap-servers: cluster1-kafka:9093
                consumer:
                  group-id: group-cluster1
                  auto-offset-reset: earliest
                ssl:
                  trust-store-location: classpath:cluster1-truststore.jks
                  trust-store-password: cluster1-trust-pass
                  key-store-location: classpath:cluster1-keystore.jks
                  key-store-password: cluster1-key-pass
                  key-password: cluster1-key-pass
        kafka-cluster2:
          type: kafka
          environment:
            spring:
              kafka:
                bootstrap-servers: cluster2-kafka:9093
                consumer:
                  group-id: group-cluster2
                  auto-offset-reset: earliest
                ssl:
                  trust-store-location: classpath:cluster2-truststore.jks
                  trust-store-password: cluster2-trust-pass
                  key-store-location: classpath:cluster2-keystore.jks
                  key-store-password: cluster2-key-pass
                  key-password: cluster2-key-pass
      # 绑定消费者到对应绑定器
      bindings:
        cluster1-in:
          destination: topic-cluster1
          binder: kafka-cluster1
          group: group-cluster1
        cluster2-in:
          destination: topic-cluster2
          binder: kafka-cluster2
          group: group-cluster2

2. 消费者代码 #

通过@StreamListener绑定对应集群的输入通道：

import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

@Component
public class StreamMultiClusterConsumer {

    @StreamListener("cluster1-in")
    public void consumeCluster1(@Payload String message) {
        System.out.println("Stream-Cluster1消息: " + message);
    }

    @StreamListener("cluster2-in")
    public void consumeCluster2(@Payload String message) {
        System.out.println("Stream-Cluster2消息: " + message);
    }
}

你之前配置失效的原因 #

你混用了Spring Kafka和Spring Cloud Stream的配置体系：

• Spring Kafka的spring.kafka.consumer配置仅作用于自身的消费者容器，不会影响Spring Cloud Stream的绑定器
• Spring Cloud Stream的消费者必须绑定到指定的自定义绑定器，单独的spring.cloud.stream.bindings.consumer配置会默认使用未配置SSL的默认绑定器，因此第二个集群配置不生效

建议统一使用其中一套框架的方案，不要混用。

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