单个事务中的 Spring 集成和 JDBC

Question

所以设置如下：

<tx:advice id="txAdvice2" transaction-manager="dataSourceTransactionManager">
    <tx:attributes>
        <tx:method name="*" rollback-for="Throwable" no-rollback-for="ListenerExecutionFailedException"/>
    </tx:attributes>
</tx:advice>

<int-amqp:inbound-channel-adapter channel="input-channel" queue-names="probni" message-converter="jsonMessageConverter"
                                  channel-transacted="true"
                                  advice-chain="txAdvice2" />

<int:chain input-channel="input-channel" output-channel="output-channel">
    <int:service-activator ref="h1Handler" method="handle" />
    <int:service-activator ref="h2Handler" method="handle" />
    <int:service-activator ref="h3Handler" method="handle" />
    <int:splitter  />
</int:chain>

<int-amqp:outbound-channel-adapter channel="output-channel" exchange-name="outputit" amqp-template="rabbitTemplate" />

如果在这个线程的执行过程中（因为所有这个链 amqpIN-process-amqpOUT shold 在单线程中执行）我抛出 ListenerExecutionFailedException，dataSourceTransactionManager 会做提交，但是 amqp 也会重新排队消息，因为传播了异常。

在这种情况下，我如何告诉rabbit 确认消息成功？

另外，我看到我必须放入 no-rollback-for 属性实际异常类，因为我的内部异常仅存储在 RuleBasedTransactionAttribute 未检查的“原因”属性中。

还有一件事，如果我像这样进行配置：

<int-amqp:inbound-channel-adapter channel="input-channel" queue-names="probni" message-converter="jsonMessageConverter"
                                  channel-transacted="true"
                                  transaction-manager="dataSourceTransactionManager"
                                  transaction-attribute="transactionAttribute" />

根本不考虑作为 RuleBasedTransactionAttribute 的 transactionAttribute，即使我正确设置了 no-rollback-for，也始终回滚 dataSourceTransactionManager。

谢谢！

Answer 1

您可以将自定义ErrorHandler 添加到侦听器容器中（您必须在外部配置容器并在container 属性中提供引用）。

默认错误处理程序是 ConditionalRejectingErrorHandler 和 DefaultExceptionStrategy，它将某些 LEFE 原因异常视为致命：

    private boolean isCauseFatal(Throwable cause) {
        return cause instanceof MessageConversionException
                || cause instanceof org.springframework.messaging.converter.MessageConversionException
                || cause instanceof MethodArgumentNotValidException
                || cause instanceof MethodArgumentTypeMismatchException
                || cause instanceof NoSuchMethodException
                || cause instanceof ClassCastException
                || isUserCauseFatal(cause);
    }

从版本 1.6.4 开始，您可以将默认的 DefaultExceptionStrategy 子类化并将您的原因添加到 isUserCauseFatal()。

在 1.6.4 之前，您必须提供自己的 FatalExceptionStrategy（或错误处理程序实现）。

对于致命的原因，处理程序抛出一个AmqpRejectAndDontRequeueException，告诉容器nack（而不是重新排队）消息。

编辑

顺便说一句，你不需要包装异常，容器会为你做这些......

protected Exception wrapToListenerExecutionFailedExceptionIfNeeded(Exception e, Message message) {
    if (!(e instanceof ListenerExecutionFailedException)) {
        // Wrap exception to ListenerExecutionFailedException.
        return new ListenerExecutionFailedException("Listener threw exception", e, message);
    }
    return e;
}

EDIT2

我的错误，ErrorHandler 可以使用 error-handler 属性指定。

EDIT3

或者，只需抛出一个AmqpRejectAndDontRequeueException（将包裹在LEFE 中）。