Spring AMQP事务处理和重试

Question

我正在尝试SRSpring AMQP关于事务性消息处理的功能。

我有以下设置-我有一个消息消费者，被注释为@Transactional

@Transactional
    public void handleMessage(EventPayload event) {
        Shop shop = new Shop();
        shop.setName(event.getName());

        Shop savedShop = shopService.create(shop);

        log.info("Created shop {} from event {}", shop, event);
    }

在商店服务中。创建I保存店铺并发送另一条关于创建的消息：

@Transactional(propagation = REQUIRED)
@Component
public class ShopService {

   ...

    public Shop create(Shop shop) {
        eventPublisher.publish(new EventPayload(shop.getName()));
        return shopRepository.save(shop);
    }
}

我想实现以下目标——如果数据库操作成功，则在create方法中发送的消息应该直接发送到代理。如果失败，则不发送消息，并回滚接收到的消息。

我还配置了重试功能，所以我希望每条消息在被拒绝之前都会重试3次：

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

我观察到以下行为：

当我按照以下方式配置容器时，消息会重试3次，但每次都会在shopService中重试消息。创建被发送到代理：

@Bean
    SimpleMessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory,
                                                            MessageListenerAdapter listenerAdapter) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames(testEventSubscriberQueue().getName());
        container.setMessageListener(listenerAdapter);
        container.setChannelTransacted(true);
        container.setAdviceChain(new Advice[]{retryOperationsInterceptor()});
        return container;
    }

因此，我尝试将平台TransactionManager传递给容器-

@Bean
    SimpleMessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory,
                                                            MessageListenerAdapter listenerAdapter,
                                                            PlatformTransactionManager transactionManager) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames(testEventSubscriberQueue().getName());
        container.setMessageListener(listenerAdapter);
        container.setChannelTransacted(true);
        container.setTransactionManager(transactionManager);
        container.setAdviceChain(new Advice[]{retryOperationsInterceptor()});
        return container;
    }

现在，在shop Service.create中发送的消息只有在数据库事务成功的情况下才会发送给代理——这就是我想要的——但是现在消息会无限期重试——并且不会在3次按配置退出后丢弃。但是似乎应用了BackOff设置——所以重试之间有一段时间。

从业务的角度来看，所描述的设置没有真正意义——我试图理解和评估事务功能。

我正在使用sping-amqp 1.5.1。释放

谢谢你的提示。

Answer 1

我有同样的要求，一个用@Transactional注释的@RabbitListener，我想重试后退。它甚至可以在以下配置中无状态工作：

   @Bean
   public RetryOperationsInterceptor retryOperationsInterceptor( ) {
      return RetryInterceptorBuilder.stateless()
            .maxAttempts( 3 )
            .recoverer( new RejectAndDontRequeueRecoverer() )
            .backOffOptions(1000, 2, 10000)
            .build();
   }

   @Bean
   public Jackson2JsonMessageConverter producerJackson2MessageConverter( ObjectMapper objectMapper ) {
      Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter( objectMapper );
      jackson2JsonMessageConverter.setCreateMessageIds( true );
      return jackson2JsonMessageConverter;
   }      

   @Bean
   SimpleRabbitListenerContainerFactory  rabbitListenerContainerFactory( ConnectionFactory connectionFactory,
                                                            PlatformTransactionManager transactionManager,
                                                                   Jackson2JsonMessageConverter converter) {
      SimpleRabbitListenerContainerFactory  container = new SimpleRabbitListenerContainerFactory ();
      container.setConnectionFactory(connectionFactory);

      container.setChannelTransacted(true);
      container.setTransactionManager(transactionManager);

      container.setAdviceChain( retryOperationsInterceptor() );

      container.setMessageConverter( converter );
      return container;
   }

要使用stateless（），使用RejectAndDontRequeueRecoverer非常重要，因为否则重试将有效，但消费者会在默认情况下将消息放回队列。然后消费者将再次检索它，应用重试策略，然后无限次地将其放回队列。