Spring集成：2个激活器之间的交易困难

Posted 2023-02-26

【中文标题】Spring集成：2个激活器之间的交易困难

【英文标题】：Spring integration: difficulty with transaction between 2 activators

【发布时间】：2011-11-13 19:31:08

【问题描述】：

我有这个用例。

第一链：

<int:chain input-channel="inserimentoCanaleActivate" output-channel="inserimentoCanalePreRouting">      
    <int:service-activator ref="inserimentoCanaleActivator" method="activate" />                
</int:chain>

这是相关代码：

@Override
@Transactional(propagation = Propagation.REQUIRES_NEW)
public EventMessage<ModificaOperativitaRapporto> activate(EventMessage<InserimentoCanale> eventMessage) 
    ...
    // some Database changes
    dao.save(myObject);

一切都很好。

那我还有一条链：

<int:chain id="onlineCensimentoClienteChain" input-channel="ONLINE_CENSIMENTO_CLIENTE" output-channel="inserimentoCanaleActivate">
    <int:service-activator ref="onlineCensimentoClienteActivator" method="activate" />
    <int:splitter expression="payload.getPayload().getCanali()" />
</int:chain>

以及相关的激活器：

@Override
public EventMessage<CensimentoCliente> activate(EventMessage<CensimentoCliente> eventMessage) 
    ...
    // some Database changes
    dao.save(myObject);

下面描述的CensimentoCliente有效载荷有一个List的第一个链的有效载荷，因此我使用拆分器拆分列表并重用第一个链的代码。

public interface CensimentoCliente extends Serializable 

    Collection<? extends InserimentoCanale> getCanali();

    void setCanali(Collection<? extends InserimentoCanale> canali);
    ...

但是由于每个激活器都有他的事务定义（因为第一个激活器可以在没有第二个激活器的情况下存在）我有一个事务被分离的用例。

目标是让两条链的数据库修改成为同一事务的一部分。

有什么帮助吗？

亲切的问候 马西莫

【问题讨论】：

您找到解决方案了吗？

【参考方案1】：

您可以通过创建自定义通道（或其他自定义组件，但这是最简单的方法）将消息调度包装在 TransactionTemplate 回调执行中来实现：

public class TransactionalChannel extends AbstractSubscribableChannel 

    private final MessageDispatcher dispatcher = new UnicastingDispatcher();
    private final TransactionTemplate transactionTemplate;

    TransactionalChannel(TransactionTemplate transactionTemplate) 
        this.transactionTemplate = transactionTemplate;
    

    @Override
    protected boolean doSend(final Message<?> message, long timeout) 
        return transactionTemplate.execute(new TransactionCallback<Boolean>() 
            @Override
            public Boolean doInTransaction(TransactionStatus status) 
                return getDispatcher().dispatch(message);
            
        );
    

    @Override
    protected MessageDispatcher getDispatcher() 
        return dispatcher;
    

在您的 XML 中，您可以定义您的渠道和交易模板，并像引用任何其他渠道一样引用您的自定义渠道：

    <bean id="transactionalChannel" class="com.***.TransactionalChannel">
        <constructor-arg>
           <bean class="org.springframework.transaction.support.TransactionTemplate">
              <property name="transactionManager" ref="transactionManager"/>
              <property name="propagationBehavior" value="#T(org.springframework.transaction.TransactionDefinition).PROPAGATION_REQUIRES_NEW"/>
          </bean>
       </constructor-arg>
    </bean>

对于您的示例，您也许可以使用桥接器通过新通道传递消息：

<int:bridge input-channel="inserimentoCanaleActivate" output-channel="transactionalChannel" /> 
<int:chain input-channel="transactionalChannel" output-channel="inserimentoCanalePreRouting">      
    <int:service-activator ref="inserimentoCanaleActivator" method="activate" />                
</int:chain>

【参考方案2】：

您在服务方法上有<service-activator> 和@Transactional，事务将仅限于该方法调用。 如果您想对整个消息流（或其部分）进行交易，您应该在之前的某个地方声明 TX 通知。 如果您的通道是直接的，则所有服务调用都将使用相同的事务进行包装。 实现愿望的最简单方法，用@Transactional 编写简单的@Gateway 接口并从消息流的开头调用它。

澄清一点关于交易 Understanding Transactions in Message flows

【讨论】：

您可以发布或链接一些代码示例吗？我有同样的问题。谢谢！

我在答案中添加了一个关于 SI 中 TX 的链接

我以前读过这个文档，但仍有疑问。我已在***.com/questions/18901510/…

中发布我的案例

【参考方案3】：

这些是在修改 2 个独立的关系数据库吗？如果是这样，您正在查看 XA 事务。现在，如果您在像 tomcat 这样的非 XA 容器上运行它，所有这些都必须在由事务管理器监视的单个线程中完成 - （您将不得不捎带实际触发这些事件的事务管理器）。事务管理器可以是 JMS 消息或针对某些数据源的轮询器。此外，此处理必须在单个线程中完成，以便 spring 可以帮助您在单个事务中运行整个过程。

最后一点，不要在服务激活器之间引入线程池/队列。这会导致激活器在不同的线程中运行