在Spring Boot运行时动态创建N个KafkaTemplate实例的教程

import lombok.Getter;
import lombok.Setter;
import java.util.List;

@Getter
@Setter
public class KafkaCluster {
  private String beanName; // KafkaTemplate的Bean名称
  private List<String> bootstrapServers; // Kafka集群的引导服务器地址
}

kafka.clusters[0].bean-name=cluster1KafkaTemplate
kafka.clusters[0].bootstrap-servers=localhost:9092,localhost:9093

kafka.clusters[1].bean-name=cluster2KafkaTemplate
kafka.clusters[1].bootstrap-servers=anotherhost:9092,anotherhost:9093

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor;
import org.springframework.beans.factory.support.GenericBeanDefinition;
import org.springframework.boot.context.properties.bind.Bindable;
import org.springframework.boot.context.properties.bind.Binder;
import org.springframework.core.env.Environment;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

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

public class KafkaTemplateDefinitionRegistrar implements BeanDefinitionRegistryPostProcessor {

  private final List<KafkaCluster> clusters;

  // 构造器：使用Binder API绑定环境属性
  public KafkaTemplateDefinitionRegistrar(Environment environment) {
    this.clusters = Binder.get(environment)
        .bind("kafka.clusters", Bindable.listOf(KafkaCluster.class))
        .orElseThrow(() -> new IllegalStateException("Kafka集群配置未找到或绑定失败！"));
  }

  @Override
  public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) throws BeansException {
    clusters.forEach(cluster -> {
      GenericBeanDefinition beanDefinition = new GenericBeanDefinition();
      beanDefinition.setBeanClass(KafkaTemplate.class);
      // 使用InstanceSupplier延迟创建KafkaTemplate实例
      beanDefinition.setInstanceSupplier(() -> kafkaTemplate(cluster));
      registry.registerBeanDefinition(cluster.getBeanName(), beanDefinition);
      System.out.println("动态注册KafkaTemplate Bean: " + cluster.getBeanName());
    });
  }

  @Override
  public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
    // 此方法在此场景下不需要特殊处理
  }

  // 根据KafkaCluster配置创建ProducerFactory
  public ProducerFactory<String, String> producerFactory(KafkaCluster kafkaCluster) {
    Map<String, Object> configProps = new HashMap<>();
    configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaCluster.getBootstrapServers());
    configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    // 其他生产者配置可在此处添加
    return new DefaultKafkaProducerFactory<>(configProps);
  }

  // 根据KafkaCluster配置创建KafkaTemplate
  public KafkaTemplate<String, String> kafkaTemplate(KafkaCluster kafkaCluster) {
    return new KafkaTemplate<>(producerFactory(kafkaCluster));
  }
}

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;

@Configuration
public class KafkaTemplateDefinitionRegistrarConfiguration {

  @Bean
  // 注意：BeanDefinitionRegistryPostProcessor必须声明为static，以确保它在Bean定义处理阶段被优先实例化
  public static KafkaTemplateDefinitionRegistrar beanDefinitionRegistrar(Environment environment) {
    return new KafkaTemplateDefinitionRegistrar(environment);
  }
}

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;

@SpringBootApplication(exclude={KafkaAutoConfiguration.class})
public class YourApplication {
    public static void main(String[] args) {
        SpringApplication.run(YourApplication.class, args);
    }
}

import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.List;

@SpringBootTest
class DynamicKafkaTemplateTest {

    // 自动注入所有KafkaTemplate实例
    @Autowired
    private List<KafkaTemplate<String,String>> kafkaTemplates;

    @Autowired
    private KafkaTemplate<String, String> cluster1KafkaTemplate; // 通过名称注入特定Bean

    @Autowired
    private KafkaTemplate<String, String> cluster2KafkaTemplate; // 通过名称注入特定Bean

    @Test
    void kafkaTemplatesSizeTest() {
        // 验证KafkaTemplate的数量是否与配置中定义的集群数量一致
        Assertions.assertEquals(2, kafkaTemplates.size(), "动态创建的KafkaTemplate数量不正确");
    }

    @Test
    void specificKafkaTemplateInjectionTest() {
        // 验证特定名称的KafkaTemplate是否可以成功注入
        Assertions.assertNotNull(cluster1KafkaTemplate, "cluster1KafkaTemplate未能成功注入");
        Assertions.assertNotNull(cluster2KafkaTemplate, "cluster2KafkaTemplate未能成功注入");

        // 可以在这里进一步测试KafkaTemplate的功能，例如发送消息
        // cluster1KafkaTemplate.send("test-topic-1", "Hello from cluster 1");
        // cluster2KafkaTemplate.send("test-topic-2", "Hello from cluster 2");
    }
}