1

非同期プロトコルで送信する必要があるメッセージを含む永続的なトランザクション キューがあります。各メッセージは独自のトランザクションで送信する必要がありますが、特定の時間に処理中のメッセージの数が原因でthread-per-messageを使用できず、スループット要件により中間状態が持続することができません。

のコードを見ると、トランザクション リソースを に格納するJmsTransactionManagerを使用していることがわかります。そのため、単一のスレッド内で複数のトランザクションをどうにかしてシェパードするために を実装する必要があるようです。これは少し極端なようです...TransactionSynchronizationManagerThreadLocalPlatformTransactionManager

この複雑さを解消する Spring Integration ユニットの配置はありますか? 私はむしろJTA/XAの情報を見るべきですか？

4

1 に答える 1

1

Spring Integration のこれらのキューベースのチャネルはすべて、デフォルトでメッセージをメモリ内にのみ保存します。永続性が必要な場合はmessage-store、' ' 要素内に ' ' 属性をqueue指定して永続的な MessageStore 実装を参照するか、ローカル チャネルを永続的なブローカー (JMS に基づくチャネルやチャネル アダプタ。後者のオプションを使用すると、任意の JMS プロバイダーの実装を利用してメッセージを永続化できます。

QueueChannel以下に示すように、message-store 属性を追加することで、任意のメッセージ ストアを構成できます。

org.springframework.integration.store.MessageStoreSpring Integration は、a) 戦略インターフェースを定義する、b) このインターフェースのいくつかの実装を提供する、および c) 注入できるようにメッセージをバッファリングする機能を持つすべてのコンポーネントで message-store 属性を公開することにより、Message Store パターンのサポートを提供します。 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>
于 2012-03-08T17:03:39.977 に答える