Mule交易範圍和XA交易

Question

我想在同一個交易的數據庫表和JMS隊列中插入一條記錄。目前我正在使用事務範圍來實現這一點。

<transactional action="ALWAYS_BEGIN" doc:name="Transactional"> 
    <db:insert config-ref="MySQL_Configuration" doc:name="Database"> 
     <db:parameterized-query><![CDATA[INSERT INTO CUSTOMER(NAME, AGE) VALUES(#[payload.name],#[payload.age])]]></db:parameterized-query> 
    </db:insert> 
    <jms:outbound-endpoint queue="${queue.name}" connector-ref="Active_MQ" doc:name="JMS"> 
     <jms:transaction action="ALWAYS_JOIN"/> 
    </jms:outbound-endpoint> 
</transactional> 

但我得到以下錯誤：

org.mule.api.transaction.TransactionException: Endpoint is transactional but transaction does not support it 

現在我使用XA-交易

<jms:activemq-xa-connector name="Active_MQ_XA" specification="1.1" brokerURL="${activemq.broker.url}" validateConnections="true" doc:name="Active MQ" /> 

<db:mysql-config name="MySQL_XA_Configuration" host="${db.host}" port="${db.port}" user="${db.user}" password="${db.password}" database="${db.name}" useXaTransactions="true" doc:name="MySQL Configuration" /> 

<jbossts:transaction-manager doc:name="JBoss Transaction Manager" /> 

<ee:xa-transactional action="ALWAYS_BEGIN" doc:name="Transactional"> 
    <db:insert config-ref="MySQL_XA_Configuration" doc:name="Database"> 
     <db:parameterized-query><![CDATA[INSERT INTO CUSTOMER(NAME, AGE) VALUES(#[payload.name],#[payload.age])]]></db:parameterized-query> 
    </db:insert> 
    <jms:outbound-endpoint queue="${queue.name}" connector-ref="Active_MQ_XA" doc:name="JMS"> 
     <xa-transaction action="ALWAYS_JOIN"/> 
    </jms:outbound-endpoint> 
</ee:xa-transactional> 

這個時候，我得到了另一個異常

java.lang.UnsupportedOperationException 
    at org.enhydra.jdbc.standard.StandardXAConnectionHandle.prepareStatement(StandardXAConnectionHandle.java:380) 
    at org.mule.module.db.internal.domain.xa.ConnectionWrapper.prepareStatement(ConnectionWrapper.java:312) 
    at org.mule.module.db.internal.domain.connection.DefaultDbConnection.prepareStatement(DefaultDbConnection.java:250) 
    at org.mule.module.db.internal.domain.autogeneratedkey.ColumnIndexAutoGeneratedKeyStrategy.prepareStatement(ColumnIndexAutoGeneratedKeyStrategy.java:36) 
    at org.mule.module.db.internal.domain.statement.QueryStatementFactory.create(QueryStatementFactory.java:59) 
    at org.mule.module.db.internal.domain.executor.AbstractSingleQueryExecutor.execute(AbstractSingleQueryExecutor.java:44) 
    at org.mule.module.db.internal.processor.UpdateMessageProcessor.doExecuteQuery(UpdateMessageProcessor.java:59) 
    at org.mule.module.db.internal.processor.AbstractSingleQueryDbMessageProcessor.executeQuery(AbstractSingleQueryDbMessageProcessor.java:42) 
    at org.mule.module.db.internal.processor.AbstractDbMessageProcessor.process(AbstractDbMessageProcessor.java:66) 
    at org.mule.execution.ExceptionToMessagingExceptionExecutionInterceptor.execute(ExceptionToMessagingExceptionExecutionInterceptor.java:24) 
    at org.mule.execution.MessageProcessorNotificationExecutionInterceptor.execute(MessageProcessorNotificationExecutionInterceptor.java:107) 
    at org.mule.execution.MessageProcessorExecutionTemplate.execute(MessageProcessorExecutionTemplate.java:44) 
    at org.mule.processor.BlockingProcessorExecutor.executeNext(BlockingProcessorExecutor.java:94) 
    at org.mule.processor.BlockingProcessorExecutor.execute(BlockingProcessorExecutor.java:56) 
    at org.mule.processor.TransactionalInterceptingMessageProcessor$1.process(TransactionalInterceptingMessageProcessor.java:49) 
    at org.mule.processor.TransactionalInterceptingMessageProcessor$1.process(TransactionalInterceptingMessageProcessor.java:46) 
    at org.mule.execution.ExecuteCallbackInterceptor.execute(ExecuteCallbackInterceptor.java:16) 
    at org.mule.execution.CommitTransactionInterceptor.execute(CommitTransactionInterceptor.java:35) 
    at org.mule.execution.CommitTransactionInterceptor.execute(CommitTransactionInterceptor.java:22) 
    at org.mule.execution.HandleExceptionInterceptor.execute(HandleExceptionInterceptor.java:30) 
    at org.mule.execution.HandleExceptionInterceptor.execute(HandleExceptionInterceptor.java:14) 
    at org.mule.execution.BeginAndResolveTransactionInterceptor.execute(BeginAndResolveTransactionInterceptor.java:67) 
    at org.mule.execution.ResolvePreviousTransactionInterceptor.execute(ResolvePreviousTransactionInterceptor.java:44) 
    at org.mule.execution.SuspendXaTransactionInterceptor.execute(SuspendXaTransactionInterceptor.java:50) 
    at org.mule.execution.ValidateTransactionalStateInterceptor.execute(ValidateTransactionalStateInterceptor.java:40) 
    at org.mule.execution.IsolateCurrentTransactionInterceptor.execute(IsolateCurrentTransactionInterceptor.java:41) 
    at org.mule.execution.ExternalTransactionInterceptor.execute(ExternalTransactionInterceptor.java:48) 
    at org.mule.execution.RethrowExceptionInterceptor.execute(RethrowExceptionInterceptor.java:28) 
    at org.mule.execution.RethrowExceptionInterceptor.execute(RethrowExceptionInterceptor.java:13) 
    at org.mule.execution.TransactionalErrorHandlingExecutionTemplate.execute(TransactionalErrorHandlingExecutionTemplate.java:110) 
    at org.mule.execution.TransactionalErrorHandlingExecutionTemplate.execute(TransactionalErrorHandlingExecutionTemplate.java:30) 
    at org.mule.processor.TransactionalInterceptingMessageProcessor.process(TransactionalInterceptingMessageProcessor.java:55) 
    at org.mule.execution.ExceptionToMessagingExceptionExecutionInterceptor.execute(ExceptionToMessagingExceptionExecutionInterceptor.java:24) 
    at org.mule.execution.MessageProcessorNotificationExecutionInterceptor.execute(MessageProcessorNotificationExecutionInterceptor.java:107) 
    at org.mule.execution.MessageProcessorExecutionTemplate.execute(MessageProcessorExecutionTemplate.java:44) 
    at org.mule.processor.BlockingProcessorExecutor.executeNext(BlockingProcessorExecutor.java:85) 
    at org.mule.processor.BlockingProcessorExecutor.execute(BlockingProcessorExecutor.java:56) 
    at org.mule.execution.ExceptionToMessagingExceptionExecutionInterceptor.execute(ExceptionToMessagingExceptionExecutionInterceptor.java:24) 
    at org.mule.execution.MessageProcessorNotificationExecutionInterceptor.execute(MessageProcessorNotificationExecutionInterceptor.java:107) 
    at org.mule.execution.MessageProcessorExecutionTemplate.execute(MessageProcessorExecutionTemplate.java:44) 
    at org.mule.execution.ExceptionToMessagingExceptionExecutionInterceptor.execute(ExceptionToMessagingExceptionExecutionInterceptor.java:24) 
    at org.mule.execution.MessageProcessorNotificationExecutionInterceptor.execute(MessageProcessorNotificationExecutionInterceptor.java:107) 
    at org.mule.execution.MessageProcessorExecut... 

任何人都可以請解決問題？

Answer 1

看來useXaTransactions="true"不能在mysql db連接器上工作。我配置了明確的mySqlXADataSource。它爲我工作。

<jbossts:transaction-manager doc:name="JBoss Transaction Manager" /> 

<spring:beans> 
    <spring:bean id="mySqlXADataSource" name="MySqlXADataSource" class="com.mysql.jdbc.jdbc2.optional.MysqlXADataSource"> 
     <spring:property name="user" value="${db.user}" /> 
     <spring:property name="password" value="${db.password}" /> 
     <spring:property name="url" value="${mysql.driver.url}" /> 
    </spring:bean> 
</spring:beans> 

<db:mysql-config name="MySQL_XA_Configuration" dataSource-ref="mySqlXADataSource" doc:name="MySQL Configuration" /> 

<jms:activemq-xa-connector name="Active_MQ_XA" specification="1.1" brokerURL="${activemq.broker.url}" validateConnections="true" doc:name="Active MQ" /> 

<ee:xa-transactional action="ALWAYS_BEGIN" doc:name="Transactional"> 
    <db:insert config-ref="MySQL_XA_Configuration" doc:name="Database"> 
     <db:parameterized-query><![CDATA[INSERT INTO CUSTOMER (NAME,AGE)VALUES(#[payload.customer.name],#[payload.customer.age])]]></db:parameterized-query> 
    </db:insert> 
    <jms:outbound-endpoint queue="${queue.name}" connector-ref="Active_MQ_XA" doc:name="JMS"> 
     <xa-transaction action="ALWAYS_JOIN"/> 
    </jms:outbound-endpoint> 
</ee:xa-transactional>