单个应用接入多个Kafka集群的优雅实现方案咨询

核心实现思路 #

完全基于Spring生态原生能力实现，不需要引入额外第三方组件，可实现一次配置全量生效的需求。 #

步骤1：定义结构化批量配置 #

通过@ConfigurationProperties绑定多集群嵌套配置，避免重复填写参数。
配置文件示例（yaml）：

kafka:
  clusters:
    # 集群1配置
    - cluster-id: cluster-pay
      bootstrap-servers: 10.0.0.1:9092,10.0.0.2:9092
      topics:
        - name: pay_order_notify
          serialization-type: avro_confluent
          schema-registry-url: http://10.0.0.3:8081
        - name: pay_refund_record
          serialization-type: json
    # 集群2配置
    - cluster-id: cluster-log
      bootstrap-servers: 10.0.1.1:9092,10.0.1.2:9092
      topics:
        - name: log_operation
          serialization-type: avro_single
          avro-schema-path: classpath:avro/operation_log.avsc

对应的配置绑定类示例：

@ConfigurationProperties(prefix = "kafka")
public class MultiKafkaProperties {
    private List<KafkaCluster> clusters;
    // 省略getter/setter、内部类定义
}

步骤2：动态批量注册所有Kafka资源 #

实现BeanDefinitionRegistryPostProcessor，遍历配置中的集群和Topic列表，自动注册对应实例：

• 每个集群对应1个独立的KafkaAdmin，自动创建该集群下配置的所有Topic
• 每个Topic对应1个专属的ProducerFactory和KafkaTemplate，根据配置的序列化类型动态生成对应序列化器实例，自动注入Schema Registry地址或本地Avro Schema
• 所有生成的KafkaTemplate以{topicName}KafkaTemplate为Bean名注册到Spring上下文

步骤3：封装统一调用入口 #

实现通用的消息发送模板，自动匹配对应Topic的KafkaTemplate：

@Component
public class MultiClusterKafkaTemplate {
    private final Map<String, KafkaTemplate<?, ?>> topicTemplateMap;

    // 构造方法自动注入所有Topic对应的KafkaTemplate
    public MultiClusterKafkaTemplate(Map<String, KafkaTemplate<?, ?>> allTemplates) {
        this.topicTemplateMap = allTemplates.entrySet().stream()
                .filter(entry -> entry.getKey().endsWith("KafkaTemplate"))
                .collect(Collectors.toMap(
                    entry -> entry.getKey().replace("KafkaTemplate", ""),
                    Map.Entry::getValue
                ));
    }

    @SuppressWarnings("unchecked")
    public <K, V> CompletableFuture<SendResult<K, V>> send(String topic, V data) {
        KafkaTemplate<K, V> template = (KafkaTemplate<K, V>) topicTemplateMap.get(topic);
        if (template == null) {
            throw new IllegalArgumentException("未找到Topic[" + topic + "]对应的Kafka配置");
        }
        return template.send(topic, data);
    }
}

使用方式 #

业务代码无需感知底层集群、序列化配置，直接调用统一模板发送即可：

// 发送Avro消息到pay_order_notify Topic，自动使用Confluent Schema Registry序列化
multiClusterKafkaTemplate.send("pay_order_notify", new OrderNotifyVO(...));
// 发送JSON消息到pay_refund_record Topic
multiClusterKafkaTemplate.send("pay_refund_record", new RefundRecordVO(...));

方案优势 #

• 所有配置仅需填写1次，无冗余重复配置
• 完全兼容两种Avro序列化方式、JSON序列化，可按需扩展其他序列化策略
• 资源自动创建，无需手动声明大量Bean
• 业务侧无感知，使用成本极低

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