事务

Doubao 中英对照 Transactions

Spring Rabbit 框架在同步和异步用例中支持自动事务管理，并提供多种不同语义，可以像现有 Spring 事务用户所熟悉的那样以声明方式选择。这使得许多（即使不是大多数）常见消息传递模式都易于实现。

向框架表明所需事务语义有两种方式。在 RabbitTemplate 和 SimpleMessageListenerContainer 中，都有一个 channelTransacted 标志，如果为 true，它会告诉框架使用事务性 channel，并以提交或回滚结束所有操作（发送或接收）（取决于结果），异常则表示回滚。另一种信号是用 Spring 的 PlatformTransactionManager 实现之一提供一个外部事务，作为正在进行操作的上下文。当框架正在发送或接收消息时如果已经有一个事务在进行中，并且 channelTransacted 标志为 true，则消息事务的提交或回滚会推迟到当前事务结束。如果 channelTransacted 标志为 false，则消息操作不应用任何事务语义（即自动确认）。

channelTransacted 标志是配置时的设置。它在创建 AMQP 组件时声明并处理一次，通常在应用启动时。外部事务原则上更动态，因为系统在运行时响应当前线程状态。不过在实践中，当事务以声明方式分层应用到应用程序时，它往往也是一个配置设置。

对于使用 RabbitTemplate 的同步用例，外部事务由调用者提供，可以按喜好以声明方式或命令方式提供（通常的 Spring 事务模型）。下例展示了声明式方式（通常更受青睐，因为它是非侵入式的），其中模板已配置为 channelTransacted=true：

@Transactional
public void doSomething() {
    String incoming = rabbitTemplate.receiveAndConvert();
    // do some more database processing...
    String outgoing = processInDatabaseAndExtractReply(incoming);
    rabbitTemplate.convertAndSend(outgoing);
}

在上例中，在一个标记为 @Transactional 的方法内，接收 String 负载、转换并作为消息体发送。如果数据库处理因异常失败，传入的消息会被退回 broker，传出的消息不会发送。这适用于事务性方法链中任何使用 RabbitTemplate 的操作（除非，例如直接操纵 Channel 提前提交事务）。

对于使用 SimpleMessageListenerContainer 的异步用例，如果需要外部事务，必须由容器在设置监听器时请求。为表明需要外部事务，用户在配置容器时向其提供 PlatformTransactionManager 的实现。下例展示了如何这样做：

@Configuration
public class ExampleExternalTransactionAmqpConfiguration {

    @Bean
    public SimpleMessageListenerContainer messageListenerContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(rabbitConnectionFactory());
        container.setTransactionManager(transactionManager());
        container.setChannelTransacted(true);
        container.setQueueName("some.queue");
        container.setMessageListener(exampleListener());
        return container;
    }

}

在上例中，事务管理器作为从另一个 bean 定义（未显示）注入的依赖添加，并且 channelTransacted 标志也被设置为 true。其效果是：如果监听器因异常失败，事务会回滚，消息也会被退回 broker。重要的是，如果事务提交失败（例如由于数据库约束错误或连接问题），AMQP 事务也会回滚，消息被退回 broker。这有时被称为“尽力而为的一阶段提交”，是可靠消息传递中一种非常强大的模式。如果上例中 channelTransacted 标志设置为 false（默认值），仍然会为监听器提供外部事务，但所有消息操作都会自动确认，因此其效果是即使业务操作回滚，消息操作也会被提交。

条件回滚

在 1.6.6 版本之前，当使用外部事务管理器（例如 JDBC）时，向容器的 transactionAttribute 添加回滚规则没有效果。异常总是会回滚事务。

此外，当在容器的通知链中使用事务通知时，条件回滚也不太有用，因为所有监听器异常都被包装在 ListenerExecutionFailedException 中。

第一个问题已被修正，规则现在能正确应用。此外，现在提供了 ListenerFailedRuleBasedTransactionAttribute。它是 RuleBasedTransactionAttribute 的子类，唯一区别是它能识别 ListenerExecutionFailedException，并使用此类异常的原因（cause）来应用规则。此事务属性可以直接在容器中使用，也可以通过事务通知使用。

下例使用了此规则：

@Bean
public AbstractMessageListenerContainer container() {
    ...
    container.setTransactionManager(transactionManager);
    RuleBasedTransactionAttribute transactionAttribute =
        new ListenerFailedRuleBasedTransactionAttribute();
    transactionAttribute.setRollbackRules(Collections.singletonList(
        new NoRollbackRuleAttribute(DontRollBackException.class)));
    container.setTransactionAttribute(transactionAttribute);
    ...
}

关于收到消息回滚的说明

AMQP 事务仅适用于发送到 broker 的消息和确认。因此，当 Spring 事务发生回滚且消息已被接收时，Spring AMQP 不仅要回滚事务，还要手动拒绝该消息（有点像 nack，但规范不这样称呼它）。消息拒绝时采取的操作与事务无关，取决于 defaultRequeueRejected 属性（默认值：true）。关于拒绝失败消息的更多信息，参见消息监听器与异步情形。

关于 RabbitMQ 事务及其限制的更多信息，参见 RabbitMQ Broker Semantics。

备注

在 RabbitMQ 2.7.0 之前，此类消息（以及 channel 关闭或中止时任何未确认的消息）会进入 Rabbit broker 中队列的尾部。从 2.7.0 开始，被拒绝的消息会进入队列头部，类似于 JMS 中回滚的消息。

备注

以前，事务回滚时的消息重新入队在本地事务与提供 TransactionManager 的情形之间不一致。在前一种情况下，应用正常的重新入队逻辑（AmqpRejectAndDontRequeueException 或 defaultRequeueRejected=false）（参见消息监听器与异步情形）。而使用事务管理器时，消息在回滚时会被无条件重新入队。从 2.0 版本开始，行为保持一致，两种情况下都应用正常的重新入队逻辑。要恢复以前的行为，你可以将容器的 alwaysRequeueWithTxManagerRollback 属性设置为 true。参见消息监听器容器配置。

使用 RabbitTransactionManager

RabbitTransactionManager 是在外部事务内部执行 Rabbit 操作并与之同步的一种替代方案。此事务管理器是 PlatformTransactionManager 接口的实现，应与单个 Rabbit ConnectionFactory 一起使用。

important

此策略无法提供 XA 事务——例如，无法在消息传递与数据库访问之间共享事务。

应用代码需要通过 ConnectionFactoryUtils.getTransactionalResourceHolder(ConnectionFactory, boolean) 获取事务性 Rabbit 资源，而不是标准的 Connection.createChannel() 调用及随后的 channel 创建。使用 Spring AMQP 的 RabbitTemplate 时，它会自动检测线程绑定的 Channel 并自动参与其事务。

使用 Java 配置时，你可以通过以下 bean 设置一个新的 RabbitTransactionManager：

@Bean
public RabbitTransactionManager rabbitTransactionManager() {
    return new RabbitTransactionManager(connectionFactory);
}

如果你更喜欢 XML 配置，可以在 XML 应用程序上下文文件中声明以下 bean：

<bean id="rabbitTxManager"
      class="org.springframework.amqp.rabbit.transaction.RabbitTransactionManager">
    <property name="connectionFactory" ref="connectionFactory"/>
</bean>

事务同步

将 RabbitMQ 事务与其他某种（例如 DBMS）事务同步，可提供“尽力而为的一阶段提交”语义。RabbitMQ 事务有可能在事务同步的 after completion 阶段提交失败。spring-tx 基础设施会将此作为错误记录日志，但不会向调用代码抛出异常。从 2.3.10 版本开始，你可以在事务于处理该事务的同一线程上提交之后，调用 ConnectionUtils.checkAfterCompletion()。如果没有异常发生，它会直接返回；否则会抛出 AfterCompletionFailedException，该异常带有一个表示完成时同步状态的属性。

通过调用 ConnectionFactoryUtils.enableAfterCompletionFailureCapture(true) 启用此功能；这是一个全局标志，适用于所有线程。