消费者多次消费同一条消息

Question

我有一个消费者消息的 Kafka 消费者。我在 KafkaListenerContainerFactory 中设置了重试模板。但不知道为什么它两次消费相同的消息。当应用程序通过指定计数调用异常重试模板时，Kafka 也会使用相同计数的相同消息。

@Bean
RetryTemplate retryTemplate() {
    RetryTemplate retryTemplate = new RetryTemplate();

    FixedBackOffPolicy fixedBackOffPolicy = new FixedBackOffPolicy();
    fixedBackOffPolicy.setBackOffPeriod(1000L);
    retryTemplate.setBackOffPolicy(fixedBackOffPolicy);

    SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
    retryPolicy.setMaxAttempts(3);
    retryTemplate.setRetryPolicy(retryPolicy);

    return retryTemplate;
}


@Bean("KafkaListenerContainerFactory")
public ConcurrentKafkaListenerContainerFactory<String, String> KafkaListenerContainerFactory(RetryTemplate retryTemplate) {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setRetryTemplate(retryTemplate);
    factory.setRecoveryCallback(context -> {
        log.error("Maximum retry policy has been reached");
        return null;
    });
    factory.setConcurrency(Integer.parseInt(kafkaConcurrency));
    return factory;
}

消费者

@KafkaListener(topics = "${kafka.topic.json}", containerFactory = "kafkaListenerContainerFactory")
public void recieveSegmentService(String KafkaPayload) throws Exception {
    KafkaSegmentTrigger kafkaSegmentTrigger;
    kafkaSegmentTrigger = TransformUtil.fromJson(KafkaPayload, KafkaSegmentTrigger.class);
    log.info("Trigger recieved from segment service {}", kafkaSegmentTrigger);
    try {
        processMessage(kafkaSegmentTrigger);
    } catch (Exception e) {
        retryTemplate.execute(arg0 -> {
            processMessage(kafkaSegmentTrigger);
            return null;
        });
    }finally {
    }
}

processMessage 正在抛出异常

Answer 1

您已经嵌套了RetryTemplates - 一个在侦听器外部，一个在内部。如果您在两个地方都使用相同的模板，您将获得 12 次尝试（3 次由侦听器内部的 x4 侦听器适配器驱动）。

使用其中一个。