SpringBoot2.1.9 多Kafka消费者配置

ゝ一世哀愁。 2023-06-30 15:51 8阅读 0赞

### 一、配置文件 ###

pom.xml

<dependency>
    	<groupId>org.springframework.kafka</groupId>
    	<artifactId>spring-kafka</artifactId>
    </dependency>

application.yml

spring:
      application:
        name: double-kafka-consumer
      profiles:
        active: dev
    
      jackson:
        date-format: yyyy-MM-dd HH:mm:ss
        time-zone: GMT+8
    
    server:
      port: 8008
    
    sys:
      kafka:
        one:
          bootstrap-servers: 192.168.1.2:5021
          consumer:
            group-id: one-group
            auto-offset-reset: latest
            enable-auto-commit: false
            key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
            value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
            max-poll-records: 5
          topic: one-kafka-test
        two:
          bootstrap-servers: 192.168.1.3:5021
          consumer:
            group-id: two-group
            auto-offset-reset: latest
            enable-auto-commit: false
            key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
            value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
            max-poll-records: 5
          topic: two-kafka-test

### 二、配置Configuration Bean ###

（1）第一个kafka配置

public class OneKafkaConfig {
    
        @Bean
        public KafkaListenerContainerFactory oneKafkaFactory(@Autowired @Qualifier("oneConsumerFactory") ConsumerFactory oneConsumerFactory) {
            ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
            factory.setConsumerFactory(oneConsumerFactory);
            factory.setConcurrency(10);
            factory.setBatchListener(true);
            factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);//手动提交
    
            return factory;
        }
    
        @Bean
        public ConsumerFactory oneConsumerFactory(@Autowired @Qualifier("oneKafkaProperties") KafkaProperties oneKafkaProperties){
            return new DefaultKafkaConsumerFactory(oneKafkaProperties.buildConsumerProperties());
        }
    
    
        @ConfigurationProperties(prefix = "sys.kafka.one")
        @Bean
        public KafkaProperties oneKafkaProperties(){
            return new KafkaProperties();
        }
    
    }

（2）第二个kafka配置(主)

public class TwoKafkaConfig {
    
    
        @Bean
        public KafkaListenerContainerFactory twoKafkaFactory(@Autowired @Qualifier("twoConsumerFactory") ConsumerFactory twoConsumerFactory) {
            ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
            factory.setConsumerFactory(twoConsumerFactory);
            factory.setConcurrency(10);
            factory.setBatchListener(true);
            factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
    
            return factory;
        }
    
        @Primary//必须指定一个默认的消费者工厂
        @Bean
        public ConsumerFactory twoConsumerFactory(@Autowired @Qualifier("twoKafkaProperties") KafkaProperties twoKafkaProperties){
    
             return new DefaultKafkaConsumerFactory(twoKafkaProperties.buildConsumerProperties());
        }
    
    
        @Primary//必须指定一个默认的kafka配置
        @ConfigurationProperties(prefix = "sys.kafka.two")
        @Bean
        public KafkaProperties twoKafkaProperties(){
           return new KafkaProperties();
        }
    
    }

（3）kakfka配置导入

@Configuration
    @Import({OneKafkaConfig.class, TwoKafkaConfig.class})
    public class KafkaConfig {
    
        @Bean
        public KafkaConsumer kafkaConsumer(){
            KafkaConsumer consumer = new KafkaConsumer();
            return consumer;
        }
    
    }

（4） 消费者监听

public class KafkaConsumer {
    
    
        @KafkaListener(containerFactory = "oneKafkaFactory", topics = "${sys.kafka.one.topic}")
        public void oneKafkaHandle(List<ConsumerRecord<String,String>> consumerRecords, Acknowledgment ack){
           
            //do somthing 
        }
    
        @KafkaListener(containerFactory = "twoKafkaFactory", topics = "${sys.kafka.two.topic}")
        public void twoKafkaHandle(List<ConsumerRecord<String,String>> consumerRecords, Acknowledgment ack){
    	
            //do somthing
        }
    
    }