如何通过application.yml文件配置KafkaRoutingTemplate并实现双Kafka生产者的无编码配置？

Spring Kafka 完全支持通过 application.yml 配置多个生产者实例，无需硬编码配置类就能实现你的需求。下面是具体的配置和使用步骤：

1. 在application.yml中配置两个Kafka集群的生产者参数 #

你可以给每个集群的配置加上自定义前缀，比如分别用 spring.kafka.azure-confluent 和 spring.kafka.azure-vm 来区分：

spring:
  kafka:
    # Azure Confluent 集群的生产者配置
    azure-confluent:
      bootstrap-servers: "your-confluent-cluster-bootstrap-servers"
      producer:
        key-serializer: org.apache.kafka.common.serialization.StringSerializer
        value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
        # Confluent 通常需要SASL认证，这里根据实际情况配置
        properties:
          sasl.mechanism: PLAIN
          security.protocol: SASL_SSL
          sasl.jaas.config: >
            org.apache.kafka.common.security.plain.PlainLoginModule required
            username="your-confluent-api-key"
            password="your-confluent-api-secret";
    # Azure VM 上的Kafka集群生产者配置
    azure-vm:
      bootstrap-servers: "your-vm-kafka-bootstrap-servers"
      producer:
        key-serializer: org.apache.kafka.common.serialization.StringSerializer
        value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
        # 如果VM集群不需要认证，这里可以省略security相关配置

2. 通过@ConfigurationProperties绑定配置，生成生产者工厂和KafkaTemplate #

这一步只需要少量的代码来绑定yml配置，不需要硬编码生产者参数：

@Configuration
public class MultiKafkaConfig {

    // 绑定Azure Confluent的配置
    @Bean
    @ConfigurationProperties(prefix = "spring.kafka.azure-confluent")
    public KafkaProperties confluentKafkaProperties() {
        return new KafkaProperties();
    }

    @Bean
    public ProducerFactory<String, Object> confluentProducerFactory() {
        return confluentKafkaProperties().buildProducerFactory();
    }

    @Bean("confluentKafkaTemplate")
    public KafkaTemplate<String, Object> confluentKafkaTemplate() {
        return new KafkaTemplate<>(confluentProducerFactory());
    }

    // 绑定Azure VM的配置
    @Bean
    @ConfigurationProperties(prefix = "spring.kafka.azure-vm")
    public KafkaProperties vmKafkaProperties() {
        return new KafkaProperties();
    }

    @Bean
    public ProducerFactory<String, Object> vmProducerFactory() {
        return vmKafkaProperties().buildProducerFactory();
    }

    @Bean("vmKafkaTemplate")
    public KafkaTemplate<String, Object> vmKafkaTemplate() {
        return new KafkaTemplate<>(vmProducerFactory());
    }
}

3. 使用KafkaRoutingTemplate实现路由发送 #

接下来你可以根据Topic或者自定义规则，用KafkaRoutingTemplate将消息发送到对应的集群：

@Component
public class KafkaMessageDispatcher {

    private final KafkaRoutingTemplate routingTemplate;

    // 注入两个不同的KafkaTemplate
    public KafkaMessageDispatcher(@Qualifier("confluentKafkaTemplate") KafkaTemplate<String, Object> confluentTemplate,
                                  @Qualifier("vmKafkaTemplate") KafkaTemplate<String, Object> vmTemplate) {
        // 定义路由映射：key是Topic名称，value是对应的KafkaTemplate
        Map<String, KafkaTemplate<String, Object>> templateMap = new HashMap<>();
        templateMap.put("your-confluent-topic-name", confluentTemplate);
        templateMap.put("your-vm-kafka-topic-name", vmTemplate);

        this.routingTemplate = new KafkaRoutingTemplate(templateMap);
    }

    // 发送消息时，只需指定Topic，路由模板会自动选择对应的生产者
    public void sendToTargetTopic(String topic, Object message) {
        routingTemplate.send(topic, message);
    }
}

关键优势 #

• 配置与代码分离：所有集群连接信息、生产者参数都集中在application.yml中，后续调整无需修改代码。
• 扩展性强：如果后续需要新增更多Kafka集群，只需在yml中添加新的配置块，再补充对应的绑定Bean即可。

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