springboot多kafka配置

第一步：引入maven依赖

    org.springframework.kafka
    spring-kafka

第二步：新增配置文件

以下为大致结构，供参考

spring:
  kafka:
    # 第一个kafka的配置
    first:
      bootstrap-servers: xxx.xxx.xxx.xxx:xxxx
      producer:
        retries: x
        acks: -1
      consumer:
        enable-auto-commit: false
        group-id: first-consumer
      listener:
      	ack-mode: xx
    # 第二个kafka的配置
    second:
      bootstrap-servers: xxx.xxx.xxx.xxx:xxxx
      producer:
        batch-size: xxxx
        buffer-memory: xxxxxx
      consumer:
        auto-offset-reset: earliest
        group-id: second-consumer
      listener:
      	concurrency: xx

第三步：新增配置类

第一个kafka的配置类

/**
 * 第一个kafka配置
 *
 * @author 小流慢影
 * @date 2023年5月15日
 */
@Configuration
public class FirstKafkaConfig {

    /**
     * 读取第一个kafka配置
     * Primary注解表示默认以这个为准
     *
     * @return 第一个kafka配置
     */
    @Primary
    @ConfigurationProperties(prefix = "spring.kafka.first")
    @Bean
    public KafkaProperties firstKafkaProperties() {
        return new KafkaProperties();
    }

    /**
     * 构建第一个kafka的生产者发送template
     *
     * @param firstKafkaProperties 第一个kafka配置
     * @return 第一个kafka的生产者发送template
     */
    @Primary
    @Bean
    public KafkaTemplate firstKafkaTemplate(
            @Autowired @Qualifier("firstKafkaProperties") KafkaProperties firstKafkaProperties) {
        return new KafkaTemplate(firstProducerFactory(firstKafkaProperties));
    }

    /**
     * 构建第一个kafka的消费者监听容器工厂
     *
     * @param firstKafkaProperties 第一个kafka配置
     * @return 第一个kafka的消费者监听容器工厂
     */
    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer>
    firstKafkaListenerContainerFactory(@Autowired @Qualifier("firstKafkaProperties") KafkaProperties firstKafkaProperties) {
        ConcurrentKafkaListenerContainerFactory factory =
                new ConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(firstConsumerFactory(firstKafkaProperties));
        return factory;
    }

    /**
     * 新建第一个kafka的消费者工厂
     *
     * @param firstKafkaProperties 第一个kafka配置
     * @return 第一个kafka的消费者工厂
     */
    private ConsumerFactory firstConsumerFactory(KafkaProperties firstKafkaProperties) {
        return new DefaultKafkaConsumerFactory(firstKafkaProperties.buildConsumerProperties());
    }

    /**
     * 新建第一个kafka的生产者工厂
     *
     * @param firstKafkaProperties 第一个kafka配置
     * @return 第一个kafka的生产者工厂
     */
    private DefaultKafkaProducerFactory firstProducerFactory(KafkaProperties firstKafkaProperties) {
        return new DefaultKafkaProducerFactory(firstKafkaProperties.buildProducerProperties());
    }
}

第二个kafka的配置类

/**
 * 第二个kafka配置
 *
 * @author 小流慢影
 * @date 2023年5月15日
 */
@Configuration
public class SecondKafkaConfig {

    /**
     * 读取第二个kafka配置
     *
     * @return 第二个kafka配置
     */
    @ConfigurationProperties(prefix = "spring.kafka.second")
    @Bean
    public KafkaProperties secondKafkaProperties() {
        return new KafkaProperties();
    }

    /**
     * 构建第二个kafka的生产者发送template
     *
     * @param secondKafkaProperties 第二个kafka配置
     * @return 第二个kafka的生产者发送template
     */
    @Bean
    public KafkaTemplate secondKafkaTemplate(
            @Autowired @Qualifier("secondKafkaProperties") KafkaProperties secondKafkaProperties) {
        return new KafkaTemplate(secondProducerFactory(secondKafkaProperties));
    }

    /**
     * 构建第二个kafka的消费者监听容器工厂
     *
     * @param secondKafkaProperties 第二个kafka配置
     * @return 第二个kafka的消费者监听容器工厂
     */
    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer>
    secondKafkaListenerContainerFactory(@Autowired @Qualifier("secondKafkaProperties") KafkaProperties secondKafkaProperties) {
        ConcurrentKafkaListenerContainerFactory factory =
                new ConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(secondConsumerFactory(secondKafkaProperties));
        return factory;
    }

    /**
     * 新建第二个kafka的消费者工厂
     *
     * @param secondKafkaProperties 第二个kafka配置
     * @return 第二个kafka的消费者工厂
     */
    private ConsumerFactory secondConsumerFactory(KafkaProperties secondKafkaProperties) {
        return new DefaultKafkaConsumerFactory(secondKafkaProperties.buildConsumerProperties());
    }

    /**
     * 新建第二个kafka的生产者工厂
     *
     * @param secondKafkaProperties 第二个kafka配置
     * @return 第二个kafka的生产者工厂
     */
    private DefaultKafkaProducerFactory secondProducerFactory(KafkaProperties secondKafkaProperties) {
        return new DefaultKafkaProducerFactory(secondKafkaProperties.buildProducerProperties());
    }
}

第四步：用

生产者用法

/**
 * 第一个kafka配置默认就可以用，因为配置了@Primary
 */
@Resource
private KafkaTemplate firstKafkaTemplate;

/**
 * 第二个kafka配置需要指定下名字
 */
@Resource(name = "secondKafkaTemplate")
private KafkaTemplate secondKafkaTemplate;

消费者用法

/**
 * 测试消费者
 * 这里要消费哪一个kafka消息，containerFactory就需要配成上面相对应的消费者监听容器工厂
 * @param record 消息
 * @param ack ack
 */
@KafkaListener(
        containerFactory = "secondKafkaListenerContainerFactory",
        topics = {"xxxx"},
        groupId = "second-consumer")
public void testConsumer(ConsumerRecord record, Acknowledgment ack) {
    //do something
}