java - Kafka 2 消费者工厂监听器没有不断连接

问题描述

我们在项目中使用 Spring Kafka 2.1.4.RELEASE 版本，我们有以下配置：

@EnableKafka
public class KafkaConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Configuration
    class ProducerConfig {
        @Bean
        public Map<String, Object> producerConfigs() {
            Map<String, Object> props = new HashMap<>();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ASerializer.class);
            return props;
        }

        @Bean
        public ProducerFactory<String, A> producerFactory() {
            return new DefaultKafkaProducerFactory<>(producerConfigs());
        }

        @Bean
        public KafkaTemplate<String, A> kafkaTemplate() {
            return new KafkaTemplate<>(producerFactory());
        }
    }

    @Configuration
    class ConsumerConfig {
        @Value("${spring.kafka.consumer.group-id}")
        private String groupId;
        @Value("${spring.kafka.consumer.auto-offset-reset}")
        private String autoOffsetReset;
        @Value("${spring.kafka.consumer.enable-auto-commit}")
        private boolean enableAutoCommit;
        @Value("${spring.kafka.consumer.max-poll-records}")
        private Integer maxPollRecords;

        @Bean
        public Map<String, Object> firstConsumerConfig() {
            Map<String, Object> props = getCommonConsumerConfig();
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ADeserializer.class);
            return props;
        }

        @Bean
        public Map<String, Object> secondConsumerConfig() {
            Map<String, Object> props = getCommonConsumerConfig();
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, BDeserializer.class);
            return props;
        }

        @Bean
        public ConsumerFactory<String, A> firstConsumerFactory() {
            return new DefaultKafkaConsumerFactory<>(firstConsumerConfig());
        }

        @Bean
        public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, A>> firsttContainerFactory() {
            ConcurrentKafkaListenerContainerFactory<String, A> factory = new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(firstConsumerFactory());
            return factory;
        }

        @Bean
        public ConsumerFactory<String, B> secondConsumerFactory() {
            return new DefaultKafkaConsumerFactory<>(secondConsumerConfig());
        }

        @Bean
        public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, B>> outputTopicContainerFactory() {
            ConcurrentKafkaListenerContainerFactory<String, B> factory = new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(secondConsumerFactory());
            return factory;
        }


        private Map<String, Object> getCommonConsumerConfig() {
            Map<String, Object> props = new HashMap<>();
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
            props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
            props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
            props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
            return props;
        }
    }
}

如您所见，我们使用了 2 个消费者工厂。我们有以下消费类：

@ConfigurationProperties
@Service
public class Listener {

    @KafkaListener(containerFactory = "firstContainerFactory", topics = "someTopic")
    public void firstListener(@Payload A first) {
        //some logic
    }

    @KafkaListener(containerFactory = "secondContainerFactory", topics = {
     //topic list       
    })
    public void secondTopicListener(@Payload B second) {
       //some logic

    }


}

所以我们在启动这个应用程序时注意到的是，它并没有一直连接到这两个主题。有时它只连接到第二个主题或只连接到第一个主题，并且可能连接到第一个和第二个主题（这是正确的）。那么您能否帮助了解这里配置错误的内容？

标签： javaspring-bootkafka-consumer-apispring-kafka

java - Kafka 2 消费者工厂监听器没有不断连接

问题描述

解决方案

推荐阅读