トランザクションを持つ非同期サービスへのメッセージの送信

Question

非同期プロトコルを使用して送信する必要のあるメッセージを含む永続的なトランザクションキューがあります。各メッセージはそれ自身のトランザクションで送信する必要がありますが、のフライト中のメッセージ数は、メッセージごとにスレッドを使用することを排除します。スループット要件により、中間状態が持続することはありません。

JmsTransactionManagerのコードを見ると、トランザクションのリソースをThreadLocalに格納するTransactionSynchronizationManagerが使用されています。ですから、私はPlatformTransactionManagerを実装して、何とか1つのスレッド内で複数のトランザクションを管理する必要があるようです。このはと思われます。極端な...

この複雑さを解消するためのSpring Integrationユニットがいくつか用意されていますか？ JTA/XAの情報を探しているべきですか？

Answer 1

これらのキュー統合されたすべてのチャネルでは、メッセージはデフォルトでのみメモリに格納されます。永続性が必要な場合は、永続的なMessageStore実装を参照するために 'queue'要素内に 'message-store'属性を指定するか、永続的なMessageStore実装を参照するために永続的なブローカによってサポートされているものチャネルまたはチャネルアダプタを使用します。後者のオプションを使用すると、任意のJMSプロバイダのメッセージ永続性の実装を利用できます。

QueueChannelにメッセージストアを設定するには、メッセージストア属性を次のように追加します。

バグ・インテグレーションは、a）org.springframework.integration.store.MessageStore戦略インタフェースを定義し、b）このインタフェースのいくつかの実装を提供し、c）メッセージをバッファリングする機能を持つすべてのコンポーネントにメッセージストア属性を公開するMessageStoreインターフェイスを実装するインスタンスをインジェクトすることができます。

私の例ではJDBC Message Storeが使用されていますが、他にもいくつかのオプションがあります。

<bean id="myDataSource" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close"> 
    <property name="driverClassName" value="${my.jdbc.driver}"/> 
    <property name="url" value="${my.jdbc.url}"/> 
    <property name="username" value="${my.jdbc.username}"/> 
    <property name="password" value="${my.jdbc.password}"/> 
    <property name="minEvictableIdleTimeMillis" value="300000"/> 
    <property name="timeBetweenEvictionRunsMillis" value="60000"/> 
    <property name="connectionProperties" value="SetBigStringTryClob=true;"/> 
</bean> 

<!-- The poller is needed by any of the QueueChannels --> 
<integration:poller id="myPoller" fixed-rate="5000" default="true"/> 

<!-- The MessageStore is needed to persist messages used by any of the QueueChannels --> 
<int-jdbc:message-store id="myMessageStore" data-source="myDataSource" table-prefix="MY_INT_"/> 

<!-- Main entry point into the process --> 
<integration:gateway id="myGateway" 
        service-interface="com.mycompany.myproject.integration.gateways.myGateway" 
        default-request-channel="myGatewayChannel" 
     /> 

<!-- Map the initial input channel to the first step, MyFirstService --> 
<integration:channel id="myGatewayChannel"> 
    <integration:queue capacity="1000"/> 
    <integration:queue message-store="myMessageStore" capacity="1000"/> 
</integration:channel> 

<!-- Step 1: My First Service --> 
<integration:service-activator 
     id="myFirstServiceActivator" 
     input-channel="myGatewayChannel" 
     output-channel="myNextChannel" 
     ref="myFirstService" 
     method="process"/>  

<!-- LONG running process. Setup asynchronous queue channel. --> 
<integration:channel id="myNextChannel"> 
    <integration:queue capacity="1000"/> 
    <integration:queue message-store="myMessageStore" capacity="1000"/> 
</integration:channel>