spring-amqp事务语义

Question

目前，我正在测试一个与spring数据库事务相关的消息传递事务的简单示例。

用例如下：

• 接收到消息
• 一条消息被发送
• 数据库更新 @Transactional公共void handleMessage(EventPayload event) { MyEntity实体=新的MyEntity()；entity.setName(event.getName())；rabbitTemplate.convertAndSend("myExchange"，"payload.create"，有效载荷)；MyEntity savedEntity =entityRepository.save(实体)}；

在数据库操作期间发生故障时，预期的行为是将接收到的消息回滚到总线(DefaultRequeueRejected = false)并进入死信队列。另外，应该回滚发送的消息。

我可以通过以下配置来实现这一点：

@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory, MessageConverter messageConverter) {
    RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
    rabbitTemplate.setMessageConverter(messageConverter);
    rabbitTemplate.setChannelTransacted(true);
    return rabbitTemplate;
}

@Bean
    SimpleMessageListenerContainer subscriberListenerContainer(ConnectionFactory connectionFactory,
                                                              MessageListenerAdapter listenerAdapter,
                                                              PlatformTransactionManager transactionManager) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames(SUBSCRIBER_QUEUE_NAME);
        container.setMessageListener(listenerAdapter);
        container.setChannelTransacted(true);
        container.setTransactionManager(transactionManager);
        container.setDefaultRequeueRejected(false);
        return container;
    }

所以这很好--我不明白的是，如果我不在SimpleMessageListenerContainer上设置事务管理器，所观察到的行为是完全相同的。因此，如果我配置了以下参数，那么bebavior不会更改：

@Bean
        SimpleMessageListenerContainer subscriberListenerContainer(ConnectionFactory connectionFactory,
                                                                  MessageListenerAdapter listenerAdapter) {
            SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
            container.setConnectionFactory(connectionFactory);
            container.setQueueNames(SUBSCRIBER_QUEUE_NAME);
            container.setMessageListener(listenerAdapter);
            container.setDefaultRequeueRejected(false);
            return container;
        }

有人能解释一下那里发生了什么吗？为什么第二种情况也起作用？如果PlatformTransactionManager在SimpleMessageListenerContainer上注册，内部有什么不同。

Answer 1

假设transactionManager是您的数据库tm，因为您的侦听器是@Transactional，所以对于这些场景没有太大的区别。

在第一种情况下，事务是在调用侦听器之前由容器启动的(实际上是在从内部队列检索消息之前启动的，因此即使没有消息，事务也会启动)。

在第二种情况下，事务由事务拦截器在调用侦听器时启动。

假设侦听器不是事务性的，但某些下游组件是事务性的；假设侦听器成功地调用了该组件，那么在抛出异常之前做一些更多的工作。在这种情况下，DB提交将是成功的，消息将被拒绝。这可能不是所期望的行为，特别是当消息被请求时。在这种情况下，通常最好通过注入数据库tm来使兔子事务与数据库事务同步。

在您的示例中，db提交和兔子ack之间的失败可能性很小，因此这实际上不适用于您的情况，也不需要容器中的tm。