spring amqp事务语义

Question

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

用例如下所示：

>

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

    rabbitTemplate.convertAndSend("myExchange", "payload.create", payload);

    MyEntity savedEntity = entityRepository.save(entity);
}

数据库操作期间发生故障时的预期行为是，接收到的消息回滚到总线（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;
        }

有人能解释一下那里发生了什么吗？为什么第二个案例也有效？如果在SimpleMessageListenerContainer上注册了PlatformTransactionManager，内部会有什么不同。

Answer 1

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

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

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

假设侦听器不是事务性的，但一些下游组件是；假设侦听器成功调用了该组件，然后在抛出异常之前做了更多的工作。在这种情况下，数据库提交将成功，消息将被拒绝。这可能不是期望的行为，尤其是在消息被重新请求的情况下。在这种情况下，通过注入数据库tm来同步兔子事务和数据库事务通常更好。

在您的情况下，在db提交和Rabbit ack之间发生失败的可能性很小，因此这实际上不适用于您的情况，并且您不需要容器中的tm。