asynchronous - Mule ESB 3.3 如何同时启动事务流

Question

我想同时处理一组消息，但我无法设法使它们成为事务性的，除非我将 VM 设置为请求-响应...在这种情况下，处理不是并发的。

Mule documentation指出“Mule 事务是在同步端点上配置的”，但我不太理解这个限制。
很明显，在您希望成为事务性的流程中，不应产生异步流程，但(对我而言)不清楚为什么不能(从非 tx 主流程)启动任何数量的异步流程，每个流程都是交易。

换句话说，为什么这个工作正常:

但如果我将 VM 更改为“单向”，它会失败:

org.mule.transaction.IllegalTransactionStateException: Can only bind "javax.sql.DataSource/java.sql.Connection" type resources

有办法解决这个问题吗？

流的 XML:

<?xml version="1.0" encoding="UTF-8"?>
<mule>
    <spring:beans>
        <spring:bean id="dataSource" name="dataSource" class="org.enhydra.jdbc.standard.StandardDataSource" destroy-method="shutdown">
            <spring:property name="driverName" value="org.h2.Driver" />
            <spring:property name="url" value="jdbc:h2:tcp://localhost/~/mule" />
            <spring:property name="user" value="sa" />
            <spring:property name="password">
                <spring:value></spring:value>
            </spring:property>
        </spring:bean>

        <spring:bean id="transactionFactory" name="transactionFactory" class="org.mule.transport.jdbc.JdbcTransactionFactory" />

    </spring:beans>

    <jdbc:connector name="dbConnector" dataSource-ref="dataSource" validateConnections="true" queryTimeout="-1" pollingFrequency="0" doc:name="Database" />

    <flow name="triggerFlow" doc:name="triggerFlow">
        <http:inbound-endpoint exchange-pattern="request-response" host="localhost" port="8081" doc:name="HTTP" />
        <set-payload value="#[[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]]" doc:name="[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]"/>
        <collection-splitter doc:name="Collection Splitter"/>
        <vm:outbound-endpoint exchange-pattern="request-response" path="flow" doc:name="flow" />
    </flow>

    <flow name="flow" doc:name="flow">
        <vm:inbound-endpoint exchange-pattern="request-response" path="flow" doc:name="flow">
            <custom-transaction factory-ref="transactionFactory" action="ALWAYS_BEGIN" timeout="10" />
        </vm:inbound-endpoint>
        <logger message="#[payload]" level="INFO" doc:name="Logger"/>
        <jdbc:outbound-endpoint exchange-pattern="request-response" queryKey="insert" queryTimeout="-1" connector-ref="dbConnector" doc:name="Insert 1">
            <jdbc:transaction action="ALWAYS_JOIN" />
            <jdbc:query key="insert" value="insert into test values (#[payload], 'Test 1')" />
        </jdbc:outbound-endpoint>
        <jdbc:outbound-endpoint exchange-pattern="request-response" queryKey="insert2" queryTimeout="-1" connector-ref="dbConnector" doc:name="Insert 2">
            <jdbc:transaction action="ALWAYS_JOIN" />
            <jdbc:query key="insert2" value="insert into test values (#[payload + 10], 'Test 2')" />
        </jdbc:outbound-endpoint>
    </flow>

</mule>

提前致谢。

Answer 1

Mule documentation states that "Mule transactions are configured on synchronous endpoints", but I don't quite understand this restriction.

这个限制是因为在 Mule 中，就像在 Spring 中，更普遍的是在 Java 中，事务是线程绑定(bind)的。对于异步流程，涉及多个线程，因此无法维护事务线程关联。

所以不，你不能拆分/ fork /并行化/异步处理消息，也不能在 Mule 中进行交易。

org.mule.transaction.IllegalTransactionStateException: Can only bind "javax.sql.DataSource/java.sql.Connection" type resources

但这与 IMO 的第一个问题无关:这是因为您通过 custom-transaction 强行尝试在 JDBC 事务中注册 VM 端点。这是行不通的。如果要注册异构资源，请使用 XA 事务。

编辑:根据您在评论中所说的，您不想在事务中注册 VM 端点，因此只需在此处注册 JDBC 端点:

<transactional action="ALWAYS_BEGIN">
    <jdbc:outbound-endpoint exchange-pattern="request-response"
        queryKey="insert" queryTimeout="-1" connector-ref="dbConnector"
        doc:name="insert into test values (1, 'Test 1')">
        <jdbc:transaction action="ALWAYS_JOIN" />
        <jdbc:query key="insert"
            value="insert into test values (1, 'Test 1')" />
    </jdbc:outbound-endpoint>
    <jdbc:outbound-endpoint exchange-pattern="request-response"
        queryKey="insert2" queryTimeout="-1" connector-ref="dbConnector"
        doc:name="insert into test values (2, 'Test 2')">
        <jdbc:transaction action="ALWAYS_JOIN" />
        <jdbc:query key="insert2"
            value="insert into test values (2, 'Test 2')" />
    </jdbc:outbound-endpoint>
</transactional>

这适用于单向入站端点。