Kafka Consumer 未处理 spring boot 中发布的所有请求

Question

已经使用springboot实现了KafkaProducer和KafkaConsumers。 消费者收到消息后，我将调用 REST API 将记录保存到 MongoDB。 但是所有发布到 Kafka 消费者的记录都没有保存到 mongoDB.
例如：发布到 kafkaconsumer 的 5 条记录中只有 3 条记录保存到 mongoDB.

附上kafkaconsumer代码：

KafkaConsumerConfiguration

@EnableKafka
@Configuration
public class KafkaConsumerConfiguration {

    @Value("${spring.kafka.hostname}")
    private String kafkaHostname;
    public static final String GROUP_ID_CONFIG = "group_json1";
    public static final String ENABLE_AUTO_COMMIT_CONFIG = "false";
    public static final String AUTO_OFFSET_RESET_CONFIG = "earliest";


    @Bean
    public ConsumerFactory<String, UserKafkaDTO> userConsumerFactory() {

        Map<String, Object> config = new HashMap<>();

        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaHostname);
        config.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID_CONFIG);
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, ENABLE_AUTO_COMMIT_CONFIG);
        config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, AUTO_OFFSET_RESET_CONFIG);

        JsonDeserializer<UserKafkaDTO> deserializer = new JsonDeserializer<>(UserKafkaDTO.class);
        deserializer.setRemoveTypeHeaders(false);
        deserializer.addTrustedPackages("*");
        deserializer.setUseTypeMapperForKey(true);

        return new DefaultKafkaConsumerFactory<>(config, new StringDeserializer(), deserializer);
    }

    

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, UserKafkaDTO> userKafkaListenerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, UserKafkaDTO> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(userConsumerFactory());
        factory.setBatchListener(true);
        return factory;
    }


  
    @Bean
    @ConditionalOnMissingBean(name = "kafkaListenerContainerFactory")
    public ConcurrentKafkaListenerContainerFactory<String, Object>kafkaListenerContainerFactory() {

        ConcurrentKafkaListenerContainerFactory<String, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setBatchListener(true);
        return factory;

    }

}

KafkaConsumerListener

@EnableKafka
@Configuration
public class KafkaConsumerListener {


    @KafkaListener(topics = { "topicName" }, containerFactory = "userKafkaListenerFactory",autoStartup = "${listen.auto.start}")
    public void consumeJson(UserKafkaDTO userKafkaDTO) {
        invokeAPItoSaveRecordTOMongoDB(userKafkaDTO);

    }

Answer 1

尝试改变
factory.setBatchListener(true); 到 factory.setBatchListener(false);

因为您的侦听器仅处理单个对象。