Spring与JMS的集成和XA事务的条件回滚

Question

在通过MQ处理消息时，我想有条件地回滚XA事务并将MQ消息放回原始队列。
失败将被记录到数据库中，并且可以使用基于消息类型和错误的自定义逻辑从数据库中重试。
如果我们无法将错误记录到数据库中，则应回滚整个XA事务并将消息放回队列中。
每条消息都经过多个步骤处理，代码可以处理消息的重新提交/复制。

我有一个解决方案，但它的结果是在丑陋的配置，我想知道是否有更好的方法来实现同样的结果？我在考虑使用一个链，如果消息出错，它会忽略消息
我讨厌服务激活器不是被调用的实际服务。有更好的方法吗？

xml prettyprint-override"><!-- transactionManager is an XA transaction manager --> 
<jms:message-driven-channel-adapter id="batchMessagesIn" 
    destination="batchQueue" 
    error-channel="batchErrorChannel" 
    connection-factory="batchConnectionFactory" 
    channel="batchMessageInChannel" 
    task-executor="integrationTaskExecutor"
    recovery-interval="10000"
    concurrent-consumers="1"
    max-concurrent-consumers="1"
    cache-level="0"
    transaction-manager="transactionManager"/>

<channel id="processMessageFirstStage" />

<!-- The number of stages will depend on the type of message and this type of configuration will be duplicated multiple times -->
<int:service-activator input-channel="processMessageFirstStage" ref="messageServiceAdatper" method="processFirstStage" output-channel="checkIfFirstStageResultedInError"/>  

<!-- Payload is an instance of CustomMessage -->
<int:router input-channel="checkIfFirstStageResultedInError"
    expression="payload.inError"  >
    <mapping value="true" channel="messageInError" />
    <mapping value="false" channel="processMessageSecondStage" />
</int:router>   

<int:service-activator input-channel="processMessageSecondStage" ref="messageServiceAdatper" method="processSecondStage" output-channel="checkIfFirstStageResultedInError"/>    

<int:router input-channel="checkIfSecondStageResultedInError"
    expression="payload.inError"  >
    <mapping value="true" channel="messageInError" />
    <mapping value="false" channel="nullChannel" />
</int:router>   

<channel id="messageInError" />

<int:service-activator input-channel="messageInError" ref="errorMessageProcessor" method="handleMessageError" output-channel="nullChannel"/>

<beans:bean id="messageServiceAdatper" class="com.foo.messaging.MessageServiceAdatperImpl"/>
<beans:bean id="errorMessageProcessor" class="com.foo.messaging.ErrorMessageProcessorImpl"/>


<!-- this error channel will only be used for logging -->
<channel id="batchErrorChannel" />
<stream:stderr-channel-adapter channel="batchErrorChannel" append-newline="true" />

public class CustomMessage {
    private Throwable throwable;
    private String originalMessage;
    private boolean inError;
    private Object payload;
}

public class MessageServiceAdatperImpl {

    @Autowired
    private FirstStageService firstStageService;
    @Autowired
    private SecondStageService secondStageService;

    //Don't let a failure rollback the XA transaction
    @Transactional
    public CustomMessage processFirstStage(CustomMessage customMessage) {
        try {
            firstStageService.processFirstStage(customMessage.getPayload());
        } catch(Throwable e) {
            customMessage.setException(e);
        }
        return customMessage;
    }

    //Don't let a failure rollback the XA transaction
    @Transactional
    public CustomMessage processSecondStage(CustomMessage customMessage) {
        try {
            secondStageService.processSecondStage(customMessage.getPayload());
        } catch(Throwable e) {
           markMessageInError(customMessage,e)
        }
        return customMessage;
    }

    private void markMessageInError(CustomMessage customMessage, Throwable e) {
        customMessage.setThrowable(e);
        customMessage.setInError(true);
    }
}

public class FirstStageService () {
    //Start a new transaction. Code also handles duplicate messages
    @Transactional(propagation=Propagation.REQUIRES_NEW)
    public void processFirstStage() {
        //Do some work
    }
}

public class ErrorMessageProcessorImpl() {
   private static final Marker fatal = MarkerFactory.getMarker("FATAL");

    @Transactional
    public void handleMessageError(CustomMessage customMessage) {
         if (customMessage != null) {

            if (customMessage.isInError()) {
                try {

                    //At this point implment custom logic for logging message into the database. Message can be reprocessed from
                    //database with custom retry limits depending on message type and type of error.

                }
                catch (Throwable e) {
                    //At this point roll back the XA transaction and put the message back on the queue
                    logger.error(fatal, String.format("Fatal error attempting to save error", e));
                    throw new RuntimeException("Fatal error attempting to save error", e);
                }

            }
        }
    }
}

Answer 1

既然您的逻辑与MessageServiceAdatperImpl非常接近，那么使用Spring集成（我的意思是

从另一方面，您可以编写自定义通用路由器，它只是通过其内部逻辑返回通道名称。

或者...路由滑自Spring集成4.1以来