在第一次重试时执行操作,以将消息保存到数据库中

我试图将到达数据库上RabbitMQ队列的每条消息都保存下来,只是为了记录日志。之后,需要正常处理消息。

问题是:此队列的重试策略配置为 RetryOperationsInterceptor ,并且每次处理消息时出现错误,都会重新查询并再次处理该消息。保存消息的逻辑是在读取队列的侦听器中,因此,我没有只在数据库中保存一条消息,而是 3 (我配置的重试次数)。

查看我的 RetryOperationsInterceptor :

@Bean
public RetryOperationsInterceptor defaultRetryOperationsInterceptor() {
    return RetryInterceptorBuilder.stateless()
            .maxAttempts(3) 
            .backOffOptions(2000, 2.0, 10000)
            .build();
}

集装箱工厂:

@Bean(name = FACTORY_CONTAINER_NAME)
public SimpleRabbitListenerContainerFactory factoryQueueExample(ConnectionFactory connectionFactory,
                                                                        RetryOperationsInterceptor defaultRetryOperationsInterceptor) {

    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setMessageConverter(getMessageConverter());
    factory.setDefaultRequeueRejected(false);

    Advice[] adviceChain = {defaultRetryOperationsInterceptor};
    factory.setAdviceChain(adviceChain);

    return factory;
}

队列侦听器:

@Slf4j
@Component
@AllArgsConstructor
public class MessageListener {

    private final MessageRepository messageRepository;
    private final MessageService messageService;

    @RabbitListener(queues = MessageConfiguration.QUEUE,
        containerFactory = MessageConfiguration.FACTORY_CONTAINER_NAME)
    public void process(SomeMessage someMessage) {

        messageRepository.save(someMessage); // transform to a entity and save on database
        messageService.process(someMessage); // process message
    }
}

我不知道是否是相关信息,但此队列还关联了一个DLQ。重试后,消息将转到DLQ队列。

我的想法是在重试拦截器中找到一些可以在第一次尝试时调用服务的东西,以便只保存一次消息。

我也愿意用其他方法来解决这个问题,比如将尝试次数与消息一起保存,只是为了表明不是在数据库中保存重复的消息,而是在不同的尝试中保存相同的消息。