メッセージングエンドポイント
メッセージエンドポイント
この章の最初の部分では、いくつかの背景理論を取り上げ、Spring Integration のさまざまなメッセージングコンポーネントを駆動する基盤となる API についてかなり明らかにします。この情報は、バックグラウンドで何が起こっているのかを本当に理解したい場合に役立ちます。ただし、さまざまな要素の単純化された名前空間ベースの構成で起動して実行する場合は、当面はエンドポイント名前空間のサポートに進んでください。
概要で記述されていたように、メッセージエンドポイントは、さまざまなメッセージングコンポーネントをチャネルに接続するロールを果たします。次のいくつかの章では、メッセージを消費するさまざまなコンポーネントについて説明します。これらの一部は、返信メッセージを送信することもできます。メッセージの送信は非常に簡単です。メッセージチャンネルで前述したように、メッセージチャネルにメッセージを送信できます。ただし、受信はもう少し複雑です。主な理由は、ポーリングコンシューマー (英語) とイベント駆動型コンシューマー (英語) の 2 種類のコンシューマーがあることです。
2 つのうち、イベント駆動型のコンシューマーははるかに単純です。別のポーラースレッドを管理およびスケジュールする必要がない場合、これらは本質的にコールバックメソッドを持つリスナーです。Spring Integration のサブスクライブ可能なメッセージチャネルの 1 つに接続する場合、このシンプルなオプションは非常に効果的です。ただし、バッファリング可能なポーリング可能なメッセージチャネルに接続する場合、一部のコンポーネントはポーリングスレッドをスケジュールおよび管理する必要があります。Spring Integration は、これら 2 つの型のコンシューマーに対応するために、2 つの異なるエンドポイント実装を提供します。コンシューマー自体はコールバックインターフェースを実装するだけで済みます。ポーリングが必要な場合、エンドポイントはコンシューマーインスタンスのコンテナーとして機能します。利点は、メッセージ駆動型 Bean をホストするためにコンテナーを使用することと似ていますが、これらのコンシューマーは ApplicationContext 内で実行される Spring 管理のオブジェクトであるため、Spring 自身の MessageListener コンテナーにより似ています。
メッセージハンドラー
Spring Integration の MessageHandler インターフェースは、フレームワーク内の多くのコンポーネントによって実装されています。つまり、これはパブリック API の一部ではなく、通常、MessageHandler を直接実装することはありません。それにもかかわらず、消費されたメッセージを実際に処理するためにメッセージコンシューマーによって使用されるため、この戦略インターフェースを認識することは、コンシューマーの全体的なロールを理解する上で役立ちます。インターフェースは次のように定義されます。
public interface MessageHandler {
void handleMessage(Message<?> message);
}シンプルであるにもかかわらず、このインターフェースは、次の章で説明するほとんどのコンポーネント（ルーター、トランス、スプリッター、アグリゲーター、サービスアクティベーターなど）の基盤を提供します。これらのコンポーネントはそれぞれ、処理するメッセージに対して非常に異なる機能を実行しますが、実際にメッセージを受信するための要件は同じであり、ポーリングとイベント駆動型の動作の選択も同じです。Spring Integration は、これらのコールバックベースのハンドラーをホストし、メッセージチャンネルに接続できるようにする 2 つのエンドポイント実装を提供します。
イベント駆動型のコンシューマー
2 つの方が単純なので、最初にイベント駆動型のコンシューマーエンドポイントについて説明します。SubscribableChannel インターフェースが subscribe() メソッドを提供し、そのメソッドが MessageHandler パラメーターを受け入れることを思い出してください（SubscribableChannel に示すように）。次のリストは、subscribe メソッドの定義を示しています。
subscribableChannel.subscribe(messageHandler); チャネルにサブスクライブされたハンドラーはそのチャネルをアクティブにポーリングする必要がないため、これはイベント駆動型コンシューマーであり、Spring Integration によって提供される実装は、次の例に示すように SubscribableChannel および MessageHandler を受け入れます。
SubscribableChannel channel = context.getBean("subscribableChannel", SubscribableChannel.class);
EventDrivenConsumer consumer = new EventDrivenConsumer(channel, exampleHandler);ポーリングコンシューマー
Spring Integration は PollingConsumer も提供し、次の例に示すように、チャネルが PollableChannel を実装する必要があることを除いて、同じ方法でインスタンス化できます。
PollableChannel channel = context.getBean("pollableChannel", PollableChannel.class);
PollingConsumer consumer = new PollingConsumer(channel, exampleHandler);| ポーリングコンシューマーの詳細については、ポーラーおよびチャンネルアダプターを参照してください。 |
ポーリングコンシューマーには、他にも多くの構成オプションがあります。例: トリガーは必須プロパティです。次の例は、トリガーを設定する方法を示しています。
PollingConsumer consumer = new PollingConsumer(channel, handler);
consumer.setTrigger(new IntervalTrigger(30, TimeUnit.SECONDS));Spring Integration は現在、Trigger インターフェースの 2 つの実装、IntervalTrigger と CronTrigger を提供しています。IntervalTrigger は通常、単純な間隔（ミリ秒単位）で定義されますが、initialDelay プロパティとブール fixedRate プロパティもサポートします（デフォルトは false — つまり、固定遅延はありません）。次の例では、両方のプロパティを設定します。
IntervalTrigger trigger = new IntervalTrigger(1000);
trigger.setInitialDelay(5000);
trigger.setFixedRate(true);前の例の 3 つの設定の結果は、5 秒待機してから 1 秒ごとにトリガーするトリガーです。
CronTrigger には有効な cron 式が必要です。詳細については、Javadoc を参照してください。次の例では、新しい CronTrigger を設定します。
CronTrigger trigger = new CronTrigger("*/10 * * * * MON-FRI");前の例で定義されたトリガーの結果は、月曜日から金曜日までの 10 秒ごとにトリガーするトリガーです。
トリガーに加えて、maxMessagesPerPoll および receiveTimeout の 2 つの他のポーリング関連の構成プロパティを指定できます。次の例は、これら 2 つのプロパティを設定する方法を示しています。
PollingConsumer consumer = new PollingConsumer(channel, handler);
consumer.setMaxMessagesPerPoll(10);
consumer.setReceiveTimeout(5000);maxMessagesPerPoll プロパティは、特定のポーリング操作内で受信するメッセージの最大数を指定します。つまり、ポーラーは、null が返されるか最大値に達するまで、待機することなく receive() を呼び出し続けます。例: ポーラーに 10 秒の間隔トリガーと 25 の maxMessagesPerPoll 設定があり、キューに 100 個のメッセージがあるチャネルをポーリングしている場合、100 個すべてのメッセージを 40 秒以内に取得できます。25 を取得し、10 秒間待機し、次の 25 を取得します。
receiveTimeout プロパティは、受信操作を呼び出すときに使用可能なメッセージがない場合にポーラーが待機する時間を指定します。例: 表面上は似ているが実際にはまったく異なる 2 つのオプションを検討します。最初のオプションには 5 秒の間隔トリガーと 50 ミリ秒の受信タイムアウトがあり、2 番目のオプションには 50 ミリ秒の間隔トリガーと 5 の受信タイムアウトがあります。秒。最初のメッセージは、チャネルに到着してから最大 4950 ミリ秒後にメッセージを受信する場合があります（ポーリングコールの 1 つが返された直後にメッセージが到着した場合）。一方、2 番目の構成では、メッセージが 50 ミリ秒を超えることはありません。違いは、2 番目のオプションでは待機するスレッドが必要なことです。ただし、その結果、到着したメッセージにはるかに迅速に応答できます。「ロングポーリング」と呼ばれるこの手法は、ポーリングされたソースでイベント駆動型の動作をエミュレートするために使用できます。
次の例に示すように、ポーリングコンシューマーは Spring TaskExecutor に委譲することもできます。
PollingConsumer consumer = new PollingConsumer(channel, handler);
TaskExecutor taskExecutor = context.getBean("exampleExecutor", TaskExecutor.class);
consumer.setTaskExecutor(taskExecutor); さらに、PollingConsumer には adviceChain というプロパティがあります。このプロパティを使用すると、トランザクションを含む追加の横断的関心事を処理するための AOP アドバイスの List を指定できます。これらのアドバイスは、doPoll() メソッドに適用されます。詳細については、AOP アドバイスチェーンおよびエンドポイント名前空間のサポートのトランザクションサポートに関するセクションを参照してください。
前の例は、依存関係のルックアップを示しています。ただし、これらのコンシューマーはほとんどの場合 Spring Bean 定義として構成されていることに注意してください。実際、Spring Integration は、チャネルの型に基づいて適切なコンシューマー型を作成する ConsumerEndpointFactoryBean と呼ばれる FactoryBean も提供します。また、Spring Integration はこれらの詳細をさらに隠すために完全な XML 名前空間をサポートしています。このガイドでは、名前空間ベースの構成が各コンポーネント型の導入に合わせて取り上げられています。
MessageHandler 実装の多くは、応答メッセージを生成できます。前述のように、メッセージの送信はメッセージの受信と比較すると簡単です。それでも、送信される応答メッセージのタイミングと数は、ハンドラーの型によって異なります。例: アグリゲーターは、多数のメッセージの到着を待機し、多くの場合、スプリッターのダウンストリームコンシューマーとして構成されます。スプリッターは、処理する各メッセージに対して複数の応答を生成できます。名前空間の構成を使用する場合、すべての詳細を厳密に知る必要はありません。ただし、これらのコンポーネントのいくつかが共通の基本クラスである AbstractReplyProducingMessageHandler を共有し、setOutputChannel(..) メソッドを提供することを知っておく価値があるかもしれません。 |
エンドポイント名前空間のサポート
このリファレンスマニュアル全体で、ルーター、トランスフォーマー、サービスアクティベーターなどのエンドポイント要素の特定の構成例を見つけることができます。これらのほとんどは input-channel 属性をサポートし、多くは output-channel 属性をサポートします。解析後、これらのエンドポイント要素は、参照される input-channel の型に応じて、それぞれ PollableChannel または SubscribableChannel の PollingConsumer または EventDrivenConsumer のいずれかのインスタンスを生成します。チャネルがポーリング可能な場合、ポーリングの動作はエンドポイント要素の poller サブ要素とその属性に基づいています。
以下のリストは、poller で使用可能なすべての構成オプションをリストしています。
<int:poller cron="" (1)
default="false" (2)
error-channel="" (3)
fixed-delay="" (4)
fixed-rate="" (5)
id="" (6)
max-messages-per-poll="" (7)
receive-timeout="" (8)
ref="" (9)
task-executor="" (10)
time-unit="MILLISECONDS" (11)
trigger=""> (12)
<int:advice-chain /> (13)
<int:transactional /> (14)
</int:poller>| 1 | Cron 式を使用してポーラーを構成する機能を提供します。基礎となる実装は org.springframework.scheduling.support.CronTrigger を使用します。この属性が設定されている場合、次の属性のいずれも指定する必要はありません: fixed-delay、trigger、fixed-rate、ref。 |
| 2 | この属性を true に設定することにより、グローバルデフォルトポーラーを 1 つだけ定義できます。アプリケーションコンテキストで複数のデフォルトポーラーが定義されている場合、例外が発生します。PollableChannel （PollingConsumer）または明示的に構成されたポーラーを持たない SourcePollingChannelAdapter に接続されたエンドポイントは、グローバルデフォルトポーラーを使用します。デフォルトは false です。オプション。 |
| 3 | このポーラーの呼び出しで障害が発生した場合にエラーメッセージが送信されるチャネルを識別します。例外を完全に抑制するには、nullChannel への参照を提供できます。オプション。 |
| 4 | 固定遅延トリガーは、内部で PeriodicTrigger を使用します。time-unit 属性を使用しない場合、指定された値はミリ秒単位で表されます。この属性が設定されている場合、次の属性のいずれも指定する必要はありません: fixed-rate、trigger、cron、ref。 |
| 5 | 固定レートトリガーは、内部で PeriodicTrigger を使用します。time-unit 属性を使用しない場合、指定された値はミリ秒単位で表されます。この属性が設定されている場合、次の属性のいずれも指定する必要はありません: fixed-delay、trigger、cron、ref。 |
| 6 | 型 org.springframework.integration.scheduling.PollerMetadata の、ポーラーの基礎となる Bean 定義を参照する ID。id 属性は、デフォルトのポーラー（default="true"）でない限り、最上位のポーラー要素に必要です。 |
| 7 | 詳細については、受信チャネルアダプターの構成を参照してください。指定しない場合、デフォルト値はコンテキストによって異なります。PollingConsumer を使用する場合、この属性はデフォルトで -1 になります。ただし、SourcePollingChannelAdapter を使用する場合、max-messages-per-poll 属性はデフォルトで 1 になります。オプション。 |
| 8 | 基になるクラス PollerMetadata に値が設定されます。指定しない場合、デフォルトは 1000（ミリ秒）です。オプション。 |
| 9 | 別のトップレベルのポーラーへの Bean 参照。ref 属性は、最上位の poller 要素に存在してはなりません。ただし、この属性が設定されている場合は、次の属性を指定する必要はありません: fixed-rate、trigger、cron、fixed-delay。 |
| 10 | カスタムタスクエグゼキューターを参照する機能を提供します。詳細については、TaskExecutor サポートを参照してください。オプション。 |
| 11 | この属性は、基になる org.springframework.scheduling.support.PeriodicTrigger の java.util.concurrent.TimeUnit 列挙値を指定します。この属性は fixed-delay または fixed-rate 属性と組み合わせてのみ使用できます。cron または trigger 参照属性のいずれかと組み合わせると、エラーが発生します。PeriodicTrigger でサポートされる最小の粒度はミリ秒です。使用可能なオプションはミリ秒と秒のみです。この値が提供されない場合、fixed-delay または fixed-rate 値はミリ秒として解釈されます。基本的に、この列挙型は秒ベースの間隔トリガー値に便利です。時間ごと、日ごと、月ごとの設定では、代わりに cron トリガーを使用することをお勧めします。 |
| 12 | org.springframework.scheduling.Trigger インターフェースを実装する Spring 構成の Bean への参照。ただし、この属性が設定されている場合は、次の属性を指定する必要はありません: fixed-delay、fixed-rate、cron、ref。オプション。 |
| 13 | 追加の横断的関心事を処理するための追加の AOP アドバイスを指定できます。詳細については、トランザクションサポートを参照してください。オプション。 |
| 14 | ポーラーはトランザクション対応にすることができます。詳細については、AOP アドバイスチェーンを参照してください。オプション。 |
サンプル
1 秒間隔の単純な間隔ベースのポーラーは、次のように構成できます。
<int:transformer input-channel="pollable"
ref="transformer"
output-channel="output">
<int:poller fixed-rate="1000"/>
</int:transformer>fixed-rate 属性を使用する代わりに、fixed-delay 属性を使用することもできます。
Cron 式に基づくポーラーの場合、次の例に示すように、代わりに cron 属性を使用します。
<int:transformer input-channel="pollable"
ref="transformer"
output-channel="output">
<int:poller cron="*/10 * * * * MON-FRI"/>
</int:transformer> 入力チャネルが PollableChannel である場合、ポーラー構成が必要です。具体的には、前述のように、trigger は PollingConsumer クラスの必須プロパティです。ポーリングコンシューマーエンドポイントの構成の poller サブ要素を省略すると、例外がスローされる場合があります。また、ポーリング不可能なチャネルに接続されている要素でポーラーを構成しようとすると、例外がスローされる場合があります。
トップレベルのポーラーを作成することもできます。この場合、次の例に示すように、ref 属性のみが必要です。
<int:poller id="weekdayPoller" cron="*/10 * * * * MON-FRI"/>
<int:transformer input-channel="pollable"
ref="transformer"
output-channel="output">
<int:poller ref="weekdayPoller"/>
</int:transformer>ref 属性は、内部ポーラー定義でのみ許可されます。最上位のポーラーでこの属性を定義すると、アプリケーションコンテキストの初期化中に構成例外がスローされます。 |
グローバルデフォルトポーラー
設定をさらに簡素化するために、グローバルデフォルトポーラーを定義できます。ApplicationContext 内の単一のトップレベルポーラーでは、default 属性が true に設定されている場合があります。その場合、同じ ApplicationContext 内で定義され、明示的に構成された poller サブ要素を持たない、その入力チャネルの PollableChannel を持つエンドポイントは、そのデフォルトを使用します。次の例は、このようなポーラーとそれを使用するトランスフォーマーを示しています。
<int:poller id="defaultPoller" default="true" max-messages-per-poll="5" fixed-rate="3000"/>
<!-- No <poller/> sub-element is necessary, because there is a default -->
<int:transformer input-channel="pollable"
ref="transformer"
output-channel="output"/>トランザクションサポート
Spring Integration は、ポーラーのトランザクションサポートも提供するため、各送受信操作をアトミックな作業単位として実行できます。ポーラーのトランザクションを構成するには、<transactional/> サブ要素を追加します。次の例は、使用可能な属性を示しています。
<int:poller fixed-delay="1000">
<int:transactional transaction-manager="txManager"
propagation="REQUIRED"
isolation="REPEATABLE_READ"
timeout="10000"
read-only="false"/>
</int:poller>詳しくは、ポーラートランザクションサポートを参照してください。
AOP アドバイスチェーン
Spring トランザクションサポートは、ポーラーによって開始されたメッセージフローのトランザクション動作を処理する TransactionInterceptor （AOP アドバイス）を使用したプロキシメカニズムに依存するため、ポーラーに関連する他のクロスカット動作を処理するために追加のアドバイスを提供する必要がある場合があります。そのために、poller は、MethodInterceptor インターフェースを実装するクラスにさらにアドバイスを追加できる advice-chain 要素を定義します。次の例は、poller の advice-chain を定義する方法を示しています。
<int:service-activator id="advicedSa" input-channel="goodInputWithAdvice" ref="testBean"
method="good" output-channel="output">
<int:poller max-messages-per-poll="1" fixed-rate="10000">
<int:advice-chain>
<ref bean="adviceA" />
<beans:bean class="org.something.SampleAdvice" />
<ref bean="txAdvice" />
</int:advice-chain>
</int:poller>
</int:service-activator>MethodInterceptor インターフェースの実装方法の詳細については、Spring Framework リファレンスガイドの AOP セクションを参照してください。アドバイスチェーンは、トランザクション構成を持たないポーラーにも適用でき、ポーラーによって開始されたメッセージフローの動作を強化できます。
アドバイスチェーンを使用する場合、<transactional/> 子要素は指定できません。代わりに、<tx:advice/> Bean を宣言し、<advice-chain/> に追加します。構成の詳細については、ポーラートランザクションサポートを参照してください。 |
TaskExecutor サポート
ポーリングスレッドは、Spring の TaskExecutor 抽象化の任意のインスタンスによって実行できます。これにより、エンドポイントまたはエンドポイントのグループの同時実行が可能になります。Spring 3.0 以降、コア Spring Framework には task 名前空間があり、その <executor/> 要素は単純なスレッドプールエグゼキューターの作成をサポートしています。その要素は、プールサイズやキュー容量などの一般的な同時実行設定の属性を受け入れます。スレッドプーリングエグゼキューターを構成すると、負荷がかかった状態でのエンドポイントのパフォーマンスに大きな違いが生じます。エンドポイントのパフォーマンスは考慮すべき主要な要素の 1 つであるため、これらの設定は各エンドポイントで使用できます（他の主要な要素は、エンドポイントがサブスクライブするチャネルの予想ボリュームです）。XML 名前空間のサポートを使用して構成されたポーリングエンドポイントの同時実行を有効にするには、<poller/> 要素で task-executor 参照を提供し、次の例に示す 1 つ以上のプロパティを提供します。
<int:poller task-executor="pool" fixed-rate="1000"/>
<task:executor id="pool"
pool-size="5-25"
queue-capacity="20"
keep-alive="120"/> タスクエグゼキューターを提供しない場合、コンシューマーのハンドラーは呼び出し元のスレッドで呼び出されます。通常、呼び出し元はデフォルトの TaskScheduler です（タスクスケジューラの構成を参照）。また、task-executor 属性は、Bean 名を指定することにより、Spring の TaskExecutor インターフェースの実装への参照を提供できることに留意してください。前に示した executor 要素は、便宜上提供されています。
コンシューマーのポーリングの背景セクションで前述したように、イベント駆動型の動作をエミュレートするような方法でポーリングコンシューマーを構成することもできます。長い receive-timeout と短い interval-trigger を使用すると、ポーリングされたメッセージソースであっても、受信メッセージに対する非常にタイムリーな反応を保証できます。これは、タイムアウト付きのブロッキング待機呼び出しがあるソースにのみ適用されることに注意してください。例: ファイルポーラーはブロックしません。各 receive() 呼び出しはすぐに戻り、新しいファイルが含まれているか含まれていません。ポーラーに長い receive-timeout が含まれていても、そのようなシナリオではその値は使用されません。一方、Spring Integration 独自のキューベースのチャネルを使用する場合、タイムアウト値に参加する機会があります。次の例は、ポーリングコンシューマーがほぼ瞬時にメッセージを受信する方法を示しています。
<int:service-activator input-channel="someQueueChannel"
output-channel="output">
<int:poller receive-timeout="30000" fixed-rate="10"/>
</int:service-activator>このアプローチを使用しても、オーバーヘッドはあまり多くありません。内部的には、待機スレッドに過ぎず、スラッシング、無限 while ループなどの CPU リソース使用量をほとんど必要としません。
実行時のポーリングレートの変更
fixed-delay または fixed-rate 属性でポーラーを構成する場合、デフォルトの実装は PeriodicTrigger インスタンスを使用します。PeriodicTrigger は、コア Spring Framework の一部です。コンストラクター引数としてのみ間隔を受け入れます。実行時に変更することはできません。
ただし、org.springframework.scheduling.Trigger インターフェースの独自の実装を定義できます。PeriodicTrigger を出発点として使用することもできます。次に、間隔（期間）に setter を追加するか、トリガー自体に独自の調整ロジックを埋め込むこともできます。period プロパティは、nextExecutionTime への各呼び出しで使用され、次のポーリングをスケジュールします。ポーラー内でこのカスタムトリガーを使用するには、アプリケーションコンテキストでカスタムトリガーの Bean 定義を宣言し、カスタムトリガー Bean インスタンスを参照する trigger 属性を使用して、ポーラー構成に依存関係を注入します。これで、トリガー Bean への参照を取得し、ポーリング間のポーリング間隔を変更できます。
例については、Spring Integration サンプル [GitHub] (英語) プロジェクトを参照してください。dynamic-poller というサンプルが含まれています。これは、カスタムトリガーを使用し、実行時にポーリング間隔を変更する機能を示しています。
このサンプルは、org.springframework.scheduling.Trigger (Javadoc) インターフェースを実装するカスタムトリガーを提供します。サンプルのトリガーは、Spring の PeriodicTrigger (Javadoc) 実装に基づいています。ただし、カスタムトリガーのフィールドは最終的なものではなく、プロパティには明示的な getter と setter があり、実行時にポーリング期間を動的に変更できます。
ただし、Trigger メソッドは nextExecutionTime() であるため、動的トリガーへの変更は、既存の構成に基づいて次のポーリングまで有効にならないことに注意することが重要です。現在設定されている次の実行時間の前にトリガーを強制的に起動することはできません。 |
ペイロード型変換
このリファレンスマニュアル全体を通して、メッセージまたは任意の Object を入力パラメーターとして受け入れるさまざまなエンドポイントの特定の構成および実装例を確認することもできます。Object の場合、このようなパラメーターは、メッセージペイロードまたはペイロードまたはヘッダーの一部にマップされます（Spring 式言語を使用する場合）。ただし、エンドポイントメソッドの入力パラメーターの型がペイロードまたはその部分の型と一致しない場合があります。このシナリオでは、型変換を実行する必要があります。Spring Integration は、integrationConversionService という名前の変換サービス Bean の独自のインスタンス内に（Spring ConversionService を使用して）型コンバーターを登録するための便利な方法を提供します。その Bean は、Spring Integration インフラストラクチャーを使用して最初のコンバーターが定義されるとすぐに自動的に作成されます。コンバーターを登録するには、org.springframework.core.convert.converter.Converter、org.springframework.core.convert.converter.GenericConverter、org.springframework.core.convert.converter.ConverterFactory を実装できます。
Converter 実装は最も単純で、単一の型から別の型に変換します。クラス階層への変換など、より高度な処理を行うには、GenericConverter および場合によっては ConditionalConverter を実装できます。これらにより、from および to 型記述子への完全なアクセスが可能になり、複雑な変換が可能になります。例: 変換のターゲットである Something という抽象クラス（パラメーター型、チャネルデータ型など）がある場合、Thing1 および Thing という 2 つの具体的な実装があり、どちらかに変換したい入力型に基づいて、GenericConverter が適しています。詳細については、これらのインターフェースの Javadoc を参照してください。
コンバーターを実装したら、次の例に示すように、便利な名前空間サポートでコンバーターを登録できます。
<int:converter ref="sampleConverter"/>
<bean id="sampleConverter" class="foo.bar.TestConverter"/>または、次の例に示すように、内部 Bean を使用できます。
<int:converter>
<bean class="o.s.i.config.xml.ConverterParserTests$TestConverter3"/>
</int:converter>Spring Integration 4.0 以降、次の例に示すように、アノテーションを使用して上記の構成を作成できます。
@Component
@IntegrationConverter
public class TestConverter implements Converter<Boolean, Number> {
public Number convert(Boolean source) {
return source ? 1 : 0;
}
} または、次の例に示すように、@Configuration アノテーションを使用できます。
@Configuration
@EnableIntegration
public class ContextConfiguration {
@Bean
@IntegrationConverter
public SerializingConverter serializingConverter() {
return new SerializingConverter();
}
} アプリケーションコンテキストを設定する場合、Spring Framework では 対照的に、 ただし、Spring この場合、 |
コンテンツ型の変換
バージョン 5.0 以降、デフォルトでは、メソッド呼び出しメカニズムは org.springframework.messaging.handler.invocation.InvocableHandlerMethod インフラストラクチャに基づいています。HandlerMethodArgumentResolver 実装（PayloadArgumentResolver や MessageMethodArgumentResolver など）は、MessageConverter 抽象化を使用して、受信 payload をターゲットメソッドの引数型に変換できます。変換は、contentType メッセージヘッダーに基づくことができます。この目的のために、Spring Integration は ConfigurableCompositeMessageConverter を提供します。ConfigurableCompositeMessageConverter は、登録されたコンバーターのリストに委譲され、そのうちの 1 つが非 null の結果を返すまで呼び出されます。デフォルトでは、このコンバーターは以下を提供します（厳密な順序で）:
変換の目的と適切な contentType 値の詳細については、Javadoc（前のリストにリンクされています）を参照してください。ConfigurableCompositeMessageConverter は、他の MessageConverter 実装（前述のデフォルトコンバーターを含むか除外する）で提供できるため、使用されます。また、次の例に示すように、アプリケーションコンテキストで適切な Bean として登録し、デフォルトコンバーターをオーバーライドすることもできます。
@Bean(name = IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME)
public ConfigurableCompositeMessageConverter compositeMessageConverter() {
List<MessageConverter> converters =
Arrays.asList(new MarshallingMessageConverter(jaxb2Marshaller()),
new JavaSerializationMessageConverter());
return new ConfigurableCompositeMessageConverter(converters);
} これらの 2 つの新しいコンバーターは、デフォルトの前にコンポジットに登録されます。ConfigurableCompositeMessageConverter を使用することはできませんが、integrationArgumentResolverMessageConverter という名前で Bean を登録することにより（IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME プロパティを設定することにより）独自の MessageConverter を提供することもできます。
SpZZ メソッド呼び出しを使用する場合、MessageConverter ベース（contentType ヘッダーを含む）変換は使用できません。この場合、上記のペイロード型変換での通常のクラスからクラスへの変換のみが使用可能です。 |
非同期ポーリング
ポーリングを非同期にしたい場合、ポーラーはオプションで、任意の TaskExecutor Bean の既存のインスタンスを指す task-executor 属性を指定できます（Spring 3.0 は、task 名前空間を通じて便利な名前空間構成を提供します）。ただし、TaskExecutor を使用してポーラーを構成する際に理解しておく必要がある特定の事項があります。
問題は、ポーラーと TaskExecutor の 2 つの構成があることです。それらは互いに調和している必要があります。そうしないと、人為的なメモリリークが発生する可能性があります。
次の構成を検討してください。
<int:channel id="publishChannel">
<int:queue />
</int:channel>
<int:service-activator input-channel="publishChannel" ref="myService">
<int:poller receive-timeout="5000" task-executor="taskExecutor" fixed-rate="50" />
</int:service-activator>
<task:executor id="taskExecutor" pool-size="20" />上記の構成は、不一致の構成を示しています。
デフォルトでは、タスクエグゼキューターには無制限のタスクキューがあります。ポーラーは、すべてのスレッドがブロックされていても新しいタスクのスケジューリングを続け、新しいメッセージの到着またはタイムアウトの期限切れを待ちます。5 秒のタイムアウトでタスクを実行する 20 のスレッドがあるとすると、それらは 1 秒あたり 4 の割合で実行されます。ただし、新しいタスクは 1 秒あたり 20 のレートでスケジュールされているため、タスクエグゼキューターの内部キューは 1 秒あたり 16 のレートで増加するため（プロセスがアイドル状態のとき）、メモリリークが発生します。
これを処理する方法の 1 つは、タスクエグゼキューターの queue-capacity 属性を設定することです。0 でも妥当な値です。また、Task Executor の rejection-policy 属性を設定して（たとえば DISCARD に）キューに入れられないメッセージの処理を指定することで管理できます。つまり、TaskExecutor を構成する際に理解する必要がある特定の詳細があります。このテーマの詳細については、Spring リファレンスマニュアルの “タスクの実行とスケジューリング” を参照してください。
エンドポイント内部 Bean
多くのエンドポイントは複合 Bean です。これには、すべてのコンシューマーと、ポーリングされたすべての受信チャネルアダプターが含まれます。コンシューマー (ポーリングまたはイベント駆動) は、MessageHandler に委譲します。ポーリングされたアダプターは、MessageSource に委譲することでメッセージを取得します。多くの場合、実行時またはテスト時に構成を変更する場合など、委譲 Bean への参照を取得すると便利です。これらの Bean は、よく知られている名前で ApplicationContext から取得できます。MessageHandler インスタンスは、someConsumer.handler に似た Bean ID でアプリケーションコンテキストに登録されます (「コンシューマー」はエンドポイントの id 属性の値です)。MessageSource インスタンスは、somePolledAdapter.source に似た Bean ID で登録されます ( "somePolledAdapter" はアダプターの ID です)。
上記は、フレームワークコンポーネント自体にのみ適用されます。次の例に示すように、代わりに内部 Bean 定義を使用できます。
<int:service-activator id="exampleServiceActivator" input-channel="inChannel"
output-channel = "outChannel" method="foo">
<beans:bean class="org.foo.ExampleServiceActivator"/>
</int:service-activator>Bean は、宣言された内部 Bean のように扱われ、アプリケーションコンテキストに登録されません。他の方法でこの Bean にアクセスする場合は、id を使用してトップレベルで宣言し、代わりに ref 属性を使用します。詳細については、Spring ドキュメントを参照してください。
エンドポイントのロール
バージョン 4.2 から、エンドポイントをロールに割り当てることができます。ロールを使用すると、エンドポイントをグループとして開始および停止できます。これは、リーダーシップの選択を使用する場合に特に役立ちます。リーダーシップの選択では、リーダーシップが許可または取り消されたときにエンドポイントのセットをそれぞれ開始または停止できます。このために、フレームワークは SmartLifecycleRoleController Bean をアプリケーションコンテキストに IntegrationContextUtils.INTEGRATION_LIFECYCLE_ROLE_CONTROLLER という名前で登録します。ライフサイクルを制御する必要があるときはいつでも、この Bean を挿入するか @Autowired できます:
<bean class="com.some.project.SomeLifecycleControl">
<property name="roleController" ref="integrationLifecycleRoleController"/>
</bean>XML、Java 構成、プログラムを使用して、エンドポイントをロールに割り当てることができます。次の例は、XML を使用してエンドポイントロールを設定する方法を示しています。
<int:inbound-channel-adapter id="ica" channel="someChannel" expression="'foo'" role="cluster"
auto-startup="false">
<int:poller fixed-rate="60000" />
</int:inbound-channel-adapter>次の例は、Java で作成された Bean のエンドポイントロールを構成する方法を示しています。
@Bean
@ServiceActivator(inputChannel = "sendAsyncChannel", autoStartup="false")
@Role("cluster")
public MessageHandler sendAsyncHandler() {
return // some MessageHandler
}次の例は、Java のメソッドでエンドポイントロールを構成する方法を示しています。
@Payload("#args[0].toLowerCase()")
@Role("cluster")
public String handle(String payload) {
return payload.toUpperCase();
} 次の例は、Java で SmartLifecycleRoleController を使用してエンドポイントロールを設定する方法を示しています。
@Autowired
private SmartLifecycleRoleController roleController;
...
this.roleController.addSmartLifeCycleToRole("cluster", someEndpoint);
... 次の例は、Java で IntegrationFlow を使用してエンドポイントロールを設定する方法を示しています。
IntegrationFlow flow -> flow
.handle(..., e -> e.role("cluster")); これらはそれぞれ、エンドポイントを cluster ロールに追加します。
roleController.startLifecyclesInRole("cluster") および対応する stop… メソッドを呼び出すと、エンドポイントが開始および停止します。
SmartLifecycle を実装するオブジェクトは、エンドポイントだけでなく、プログラムで追加できます。 |
SmartLifecycleRoleController は ApplicationListener<AbstractLeaderEvent> を実装し、リーダーシップが許可または取り消された場合（一部の Bean がそれぞれ OnGrantedEvent または OnRevokedEvent を発行した場合）、構成された SmartLifecycle オブジェクトを自動的に開始および停止します。
リーダーシップの選択を使用してコンポーネントを起動および停止する場合、auto-startup XML 属性（autoStartup Bean プロパティ）を false に設定して、コンテキストの初期化中にアプリケーションコンテキストがコンポーネントを起動しないようにすることが重要です。 |
バージョン 4.3.8 以降、SmartLifecycleRoleController はいくつかのステータスメソッドを提供します。
public Collection<String> getRoles() (1)
public boolean allEndpointsRunning(String role) (2)
public boolean noEndpointsRunning(String role) (3)
public Map<String, Boolean> getEndpointsRunningStatus(String role) (4)| 1 | 管理されているロールのリストを返します。 |
| 2 | ロール内のすべてのエンドポイントが実行されている場合、true を返します。 |
| 3 | ロール内のエンドポイントがどれも実行されていない場合、true を返します。 |
| 4 | component name : running status のマップを返します。コンポーネント名は通常、Bean 名です。 |
リーダーシップイベントの取り扱い
エンドポイントのグループは、それぞれリーダーシップの付与または取り消しに基づいて開始および停止できます。これは、共有リソースを単一のインスタンスのみで使用する必要があるクラスター化されたシナリオで役立ちます。この例は、共有ディレクトリをポーリングしているファイル受信チャネルアダプターです。（ファイルを読むを参照）。
リーダーの選出に参加し、リーダーが選ばれたとき、リーダーシップが取り消されたとき、リーダーになるためのリソースを獲得できなかったときに通知を受けるために、アプリケーションは「リーダーイニシエーター」と呼ばれるアプリケーションコンテキストにコンポーネントを作成します。通常、リーダーイニシエーターは SmartLifecycle であるため、コンテキストの開始時に（オプションで）開始し、リーダーシップが変化すると通知を発行します。障害が発生した場合に特定のアクションを実行したい場合は、publishFailedEvents を true （バージョン 5.0 以降）に設定することにより、障害通知を受け取ることもできます。慣例により、コールバックを受け取る Candidate を提供する必要があります。また、フレームワークによって提供される Context オブジェクトを介してリーダーシップを取り消すこともできます。コードは、o.s.i.leader.event.AbstractLeaderEvent インスタンス（OnGrantedEvent および OnRevokedEvent のスーパークラス）をリッスンし、それに応じて応答することもできます（たとえば、SmartLifecycleRoleController を使用して）。イベントには、Context オブジェクトへの参照が含まれています。次のリストは、Context インターフェースの定義を示しています。
public interface Context {
boolean isLeader();
void yield();
String getRole();
}バージョン 5.0.6 以降、コンテキストは候補者のロールへの参照を提供します。
Spring Integration は、LockRegistry 抽象化に基づいたリーダーイニシエーターの基本的な実装を提供します。これを使用するには、次の例に示すように、Bean としてインスタンスを作成する必要があります。
@Bean
public LockRegistryLeaderInitiator leaderInitiator(LockRegistry locks) {
return new LockRegistryLeaderInitiator(locks);
} ロックレジストリが正しく実装されている場合、リーダーは 1 人だけです。ロックレジストリが、有効期限が切れたり壊れたときに例外（理想的には InterruptedException）をスローするロックも提供する場合、リーダーレス期間の期間は、ロック実装の固有の待機時間で許される限り短くすることができます。デフォルトでは、busyWaitMillis プロパティは、ロックが不完全であり、ロックを再度取得しようとすると期限切れになることがわかっている（より一般的な）場合に CPU の枯渇を防ぐために、レイテンシを追加します。
Zookeeper を使用するリーダーの選出とイベントの詳細については、"Zookeeper リーダーシップイベントの取り扱い" を参照してください。
メッセージングゲートウェイ
ゲートウェイは、Spring Integration によって提供されるメッセージング API を隠します。これにより、アプリケーションのビジネスロジックは Spring Integration API を認識できなくなります。汎用ゲートウェイを使用することにより、コードは単純なインターフェースのみと対話します。
GatewayProxyFactoryBean を入力してください
前述のように、Spring Integration API（ゲートウェイクラスを含む）に依存しないことは素晴らしいことです。そのため、Spring Integration は GatewayProxyFactoryBean を提供します。GatewayProxyFactoryBean は、任意のインターフェースのプロキシを生成し、以下に示すゲートウェイメソッドを内部的に呼び出します。依存性注入を使用することにより、ビジネスメソッドにインターフェースを公開できます。
次の例は、Spring Integration との対話に使用できるインターフェースを示しています。
package org.cafeteria;
public interface Cafe {
void placeOrder(Order order);
}ゲートウェイ XML 名前空間のサポート
名前空間のサポートも提供されます。次の例に示すように、インターフェースをサービスとして設定できます。
<int:gateway id="cafeService"
service-interface="org.cafeteria.Cafe"
default-request-channel="requestChannel"
default-reply-timeout="10000"
default-reply-channel="replyChannel"/> この構成を定義すると、cafeService を他の Bean に注入できるようになり、Cafe インターフェースのプロキシされたインスタンスのメソッドを呼び出すコードは Spring Integration API を認識しなくなります。一般的なアプローチは、Spring Remoting（RMI、HttpInvoker など）のアプローチに似ています。（Cafe デモで） gateway 要素を使用する例については、“サンプル” 付録を参照してください。
上記の構成のデフォルトは、ゲートウェイインターフェースのすべてのメソッドに適用されます。応答タイムアウトが指定されていない場合、呼び出しスレッドは応答を無期限に待機します。レスポンスがない場合のゲートウェイの動作を参照してください。
デフォルトは、個々のメソッドに対してオーバーライドできます。アノテーションと XML を使用したゲートウェイ構成を参照してください。
デフォルトの応答チャネルの設定
通常、ゲートウェイは応答をリッスンする一時的な匿名応答チャネルを自動作成するため、default-reply-channel を指定する必要はありません。ただし、default-reply-channel （または HTTP、JMS などのアダプターゲートウェイを備えた reply-channel）を定義するように求められる場合があります。
背景については、ゲートウェイの内部動作について簡単に説明します。ゲートウェイは、一時的なポイントツーポイント応答チャネルを作成します。これは匿名であり、名前 replyChannel でメッセージヘッダーに追加されます。明示的な default-reply-channel （リモートアダプターゲートウェイを備えた reply-channel）を提供する場合、パブリッシュ / サブスクライブチャネルを指すことができます。このチャネルは、複数のサブスクライバを追加できるため、その名前が付けられています。内部的に、Spring Integration は一時的な replyChannel と明示的に定義された default-reply-channel の間にブリッジを作成します。
返信をゲートウェイだけでなく、他のコンシューマーにも送信するとします。この場合、次の 2 つが必要です。
サブスクライブできる名前付きチャンネル
パブリッシュ / サブスクライブチャネルになるチャネル
ヘッダーに追加された応答チャネルは匿名でポイントツーポイントであるため、ゲートウェイで使用されるデフォルトの戦略はこれらのニーズを満たしません。これは、他のサブスクライバーがそれへのハンドルを取得できないことを意味し、たとえそれが可能であっても、チャネルは 1 人のサブスクライバーのみがメッセージを取得するようなポイントツーポイントの動作を持ちます。default-reply-channel を定義することにより、選択したチャンネルを指すことができます。この場合、publish-subscribe-channel です。ゲートウェイは、そこから、ヘッダーに格納されている一時的な匿名応答チャネルへのブリッジを作成します。
また、インターセプター（盗聴など）を介してモニターまたは監査するための応答チャネルを明示的に提供することもできます。チャンネルインターセプターを設定するには、名前付きチャンネルが必要です。
アノテーションと XML を使用したゲートウェイ構成
@Gateway アノテーションを追加することにより、前の Cafe インターフェースの例を拡張する次の例を検討してください。
public interface Cafe {
@Gateway(requestChannel="orders")
void placeOrder(Order order);
}@Header アノテーションを使用すると、次の例に示すように、メッセージヘッダーとして解釈される値を追加できます。
public interface FileWriter {
@Gateway(requestChannel="filesOut")
void write(byte[] content, @Header(FileHeaders.FILENAME) String filename);
} ゲートウェイメソッドの構成に XML アプローチを好む場合、次の例に示すように、method 要素をゲートウェイ構成に追加できます。
<int:gateway id="myGateway" service-interface="org.foo.bar.TestGateway"
default-request-channel="inputC">
<int:default-header name="calledMethod" expression="#gatewayMethod.name"/>
<int:method name="echo" request-channel="inputA" reply-timeout="2" request-timeout="200"/>
<int:method name="echoUpperCase" request-channel="inputB"/>
<int:method name="echoViaDefault"/>
</int:gateway>XML を使用して、メソッド呼び出しごとに個別のヘッダーを提供することもできます。これは、設定するヘッダーが本質的に静的であり、@Header アノテーションを使用してゲートウェイのメソッドシグネチャーに埋め込みたくない場合に役立ちます。例: ローンブローカーの例では、開始されたリクエストの型（単一見積もりまたはすべての見積もり）に基づいて、ローン見積もりの集計がどのように行われるかに影響を与えたいと考えています。どのゲートウェイメソッドが呼び出されたかを評価することによってリクエストの型を決定することは、可能ではありますが、関心の分離パラダイムに違反します（メソッドは Java アーティファクトです）。ただし、メッセージヘッダーで意図（メタ情報）を表現することは、メッセージングアーキテクチャでは自然なことです。次の例は、2 つの方法のそれぞれに異なるメッセージヘッダーを追加する方法を示しています。
<int:gateway id="loanBrokerGateway"
service-interface="org.springframework.integration.loanbroker.LoanBrokerGateway">
<int:method name="getLoanQuote" request-channel="loanBrokerPreProcessingChannel">
<int:header name="RESPONSE_TYPE" value="BEST"/>
</int:method>
<int:method name="getAllLoanQuotes" request-channel="loanBrokerPreProcessingChannel">
<int:header name="RESPONSE_TYPE" value="ALL"/>
</int:method>
</int:gateway>前の例では、ゲートウェイのメソッドに基づいて、"RESPONSE_TYPE" ヘッダーに異なる値が設定されています。
式と「グローバル」ヘッダー
<header/> 要素は、value の代替として expression をサポートします。SpEL 式が評価され、ヘッダーの値が決定されます。バージョン 5.2 以降、評価コンテキストの #root オブジェクトは、getMethod() および getArgs() アクセサーを持つ MethodArgsHolder です。
これらの 2 つの式評価コンテキスト変数は、バージョン 5.2 から非推奨です。
#args: メソッド引数を含む
Object[]#gatewayMethod: 呼び出された
service-interfaceのメソッドを表すオブジェクト（java.reflect.Methodから派生）。この変数を含むヘッダーは、フローの後半で使用できます（ルーティングなど）。例: 単純なメソッド名でルーティングする場合は、次の式でヘッダーを追加できます:#gatewayMethod.name.
java.reflect.Method は直列化できません。後でメッセージを直列化すると、method の式を持つヘッダーは失われます。これらの場合は method.name または method.toString() を使用することをお勧めします。toString() メソッドは、パラメーターと戻り値の型を含むメソッドの String 表現を提供します。 |
バージョン 3.0 以降、呼び出されたメソッドに関係なく、<default-header/> 要素を定義して、ゲートウェイによって生成されたすべてのメッセージにヘッダーを追加できます。メソッドに定義された特定のヘッダーは、デフォルトのヘッダーよりも優先されます。ここでメソッドに定義された特定のヘッダーは、サービスインターフェースの @Header アノテーションをオーバーライドします。ただし、デフォルトのヘッダーは、サービスインターフェースの @Header アノテーションをオーバーライドしません。
ゲートウェイは、（オーバーライドされない限り）すべてのメソッドに適用される default-payload-expression もサポートするようになりました。
メソッドへの引数のメッセージへのマッピング
前のセクションの構成手法を使用すると、メソッドの引数をメッセージ要素（ペイロードとヘッダー）にマップする方法を制御できます。明示的な構成が使用されない場合、特定の規則がマッピングの実行に使用されます。場合によっては、これらの規則では、どの引数がペイロードであり、どの引数をヘッダーにマップする必要があるかを判断できません。次の例を考えてみましょう。
public String send1(Object thing1, Map thing2);
public String send2(Map thing1, Map thing2); 最初の場合、慣例では、最初の引数をペイロードにマップし（Map でない限り）、2 番目の引数の内容はヘッダーになります。
2 番目の場合（またはパラメーター thing1 の引数が Map である場合の最初の場合）、フレームワークはどの引数をペイロードにするかを決定できません。その結果、マッピングは失敗します。通常、これは payload-expression、@Payload アノテーション、@Headers アノテーションを使用して解決できます。
別の方法（および規則が破られるたび）では、メソッド呼び出しをメッセージにマッピングする責任をすべて負うことができます。そのためには、MethodArgsMessageMapper を実装し、mapper 属性を使用して <gateway/> に提供します。マッパーは MethodArgsHolder をマッピングします。これは、java.reflect.Method インスタンスと引数を含む Object[] をラップする単純なクラスです。カスタムマッパーを提供する場合、default-payload-expression 属性と <default-header/> 要素はゲートウェイで許可されていません。同様に、payload-expression 属性および <header/> 要素は、<method/> 要素では許可されていません。
マッピングメソッドの引数
次の例は、メソッド引数をメッセージにマップする方法と、無効な構成の例を示しています。
public interface MyGateway {
void payloadAndHeaderMapWithoutAnnotations(String s, Map<String, Object> map);
void payloadAndHeaderMapWithAnnotations(@Payload String s, @Headers Map<String, Object> map);
void headerValuesAndPayloadWithAnnotations(@Header("k1") String x, @Payload String s, @Header("k2") String y);
void mapOnly(Map<String, Object> map); // the payload is the map and no custom headers are added
void twoMapsAndOneAnnotatedWithPayload(@Payload Map<String, Object> payload, Map<String, Object> headers);
@Payload("#args[0] + #args[1] + '!'")
void payloadAnnotationAtMethodLevel(String a, String b);
@Payload("@someBean.exclaim(#args[0])")
void payloadAnnotationAtMethodLevelUsingBeanResolver(String s);
void payloadAnnotationWithExpression(@Payload("toUpperCase()") String s);
void payloadAnnotationWithExpressionUsingBeanResolver(@Payload("@someBean.sum(#this)") String s); // (1)
// invalid
void twoMapsWithoutAnnotations(Map<String, Object> m1, Map<String, Object> m2);
// invalid
void twoPayloads(@Payload String s1, @Payload String s2);
// invalid
void payloadAndHeaderAnnotationsOnSameParameter(@Payload @Header("x") String s);
// invalid
void payloadAndHeadersAnnotationsOnSameParameter(@Payload @Headers Map<String, Object> map);
}| 1 | この例では、SpEL 変数 #this が引数を参照していることに注意してください。この場合は s の値です。 |
メソッドの引数に #this コンテキストがないため、XML の同等物は少し異なります。ただし、次の例に示すように、式は #args 変数を使用してメソッド引数を参照できます。
<int:gateway id="myGateway" service-interface="org.something.MyGateway">
<int:method name="send1" payload-expression="#args[0] + 'thing2'"/>
<int:method name="send2" payload-expression="@someBean.sum(#args[0])"/>
<int:method name="send3" payload-expression="#method"/>
<int:method name="send4">
<int:header name="thing1" expression="#args[2].toUpperCase()"/>
</int:method>
</int:gateway>@MessagingGateway アノテーション
バージョン 4.0 以降、ゲートウェイサービスインターフェースは、構成のために <gateway /> xml 要素の定義を要求する代わりに、@MessagingGateway アノテーションでマークできます。次のペアの例では、同じゲートウェイを構成するための 2 つのアプローチを比較しています。
<int:gateway id="myGateway" service-interface="org.something.TestGateway"
default-request-channel="inputC">
<int:default-header name="calledMethod" expression="#gatewayMethod.name"/>
<int:method name="echo" request-channel="inputA" reply-timeout="2" request-timeout="200"/>
<int:method name="echoUpperCase" request-channel="inputB">
<int:header name="thing1" value="thing2"/>
</int:method>
<int:method name="echoViaDefault"/>
</int:gateway>@MessagingGateway(name = "myGateway", defaultRequestChannel = "inputC",
defaultHeaders = @GatewayHeader(name = "calledMethod",
expression="#gatewayMethod.name"))
public interface TestGateway {
@Gateway(requestChannel = "inputA", replyTimeout = 2, requestTimeout = 200)
String echo(String payload);
@Gateway(requestChannel = "inputB", headers = @GatewayHeader(name = "thing1", value="thing2"))
String echoUpperCase(String payload);
String echoViaDefault(String payload);
}XML バージョンと同様に、Spring Integration はコンポーネントスキャン中にこれらのアノテーションを検出すると、メッセージングインフラストラクチャを使用して proxy 実装を作成します。このスキャンを実行し、BeanDefinition をアプリケーションコンテキストに登録するには、@IntegrationComponentScan アノテーションを @Configuration クラスに追加します。標準の @ComponentScan インフラストラクチャはインターフェースを処理しません。そのため、カスタム @IntegrationComponentScan ロジックを導入して、インターフェース上の @MessagingGateway アノテーションを微調整し、GatewayProxyFactoryBean インスタンスを登録します。アノテーションサポートも参照してください。 |
@MessagingGateway アノテーションとともに、@Profile アノテーションでサービスインターフェースをマークして、そのようなプロファイルがアクティブでない場合、Bean の作成を回避できます。
XML 構成がない場合は、少なくとも 1 つの @Configuration クラスで @EnableIntegration アノテーションが必要です。詳細については、構成と @EnableIntegration を参照してください。 |
引数なしのメソッドの呼び出し
引数を持たない Gateway インターフェースでメソッドを呼び出す場合、デフォルトの動作は PollableChannel から Message を受け取ります。
ただし、引数のない SQL 呼び出しやストアドプロシージャのトリガーなど、ユーザーが提供するパラメーターを必要としないダウンストリームの他のコンポーネントと対話できるように、引数のないメソッドをトリガーしたい場合があります。
送受信のセマンティクスを実現するには、ペイロードを提供する必要があります。ペイロードを生成するために、インターフェースのメソッドパラメーターは必要ありません。@Payload アノテーションまたは method 要素の XML の payload-expression 属性を使用できます。次のリストには、ペイロードの例をいくつか示します。
リテラル文字列
#gatewayMethod.name
新しい java.util.Date()
@someBean.someMethod() の戻り値
次の例は、@Payload アノテーションの使用方法を示しています。
public interface Cafe {
@Payload("new java.util.Date()")
List<Order> retrieveOpenOrders();
}メソッドに引数および戻り値がなく、ペイロード式が含まれている場合、メソッドは送信専用操作として扱われます。
default メソッドの呼び出し
ゲートウェイプロキシのインターフェースには default メソッドも含まれる場合があり、バージョン 5.3 以降、フレームワークはプロキシに DefaultMethodInvokingMethodInterceptor を注入して、プロキシの代わりに java.lang.invoke.MethodHandle アプローチを使用して default メソッドを呼び出します。java.util.function.Function などの JDK からのインターフェースは引き続きゲートウェイプロキシに使用できますが、JDK クラスに対する MethodHandles.Lookup インスタンス化の内部 Java セキュリティ上の理由により、それらの default メソッドを呼び出すことができません。これらのメソッドは、メソッドの明示的な @Gateway アノテーション、または @MessagingGateway アノテーションまたは <gateway> XML コンポーネントの proxyDefaultMethods を使用して、プロキシ化（実装ロジックを失い、同時に以前のゲートウェイプロキシの動作を復元）することもできます。
エラー処理
ゲートウェイの呼び出しはエラーになる可能性があります。デフォルトでは、ダウンストリームで発生したエラーは、ゲートウェイのメソッド呼び出し時に「そのまま」再スローされます。例: 次の簡単なフローを検討してください。
gateway -> service-activator サービスアクティベータによって呼び出されたサービスが MyException をスローした場合（たとえば）、フレームワークはそれを MessagingException にラップし、failedMessage プロパティでサービスアクティベータに渡されたメッセージを添付します。そのため、フレームワークによって実行されるロギングには、障害の完全なコンテキストが含まれます。デフォルトでは、例外がゲートウェイによってキャッチされると、MyException はラップ解除され、呼び出し元にスローされます。原因チェーンの特定の例外型と一致するように、ゲートウェイメソッド宣言で throws 句を構成できます。例: ダウンストリームエラーの理由のすべてのメッセージング情報で MessagingException 全体をキャッチする場合、次のようなゲートウェイメソッドが必要です。
public interface MyGateway {
void performProcess() throws MessagingException;
}POJO プログラミングを推奨しているため、呼び出し元をメッセージングインフラストラクチャに公開したくない場合があります。
ゲートウェイメソッドに throws 句がない場合、ゲートウェイは原因ツリーを走査し、MessagingException ではない RuntimeException を探します。何も見つからない場合、フレームワークは MessagingException をスローします。前の説明の MyException に SomeOtherException とメソッド throws SomeOtherException の原因がある場合、ゲートウェイはそれをさらにアンラップし、呼び出し元にスローします。
service-interface なしでゲートウェイが宣言されると、内部フレームワークインターフェース RequestReplyExchanger が使用されます。
次の例を考えてみましょう。
public interface RequestReplyExchanger {
Message<?> exchange(Message<?> request) throws MessagingException;
} バージョン 5.0 より前は、この exchange メソッドには throws 節がありませんでした。その結果、例外はラップ解除されました。このインターフェースを使用して以前のアンラップ動作を復元する場合は、代わりにカスタム service-interface を使用するか、MessagingException の cause にアクセスしてください。
ただし、エラーを伝播するのではなくログに記録するか、例外を有効な応答として扱うことができます（呼び出し元が理解する「エラーメッセージ」規約に準拠するメッセージにマッピングすることにより）。これを実現するために、ゲートウェイは、error-channel 属性のサポートを含めることにより、エラー専用のメッセージチャネルのサポートを提供します。次の例では、「トランスフォーマー」が Exception から応答 Message を作成します。
<int:gateway id="sampleGateway"
default-request-channel="gatewayChannel"
service-interface="foo.bar.SimpleGateway"
error-channel="exceptionTransformationChannel"/>
<int:transformer input-channel="exceptionTransformationChannel"
ref="exceptionTransformer" method="createErrorResponse"/>exceptionTransformer は、予期されるエラーレスポンスオブジェクトの作成方法を知っている単純な POJO である可能性があります。それが発呼者に送り返されるペイロードになります。必要に応じて、このような「エラーフロー」でより多くの詳細な処理を実行できます。ルーター（Spring Integration の ErrorMessageExceptionTypeRouter を含む）、フィルターなどが含まれる場合があります。ただし、ほとんどの場合、単純な「トランス」で十分です。
または、例外のみをログに記録する（または非同期でどこかに送信する）こともできます。一方向のフローを提供する場合、呼び出し元には何も返されません。例外を完全に抑制したい場合は、グローバル nullChannel への参照を提供できます（本質的に /dev/null アプローチ）。最後に、上記のように、error-channel が定義されていない場合、例外は通常どおり伝播します。
@MessagingGateway アノテーションを使用する場合（@MessagingGateway アノテーションを参照 errorChannel 属性を使用できます。
バージョン 5.0 以降、void 戻り値の型（一方向フロー）でゲートウェイメソッドを使用すると、error-channel 参照（提供されている場合）が各送信メッセージの標準 errorChannel ヘッダーに取り込まれます。この機能により、標準の ExecutorChannel 構成（または QueueChannel）に基づいたダウンストリーム非同期フローが、デフォルトのグローバル errorChannel 例外送信動作をオーバーライドできます。以前は、@GatewayHeader アノテーションまたは <header> 要素で errorChannel ヘッダーを手動で指定する必要がありました。error-channel プロパティは、非同期フローを持つ void メソッドでは無視されました。代わりに、エラーメッセージがデフォルトの errorChannel に送信されました。
シンプルな POJI ゲートウェイを介してメッセージングシステムを公開することには利点がありますが、基盤となるメッセージングシステムの現実を「隠す」ことには代償が伴うため、考慮する必要がある特定の事項があります。Java メソッドができるだけ早く戻り、呼び出し側が戻るのを待機している間（void、戻り値、スローされた例外）無限にハングしないようにします。メッセージングシステムの前で通常のメソッドをプロキシとして使用する場合、基になるメッセージングの潜在的な非同期性を考慮する必要があります。これは、ゲートウェイによって開始されたメッセージがフィルターによってドロップされ、応答の生成を担当するコンポーネントに到達しない可能性があることを意味します。一部のサービスアクティベータメソッドでは例外が発生し、応答が返されない場合があります（null メッセージを生成しないため）。言い換えると、複数のシナリオが原因で、返信メッセージが届かないことがあります。これはメッセージングシステムではまったく自然なことです。ただし、ゲートウェイ方式への影響を考慮してください。ゲートウェイのメソッド入力引数はメッセージに組み込まれ、ダウンストリームに送信されました。応答メッセージは、ゲートウェイのメソッドの戻り値に変換されます。そのため、ゲートウェイコールごとに、必ず応答メッセージが存在するようにする必要があります。そうしないと、ゲートウェイメソッドが戻らず、無期限にハングする可能性があります。この状況を処理する 1 つの方法は、非同期ゲートウェイを使用することです（このセクションの後半で説明します）。それを処理する別の方法は、reply-timeout 属性を明示的に設定することです。そうすれば、ゲートウェイは reply-timeout で指定された時間より長くハングせず、タイムアウトが経過した場合に "null" を返します。最後に、サービスアクティベータで "requires-reply" などのダウンストリームフラグを設定したり、フィルターで "throw-exceptions-on-rejection" を設定することを検討できます。これらのオプションについては、この章の最後のセクションで詳しく説明します。 |
ダウンストリームフローが ErrorMessage を返す場合、その payload （Throwable）は通常のダウンストリームエラーとして扱われます。error-channel が構成されている場合、エラーフローに送信されます。それ以外の場合、ペイロードはゲートウェイの呼び出し元にスローされます。同様に、error-channel のエラーフローが ErrorMessage を返す場合、そのペイロードは呼び出し元にスローされます。同じことが、Throwable ペイロードを持つすべてのメッセージに当てはまります。これは、Exception を直接呼び出し元に伝搬する必要がある場合の非同期状況で役立ちます。これを行うには、Exception を返す（サービスから reply として）か、スローします。一般に、非同期フローの場合でも、フレームワークは、ダウンストリームフローによってゲートウェイにスローされる例外の伝播を処理します。TCP クライアントサーバーマルチプレックス [GitHub] (英語) サンプルは、呼び出し元に例外を返すための両方の手法を示しています。aggregator と group-timeout （アグリゲーターとグループのタイムアウトを参照）および破棄フローでの MessagingTimeoutException 応答を使用して、待機スレッドへのソケット IO エラーをエミュレートします。 |
ゲートウェイのタイムアウト
ゲートウェイには、requestTimeout と replyTimeout の 2 つのタイムアウトプロパティがあります。リクエストのタイムアウトは、チャネルがブロックできる場合にのみ適用されます（たとえば、境界のある QueueChannel がいっぱいの場合）。replyTimeout 値は、ゲートウェイが応答を待つ時間または null を返す時間です。デフォルトは無限です。
タイムアウトは、ゲートウェイ（defaultRequestTimeout および defaultReplyTimeout）または MessagingGateway インターフェースアノテーションのすべてのメソッドのデフォルトとして設定できます。個々のメソッドは、これらのデフォルト（<method/> 子要素内）または @Gateway アノテーションをオーバーライドできます。
バージョン 5.0 以降、次の例に示すように、タイムアウトは式として定義できます。
@Gateway(payloadExpression = "#args[0]", requestChannel = "someChannel",
requestTimeoutExpression = "#args[1]", replyTimeoutExpression = "#args[2]")
String lateReply(String payload, long requestTimeout, long replyTimeout); 評価コンテキストには BeanResolver （他の Bean を参照するには @someBean を使用）があり、#args 配列変数が使用可能です。
XML で設定する場合、タイムアウト属性は、次の例に示すように、長い値または SpEL 式にすることができます。
<method name="someMethod" request-channel="someRequestChannel"
payload-expression="#args[0]"
request-timeout="1000"
reply-timeout="#args[1]">
</method>非同期ゲートウェイ
パターンとして、メッセージングゲートウェイは、メッセージングシステムの全機能を公開しながら、メッセージング固有のコードを非表示にする優れたメソッドを提供します。前述のように、GatewayProxyFactoryBean は、サービスインターフェースを介してプロキシを公開する便利なメソッドを提供し、メッセージングシステムへの POJO ベースのアクセスを提供します（独自のドメイン内のオブジェクト、プリミティブ / 文字列、その他のオブジェクトに基づく）。ただし、値を返す単純な POJO メソッドを介してゲートウェイが公開されている場合、リクエストメッセージ（メソッドが呼び出されたときに生成された）ごとに、応答メッセージ（メソッドが返されたときに生成された）が必要です。メッセージングシステムは本来非同期であるため、「各リクエストに対して常に応答がある」という契約を常に保証できるとは限りません。Spring Integration 2.0 は非同期ゲートウェイのサポートを導入しました。これは、応答が予期されているかどうか、応答が到着するまでにかかる時間がわからない場合に、フローを開始する便利なメソッドを提供します。
これらの型のシナリオを処理するために、Spring Integration は java.util.concurrent.Future インスタンスを使用して非同期ゲートウェイをサポートします。
次の例に示すように、XML 構成から変更はなく、通常のゲートウェイを定義するのと同じ方法で非同期ゲートウェイを定義します。
<int:gateway id="mathService"
service-interface="org.springframework.integration.sample.gateway.futures.MathServiceGateway"
default-request-channel="requestChannel"/>ただし、ゲートウェイインターフェース（サービスインターフェース）は、次のように少し異なります。
public interface MathServiceGateway {
Future<Integer> multiplyByTwo(int i);
} 前の例が示すように、ゲートウェイメソッドの戻り値の型は Future です。GatewayProxyFactoryBean は、ゲートウェイメソッドの戻り値の型が Future であることを確認すると、AsyncTaskExecutor を使用して直ちに非同期モードに切り替えます。それが違いの範囲です。このようなメソッドの呼び出しは、常に Future インスタンスですぐに返されます。その後、自分のペースで Future を操作して、結果の取得やキャンセルなどを行うことができます。また、Future インスタンスの他の使用と同様に、get() を呼び出すと、タイムアウト、実行例外などが明らかになる場合があります。次の例は、非同期ゲートウェイから戻る Future の使用方法を示しています。
MathServiceGateway mathService = ac.getBean("mathService", MathServiceGateway.class);
Future<Integer> result = mathService.multiplyByTwo(number);
// do something else here since the reply might take a moment
int finalResult = result.get(1000, TimeUnit.SECONDS);より詳細な例については、Spring Integration サンプルの async-gateway [GitHub] (英語) サンプルを参照してください。
ListenableFuture
バージョン 4.1 以降、非同期ゲートウェイメソッドは ListenableFuture （Spring Framework 4.0 で導入）を返すこともできます。これらの戻り値の型を使用すると、結果が利用可能になった（または例外が発生した）ときに呼び出されるコールバックを提供できます。ゲートウェイがこの戻り値の型を検出し、タスクエグゼキューターが AsyncListenableTaskExecutor である場合、エグゼキューターの submitListenable() メソッドが呼び出されます。次の例は、ListenableFuture の使用方法を示しています。
ListenableFuture<String> result = this.asyncGateway.async("something");
result.addCallback(new ListenableFutureCallback<String>() {
@Override
public void onSuccess(String result) {
...
}
@Override
public void onFailure(Throwable t) {
...
}
});AsyncTaskExecutor
デフォルトでは、GatewayProxyFactoryBean は、戻り値の型が Future であるゲートウェイメソッドの内部 AsyncInvocationTask インスタンスを送信するときに org.springframework.core.task.SimpleAsyncTaskExecutor を使用します。ただし、<gateway/> 要素の構成の async-executor 属性により、Spring アプリケーションコンテキスト内で利用可能な java.util.concurrent.Executor の実装への参照を提供できます。
（デフォルト） SimpleAsyncTaskExecutor は、Future と ListenableFuture の両方の戻り値の型をサポートし、それぞれ FutureTask または ListenableFutureTask を返します。CompletableFuture を参照してください。デフォルトのエグゼキューターがありますが、次の例に示すように、ログでスレッドを識別できるように外部エグゼキューターを提供すると便利です（XML を使用する場合、スレッド名はエグゼキューターの Bean 名に基づきます）。
@Bean
public AsyncTaskExecutor exec() {
SimpleAsyncTaskExecutor simpleAsyncTaskExecutor = new SimpleAsyncTaskExecutor();
simpleAsyncTaskExecutor.setThreadNamePrefix("exec-");
return simpleAsyncTaskExecutor;
}
@MessagingGateway(asyncExecutor = "exec")
public interface ExecGateway {
@Gateway(requestChannel = "gatewayChannel")
Future<?> doAsync(String foo);
} 別の Future 実装を返す場合は、カスタムエグゼキューターを提供するか、エグゼキューターを完全に無効にして、ダウンストリームフローからの応答メッセージペイロードで Future を返すことができます。エグゼキューターを無効にするには、GatewayProxyFactoryBean で null に設定します（setAsyncTaskExecutor(null) を使用）。XML を使用してゲートウェイを構成する場合は、async-executor="" を使用します。@MessagingGateway アノテーションを使用して構成する場合、次のようなコードを使用します。
@MessagingGateway(asyncExecutor = AnnotationConstants.NULL)
public interface NoExecGateway {
@Gateway(requestChannel = "gatewayChannel")
Future<?> doAsync(String foo);
} 戻り値の型が特定の具体的な Future 実装または構成されたエグゼキューターでサポートされていない他のサブインターフェースである場合、フローは呼び出し側のスレッドで実行され、フローは応答メッセージペイロードで必要な型を返す必要があります |
CompletableFuture
バージョン 4.2 以降、ゲートウェイメソッドは CompletableFuture<?> を返すことができるようになりました。この型を返す場合、2 つの操作モードがあります。
非同期エグゼキューターが提供され、戻り値の型が（サブクラスではなく）
CompletableFutureである場合、フレームワークはエグゼキューターでタスクを実行し、すぐに呼び出し元にCompletableFutureを返します。CompletableFuture.supplyAsync(Supplier<U> supplier, Executor executor)は未来の創造に使用されます。非同期エグゼキューターが明示的に
nullに設定され、戻り値の型がCompletableFutureであるか、戻り値の型がCompletableFutureのサブクラスである場合、フローは呼び出し側のスレッドで呼び出されます。このシナリオでは、ダウンストリームフローは適切な型のCompletableFutureを返すことが期待されています。
使用シナリオ
次のシナリオでは、ダウンストリームフローが（Invoice オブジェクトを使用して）ゲートウェイに応答すると、呼び出し元スレッドはすぐに CompletableFuture<Invoice> を返します。
CompletableFuture<Invoice> order(Order order);<int:gateway service-interface="something.Service" default-request-channel="orders" /> 次のシナリオでは、ダウンストリームフローがゲートウェイへの応答のペイロードとして提供する場合、呼び出し元スレッドは CompletableFuture<Invoice> を返します。請求書の準備ができたら、他のプロセスで将来を完了する必要があります。
CompletableFuture<Invoice> order(Order order);<int:gateway service-interface="foo.Service" default-request-channel="orders"
async-executor="" /> 次のシナリオでは、ダウンストリームフローがゲートウェイへの応答のペイロードとして提供する場合、呼び出し元スレッドは CompletableFuture<Invoice> を返します。請求書の準備ができたら、他のプロセスで将来を完了する必要があります。DEBUG ロギングが有効になっている場合、ログエントリが出力され、このシナリオでは非同期エグゼキューターを使用できないことを示します。
MyCompletableFuture<Invoice> order(Order order);<int:gateway service-interface="foo.Service" default-request-channel="orders" /> 次の例に示すように、CompletableFuture インスタンスを使用して、応答に対して追加の操作を実行できます。
CompletableFuture<String> process(String data);
...
CompletableFuture result = process("foo")
.thenApply(t -> t.toUpperCase());
...
String out = result.get(10, TimeUnit.SECONDS);Reactor Mono
バージョン 5.0 以降、GatewayProxyFactoryBean では、Mono<T> [GitHub] (英語) 戻り型を使用して、ゲートウェイインターフェースメソッドで Project Reactor (英語) を使用できます。内部 AsyncInvocationTask は Mono.fromCallable() にラップされています。
Mono を使用して、後で結果を取得することができます（Future<?> と同様）。または、結果がゲートウェイに返されたときに Consumer を呼び出すことにより、ディスパッチャーから結果を取得できます。
Mono はフレームワークによってすぐにフラッシュされません。そのため、ゲートウェイメソッドが戻る前に基となるメッセージフローが開始されません (Future<?> Executor タスクの場合と同様)。フローは Mono が購読されたときに開始されます。あるいは、Mono (「コンポーザブル」であるため) は Reactor ストリームの一部である可能性があり、subscribe() は Flux 全体に関連付けられています。次の例は、Project Reactor を使用してゲートウェイを作成する方法を示しています。 |
@MessagingGateway
public static interface TestGateway {
@Gateway(requestChannel = "promiseChannel")
Mono<Integer> multiply(Integer value);
}
...
@ServiceActivator(inputChannel = "promiseChannel")
public Integer multiply(Integer value) {
return value * 2;
}
...
Flux.just("1", "2", "3", "4", "5")
.map(Integer::parseInt)
.flatMap(this.testGateway::multiply)
.collectList()
.subscribe(integers -> ...);Project Reactor を使用するもう 1 つの例は、次の例に示すようなシンプルなコールバックシナリオです。
Mono<Invoice> mono = service.process(myOrder);
mono.subscribe(invoice -> handleInvoice(invoice)); 呼び出しスレッドは継続し、フローが完了すると handleInvoice() が呼び出されます。
非同期型を返すダウンストリームフロー
上記の ListenableFuture セクションで説明したように、ダウンストリームコンポーネントが非同期ペイロード（Future、Mono など）を含むメッセージを返すようにする場合は、非同期エグゼキューターを null （または XML 構成を使用する場合は ""）に明示的に設定する必要があります。次に、フローは呼び出し元のスレッドで呼び出され、結果を後で取得できます。
void 戻り値の型
前述の戻り値の型とは異なり、メソッドの戻り値の型が void の場合、フレームワークは、ダウンストリームフローを非同期に実行することを暗黙的に決定することはできません。呼び出し元のスレッドはすぐに戻ります。この場合、次の例に示すように、インターフェースメソッドに @Async でアノテーションを付ける必要があります。
@MessagingGateway
public interface MyGateway {
@Gateway(requestChannel = "sendAsyncChannel")
@Async
void sendAsync(String payload);
}Future<?> 戻り型とは異なり、カスタム TaskExecutor （ErrorHandlingTaskExecutor など）が @Async アノテーションに関連付けられていない限り、フローによって例外がスローされた場合に呼び出し元に通知する方法はありません。
レスポンスがない場合のゲートウェイの動作
前に説明したように、ゲートウェイは、POJO メソッド呼び出しを通じてメッセージングシステムと対話する便利な方法を提供します。ただし、通常は常に例外が返されると予想される一般的なメソッド呼び出しは、メッセージ交換に 1 対 1 を常にマップするとは限りません（たとえば、応答メッセージが到着しない場合があります。戻る）。
このセクションの残りの部分では、さまざまなシナリオと、ゲートウェイをより予測可能な動作にする方法について説明します。同期ゲートウェイの動作をより予測可能にするために特定の属性を構成できますが、一部の属性は期待どおりに動作しない場合があります。それらの 1 つは reply-timeout （メソッドレベルまたはゲートウェイレベルの default-reply-timeout）です。reply-timeout 属性を調べて、さまざまなシナリオで同期ゲートウェイの動作にどのように影響を与えることができるか、およびできないかを確認します。シングルスレッドシナリオ（ダウンストリームのすべてのコンポーネントが直接チャネルを介して接続されている）とマルチスレッドシナリオ（たとえば、ダウンストリームのどこかに、シングルスレッド境界を破るポーリング可能チャネルまたはエグゼキュータチャネルがある場合）を調べます。
長期実行プロセスダウンストリーム
- 同期ゲートウェイ、シングルスレッド
コンポーネントのダウンストリームがまだ実行されている場合（おそらく無限ループまたは遅いサービスのため）、
reply-timeoutの設定は効果がなく、ダウンストリームサービスが終了するまで（例外を返すかスローすることにより）ゲートウェイメソッド呼び出しは戻りません。- 同期ゲートウェイ、マルチスレッド
マルチスレッドのメッセージフローでコンポーネントのダウンストリームが（おそらく無限ループまたは遅いサービスのために）実行されている場合、
GatewayProxyFactoryBeanがタイムアウトに達するとゲートウェイメソッドの呼び出しが返されるようにすることでreply-timeoutを設定すると効果があります。応答チャネルでポーリングし、タイムアウトが期限切れになるまでメッセージを待機します。ただし、実際の応答が生成される前にタイムアウトに達すると、ゲートウェイメソッドから "null" が返される可能性があります。ゲートウェイメソッドの呼び出しが返された後、応答メッセージ（生成された場合）が応答チャネルに送信されることを理解する必要があります。
ダウンストリームコンポーネントが "null" を返す
- 同期ゲートウェイ — single-threaded
コンポーネントダウンストリームが "null" を返し、
reply-timeoutが設定されていない場合、reply-timeoutが設定されているか、requires-reply属性がダウンストリームコンポーネント（サービスアクティベータなど）に設定されていない限り、ゲートウェイメソッド呼び出しは無期限にハングします 'null'。この場合、例外がスローされ、ゲートウェイに伝播されます。- 同期ゲートウェイ — multi-threaded
動作は前のケースと同じです。
ゲートウェイメソッドシグネチャーが非 void の場合、ダウンストリームコンポーネントのリターンシグネチャーは "void"
- 同期ゲートウェイ — single-threaded
コンポーネントダウンストリームが "void" を返し、
reply-timeoutが設定されていない場合、reply-timeoutが設定されていない限り、ゲートウェイメソッド呼び出しは無期限にハングします。- 同期ゲートウェイ — multi-threaded
動作は前のケースと同じです。
ダウンストリームコンポーネントがランタイム例外を引き起こす
- 同期ゲートウェイ — single-threaded
コンポーネントのダウンストリームがランタイム例外をスローした場合、例外はエラーメッセージを介してゲートウェイに伝播され、再スローされます。
- 同期ゲートウェイ — multi-threaded
動作は前のケースと同じです。
デフォルトでは、reply-timeout は無制限であることを理解する必要があります。そのため、reply-timeout を明示的に設定しないと、ゲートウェイメソッドの呼び出しが無期限にハングする機能があります。フローを確実に分析し、これらのシナリオの 1 つがリモートで発生する可能性がある場合でも、reply-timeout 属性を " 'safe'" 値に設定する必要があります。さらに良いことに、ダウンストリームコンポーネントの requires-reply 属性を 'true' に設定して、そのダウンストリームコンポーネントが内部で null を返すとすぐに例外をスローすることで生成されるタイムリーなレスポンスを確保できます。ただし、reply-timeout が役に立たないシナリオ（最初のシナリオを参照）があることも認識する必要があります。つまり、メッセージフローを分析し、非同期ゲートウェイではなく同期ゲートウェイをいつ使用するかを決定することも重要です。前述のように、後者の場合は、Future インスタンスを返すゲートウェイメソッドを定義する問題です。その後、その戻り値を受け取ることが保証され、呼び出しの結果をよりきめ細かく制御できます。また、ルーターを扱う場合、resolution-required 属性を "true" に設定すると、特定のチャネルを解決できない場合にルーターによって例外がスローされることに注意してください。同様に、フィルターを扱う場合、throw-exception-on-rejection 属性を設定できます。どちらの場合も、結果のフローは、'requires-reply' 属性のサービスアクティベーターを含むように動作します。つまり、ゲートウェイメソッドの呼び出しからタイムリーなレスポンスを保証できます。 |
reply-timeout は、<gateway/> エレメント（GatewayProxyFactoryBean によって作成された）に対して制限されていません。外部統合用の受信ゲートウェイ（WS、HTTP など）は、これらのゲートウェイと多くの特性と属性を共有しています。ただし、これらの受信ゲートウェイの場合、デフォルトの reply-timeout は 1000 ミリ秒（1 秒）です。別のスレッドへのダウンストリーム非同期ハンドオフが行われる場合、この属性を増やして、ゲートウェイがタイムアウトする前にフローが完了するのに十分な時間を確保する必要があります。 |
| スレッドがゲートウェイに戻ると、つまりフローが完了するか、メッセージが別のスレッドに渡されると、タイマーが開始することを理解する必要があります。その時点で、呼び出しスレッドは応答の待機を開始します。フローが完全に同期されている場合、応答はすぐに利用可能です。非同期フローの場合、スレッドはこの時間まで待機します。 |
IntegrationFlows を介してゲートウェイを定義するオプションについては、Java DSL の章のゲートウェイとしての IntegrationFlow を参照してください。
サービスアクティベーター
サービスアクティベータは、Spring が管理するオブジェクトを入力チャネルに接続して、サービスのロールを果たすようにするためのエンドポイント型です。サービスが出力を生成する場合、出力チャネルにも接続される場合があります。または、出力生成サービスを処理パイプラインまたはメッセージフローの最後に配置することもできます。この場合、受信メッセージの replyChannel ヘッダーを使用できます。これは、出力チャネルが定義されていない場合のデフォルトの動作です。ここで説明するほとんどの構成オプションと同様に、実際には同じ動作が他のほとんどのコンポーネントに適用されます。
Service Activator の構成
サービスアクティベータを作成するには、次の例に示すように、"input-channel" および "ref" 属性を使用して 'service-activator' 要素を使用します。
<int:service-activator input-channel="exampleChannel" ref="exampleHandler"/> 前述の構成では、次のメッセージング要件のいずれかを満たす exampleHandler からすべてのメソッドを選択します。
@ServiceActivatorのアノテーションが付けられていますpublicrequiresReply == trueの場合、voidを返さない
実行時の呼び出しのターゲットメソッドは、各リクエストメッセージに対して payload 型によって、またはそのようなメソッドがターゲットクラスに存在する場合は Message<?> 型へのフォールバックとして選択されます。
バージョン 5.0 以降、1 つのサービスメソッドを、一致しないすべてのケースのフォールバックとして @org.springframework.integration.annotation.Default でマークできます。これは、変換後に呼び出されるターゲットメソッドでコンテンツ型変換を使用する場合に役立ちます。
オブジェクトの明示的に定義されたメソッドに委譲するには、次の例に示すように、method 属性を追加できます。
<int:service-activator input-channel="exampleChannel" ref="somePojo" method="someMethod"/> いずれの場合でも、サービスメソッドが null 以外の値を返すと、エンドポイントは応答メッセージを適切な応答チャネルに送信しようとします。応答チャネルを決定するには、次の例に示すように、エンドポイント構成で output-channel が提供されたかどうかを最初に確認します。
<int:service-activator input-channel="exampleChannel" output-channel="replyChannel"
ref="somePojo" method="someMethod"/> メソッドが結果を返し、output-channel が定義されていない場合、フレームワークはリクエストメッセージの replyChannel ヘッダー値を確認します。その値が利用可能な場合、その型をチェックします。MessageChannel の場合、応答メッセージはそのチャネルに送信されます。String の場合、エンドポイントはチャネル名をチャネルインスタンスに解決しようとします。チャネルを解決できない場合、DestinationResolutionException がスローされます。解決できた場合、メッセージはそこに送信されます。リクエストメッセージに replyChannel ヘッダーがなく、reply オブジェクトが Message である場合、その宛先の replyChannel ヘッダーが調べられます。これは、Spring Integration でのリクエスト / 応答メッセージングに使用される手法であり、リターンアドレスパターンの例でもあります。
メソッドが結果を返し、それを破棄してフローを終了する場合は、output-channel を構成して NullChannel に送信する必要があります。便宜上、フレームワークは nullChannel という名前で登録します。詳細については、特別チャンネルを参照してください。
サービスアクティベータは、応答メッセージを生成するために必要ではないコンポーネントの 1 つです。メソッドが null を返す場合、または void 戻り型を持っている場合、サービスアクティベータはメソッド呼び出しの後にシグナルなしで終了します。この動作は、AbstractReplyProducingMessageHandler.requiresReply オプションによって制御できます。このオプションは、XML 名前空間で構成するときに requires-reply として公開されます。フラグが true に設定され、メソッドが null を返す場合、ReplyRequiredException がスローされます。
サービスメソッドの引数は、メッセージまたは任意の型のいずれかです。後者の場合、メッセージから抽出され、サービスメソッドに注入されるメッセージペイロードであると想定されます。Spring Integration を使用する場合、POJO モデルをフォローし、促進するため、一般的にこのアプローチをお勧めします。アノテーションサポートに従って、引数には @Header または @Headers アノテーションが付いている場合もあります。
| サービスメソッドに引数を指定する必要はありません。つまり、イベントスタイルのサービスアクティベーター（サービスメソッドの呼び出しだけが重要）を実装でき、メッセージの内容を心配する必要はありません。null JMS メッセージと考えてください。このような実装の使用例の例は、入力チャネルに置かれたメッセージの単純なカウンターまたはモニターです。 |
次の例に示すように、バージョン 4.1 以降、フレームワークはメッセージプロパティ（payload および headers）を Java 8 Optional POJO メソッドパラメーターに正しく変換します。
public class MyBean {
public String computeValue(Optional<String> payload,
@Header(value="foo", required=false) String foo1,
@Header(value="foo") Optional<String> foo2) {
if (payload.isPresent()) {
String value = payload.get();
...
}
else {
...
}
}
} 通常、カスタムサービスアクティベーターハンドラーの実装を他の <service-activator> 定義で再利用できる場合は、ref 属性を使用することをお勧めします。ただし、カスタムサービスアクティベーターハンドラーの実装が <service-activator> の単一の定義内でのみ使用される場合、次の例に示すように、内部 Bean 定義を提供できます。
<int:service-activator id="exampleServiceActivator" input-channel="inChannel"
output-channel = "outChannel" method="someMethod">
<beans:bean class="org.something.ExampleServiceActivator"/>
</int:service-activator> 同じ <service-activator> 構成で ref 属性と内部ハンドラー定義の両方を使用することは許可されません。曖昧な状態を作成し、例外がスローされるためです。 |
ref 属性が AbstractMessageProducingHandler を継承する Bean を参照する場合（フレームワーク自体が提供するハンドラーなど）、出力チャネルをハンドラーに直接注入することにより、構成が最適化されます。この場合、各 ref は、個別の Bean インスタンス（または prototype -scoped Bean）にするか、内部 <bean/> 構成型を使用する必要があります。誤って複数の Bean から同じメッセージハンドラーを参照すると、構成例外が発生します。 |
サービスアクティベーターと Spring 式言語 (SpEL)
Spring Integration 2.0 以降、サービスアクティベーターも SpEL の恩恵を受けることができます。
例: 次のように、ref 属性の Bean を指すことなく、または内部 Bean 定義として含めることなく、Bean メソッドを呼び出すことができます。
<int:service-activator input-channel="in" output-channel="out"
expression="@accountService.processAccount(payload, headers.accountId)"/>
<bean id="accountService" class="thing1.thing2.Account"/> 上記の構成では、ref または内部 Bean を使用して 'accountService' を挿入する代わりに、SpEL の @beanId 表記法を使用して、メッセージペイロードと互換性のある型を受け取るメソッドを呼び出します。また、ヘッダー値も渡します。有効な SpEL 式は、メッセージ内の任意のコンテンツに対して評価できます。単純なシナリオでは、次の例に示すように、すべてのロジックをそのような式にカプセル化できる場合、サービスアクティベーターは Bean を参照する必要はありません。
<int:service-activator input-channel="in" output-channel="out" expression="payload * 2"/>上記の構成では、サービスロジックはペイロード値に 2 を掛けます。SpEL を使用すると、比較的簡単に処理できます。
サービスアクティベーターの設定の詳細については、Java DSL の章のサービスアクティベーターと .handle() メソッドを参照してください。
非同期サービスアクティベーター
サービスアクティベーターは、呼び出しスレッドによって呼び出されます。入力チャネルが SubscribableChannel または PollableChannel のポーラースレッドの場合、これはアップストリームスレッドです。サービスが ListenableFuture<?> を返す場合、デフォルトのアクションは、それを出力（または応答）チャネルに送信されるメッセージのペイロードとして送信することです。バージョン 4.3 から、async 属性を true に設定できるようになりました（Java 構成の使用時に setAsync(true) を使用）。この async 属性が true に設定されているときにサービスが ListenableFuture<?> を返すと、呼び出し元のスレッドはすぐに解放され、応答メッセージが（サービス内から）スレッドで送信され、将来を完了します。これは、ポーラースレッドがフレームワーク内で他のサービスを実行するために解放されるため、PollableChannel を使用する長時間実行サービスに特に有利です。
サービスが Exception で将来を完了すると、通常のエラー処理が発生します。ErrorMessage は、存在する場合、errorChannel メッセージヘッダーに送信されます。そうでない場合、ErrorMessage はデフォルトの errorChannel に送信されます（使用可能な場合）。
サービスアクティベーターとメソッドの戻り値の型
サービスメソッドは、応答メッセージペイロードになる任意の型を返すことができます。この場合、新しい Message<?> オブジェクトが作成され、リクエストメッセージのすべてのヘッダーがコピーされます。対話が POJO メソッド呼び出しに基づいている場合、これはほとんどの Spring Integration MessageHandler 実装で同じように機能します。
完全な Message<?> オブジェクトをメソッドから返すこともできます。ただし、Transformer とは異なり、Service Activator の場合、このメッセージは、返されたメッセージにまだ存在しない場合は、リクエストメッセージからヘッダーをコピーすることによって変更されることに注意してください。メソッドパラメーターが Message<?> であり、サービスメソッドの既存のヘッダーのすべてではなく一部をコピーすると、それらは応答メッセージに再表示されます。応答メッセージからヘッダーを削除することは ServiceActivator の責任ではなく、疎結合の原則を追求するために、統合フローに HeaderFilter を追加することをお勧めします。または、Service Activator の代わりに Transformer を使用することもできますが、その場合、完全な Message<?> を返すとき、メソッドは、リクエストメッセージヘッダーのコピー（必要な場合）を含め、メッセージに対して完全に責任があります。重要なフレームワークヘッダー（replyChannel、errorChannel など）が存在する場合は、それを保持する必要があることを確認する必要があります。
遅延器
遅延器は、メッセージフローを特定の間隔で遅延させる単純なエンドポイントです。メッセージが遅延しても、元の送信者はブロックしません。代わりに、遅延が経過した後に出力チャネルに送信されるように、org.springframework.scheduling.TaskScheduler のインスタンスで遅延メッセージがスケジュールされます。このアプローチは、ブロックされた送信者スレッドの数が多くならないため、かなり長い遅延でもスケーラブルです。それどころか、典型的なケースでは、スレッドプールはメッセージを解放する実際の実行に使用されます。このセクションには、遅延器を構成するいくつかの例が含まれています。
遅延器の構成
<delayer> 要素は、2 つのメッセージチャネル間のメッセージフローを遅延させるために使用されます。他のエンドポイントと同様に、"input-channel" および "output-channel" 属性を指定できますが、遅延器にはミリ秒数を決定する "default-delay" および "expression" 属性（および 'expression' 要素）もあります。各メッセージを遅延させる必要があります。次の例では、すべてのメッセージを 3 秒遅延させます。
<int:delayer id="delayer" input-channel="input"
default-delay="3000" output-channel="output"/>各メッセージの遅延を判別する必要がある場合、次の式が示すように、'expression' 属性を使用して SpEL 式を提供することもできます。
<int:delayer id="delayer" input-channel="input" output-channel="output"
default-delay="3000" expression="headers['delay']"/> 前の例では、3 秒の遅延は、指定された受信メッセージに対して式が null と評価された場合にのみ適用されます。式評価の有効な結果を持つメッセージにのみ遅延を適用する場合は、0 （デフォルト）の「デフォルト遅延」を使用できます。0 （またはそれ以下）の遅延があるメッセージの場合、メッセージは呼び出しスレッドですぐに送信されます。
次の例は、前述の例に相当する Java 構成を示しています。
@ServiceActivator(inputChannel = "input")
@Bean
public DelayHandler delayer() {
DelayHandler handler = new DelayHandler("delayer.messageGroupId");
handler.setDefaultDelay(3_000L);
handler.setDelayExpressionString("headers['delay']");
handler.setOutputChannelName("output");
return handler;
}次の例は、前述の例に相当する Java DSL を示しています。
@Bean
public IntegrationFlow flow() {
return IntegrationFlows.from("input")
.delay("delayer.messageGroupId", d -> d
.defaultDelay(3_000L)
.delayExpression("headers['delay']"))
.channel("output")
.get();
}XML パーサーは、<beanName>.messageGroupId のメッセージグループ ID を使用します。 |
遅延ハンドラーは、ミリ秒単位の間隔を表す式評価結果（toString() メソッドが Long に解析できる値を生成する Object）および絶対時間を表す java.util.Date インスタンスをサポートします。最初のケースでは、ミリ秒は現在の時刻からカウントされます（たとえば、5000 の値は、遅延機が受信した時刻から少なくとも 5 秒間メッセージを遅延させます）。Date インスタンスの場合、メッセージはその Date オブジェクトが表す時間まで解放されません。非正の遅延または過去の日付に等しい値は、遅延なしになります。代わりに、元の送信者のスレッドの出力チャネルに直接送信されます。式の評価結果が Date ではなく、Long として解析できない場合、デフォルトの遅延（ある場合 - デフォルトは 0）が適用されます。 |
式の評価では、無効な式やその他の条件など、さまざまな理由で評価例外がスローされる場合があります。デフォルトでは、そのような例外は無視され（DEBUG レベルで記録されます）、遅延器はデフォルトの遅延（存在する場合）にフォールバックします。ignore-expression-failures 属性を設定することにより、この動作を変更できます。デフォルトでは、この属性は true に設定されており、遅延器の動作は前述のとおりです。ただし、式評価の例外を無視せずに遅延者の呼び出し元にスローする場合は、ignore-expression-failures 属性を false に設定します。 |
上記の例では、遅延式は その結果、ヘッダーが省略される可能性があり、デフォルトの遅延にフォールバックする場合、null の検出が高速になるため、一般にドットプロパティアクセサー構文の代わりにインデクサー構文を使用する方が効率的です（推奨）。例外をキャッチするよりも。 |
遅延器は、Spring の TaskScheduler 抽象化のインスタンスに委譲します。遅延器によって使用されるデフォルトのスケジューラは、起動時に Spring Integration によって提供される ThreadPoolTaskScheduler インスタンスです。タスクスケジューラの構成を参照してください。別のスケジューラに委譲する場合は、次の例に示すように、delayer 要素の 'scheduler' 属性を介して参照を提供できます。
<int:delayer id="delayer" input-channel="input" output-channel="output"
expression="headers.delay"
scheduler="exampleTaskScheduler"/>
<task:scheduler id="exampleTaskScheduler" pool-size="3"/> 外部 ThreadPoolTaskScheduler を構成する場合、このプロパティで waitForTasksToCompleteOnShutdown = true を設定できます。これにより、アプリケーションのシャットダウン時にすでに実行状態にある（メッセージを解放する）「遅延」タスクを正常に完了できます。Spring Integration 2.2 より前は、DelayHandler が独自のスケジューラをバックグラウンドで作成できるため、このプロパティは <delayer> 要素で使用できました。2.2 以降、遅延器には外部スケジューラインスタンスが必要で、waitForTasksToCompleteOnShutdown が削除されました。スケジューラー独自の構成を使用する必要があります。 |
ThreadPoolTaskScheduler にはプロパティ errorHandler があり、org.springframework.util.ErrorHandler の実装を使用して注入できます。このハンドラーにより、遅延メッセージを送信するスケジュールされたタスクのスレッドから Exception を処理できます。デフォルトでは、org.springframework.scheduling.support.TaskUtils$LoggingErrorHandler を使用し、ログにスタックトレースを表示できます。org.springframework.integration.channel.MessagePublishingErrorHandler の使用を検討することもできます。org.springframework.integration.channel.MessagePublishingErrorHandler は、失敗したメッセージのヘッダーから、またはデフォルトの error-channel に、ErrorMessage を error-channel に送信します。このエラー処理は、トランザクションがロールバックした後に実行されます（存在する場合）。リリースの失敗を参照してください。 |
遅延器とメッセージストア
DelayHandler は、遅延メッセージを指定された MessageStore のメッセージグループに保持します。( "groupId" は、<delayer> 要素の必須 'id' 属性 に基づいています) 遅延メッセージは、DelayHandler がメッセージを output-channel に送信する直前に、スケジュールされたタスクによって MessageStore から削除されます。指定された MessageStore が永続的である場合 (JdbcMessageStore など)、アプリケーションのシャットダウン時にメッセージを失わない機能が提供されます。アプリケーションの起動後、DelayHandler は MessageStore のメッセージグループからメッセージを読み取り、メッセージの元の到着時間 (遅延が数値の場合) に基づいて遅延を付けて再スケジュールします。遅延ヘッダーが Date であったメッセージの場合、再スケジュール時にその Date が使用されます。遅延メッセージが「遅延」を超えて MessageStore に残っている場合、起動後すぐに送信されます。
<delayer> は、相互に排他的な 2 つの要素 <transactional> および <advice-chain> のいずれかで強化できます。これらの AOP アドバイスの List は、遅延の後に、スケジュールされたタスクの Thread でメッセージを解放する責任を持つプロキシ化された内部 DelayHandler.ReleaseMessageHandler に適用されます。たとえば、ダウンストリームメッセージフローが例外をスローし、ReleaseMessageHandler のトランザクションがロールバックされる場合に使用できます。この場合、遅延メッセージは永続的な MessageStore に残ります。<advice-chain> 内で任意のカスタム org.aopalliance.aop.Advice 実装を使用できます。<transactional> 要素は、トランザクションアドバイスのみを持つ単純なアドバイスチェーンを定義します。次の例は、<delayer> 内の advice-chain を示しています。
<int:delayer id="delayer" input-channel="input" output-channel="output"
expression="headers.delay"
message-store="jdbcMessageStore">
<int:advice-chain>
<beans:ref bean="customAdviceBean"/>
<tx:advice>
<tx:attributes>
<tx:method name="*" read-only="true"/>
</tx:attributes>
</tx:advice>
</int:advice-chain>
</int:delayer>DelayHandler は、管理操作（getDelayedMessageCount および reschedulePersistedMessages）を備えた JMX MBean としてエクスポートできます。これにより、たとえば TaskScheduler が以前に停止されていた場合、実行時に遅延持続メッセージの再スケジュールが可能になります。これらの操作は、次の例に示すように、Control Bus コマンドを介して呼び出すことができます。
Message<String> delayerReschedulingMessage =
MessageBuilder.withPayload("@'delayer.handler'.reschedulePersistedMessages()").build();
controlBusChannel.send(delayerReschedulingMessage);| メッセージストア、JMX、コントロールバスの詳細については、システムマネジメントを参照してください。 |
リリースの失敗
バージョン 5.0.8 以降、遅延器には 2 つの新しいプロパティがあります。
maxAttempts(デフォルト 5)retryDelay(デフォルトは 1 秒)
メッセージがリリースされたときに、ダウンストリームフローが失敗すると、retryDelay の後にリリースが試行されます。maxAttempts に達すると、メッセージは破棄されます（リリースがトランザクションの場合を除き、その場合、メッセージはストアに残りますが、アプリケーションが再起動されるか、reschedulePersistedMessages() メソッドが呼び出されるまで、リリースのスケジュールは設定されません。上記のように）。
さらに、delayedMessageErrorChannel を構成できます。リリースが失敗すると、ErrorMessage がペイロードとして例外を使用してそのチャネルに送信され、originalMessage プロパティが設定されます。ErrorMessage には、現在のカウントを含むヘッダー IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT が含まれています。
エラーフローがエラーメッセージを消費して正常に終了した場合、それ以上のアクションは行われません。リリースがトランザクションの場合、トランザクションはコミットされ、メッセージはストアから削除されます。エラーフローが例外をスローした場合、リリースは上記のように maxAttempts まで再試行されます。
スクリプトのサポート
Spring Integration 2.1 は、Java バージョン 6 で導入された JSR223 Scripting for Java 仕様 (英語) のサポートを追加しました。これにより、サポートされている言語（Ruby、JRuby、Javascript、Groovy、Kotlin を含む）で記述されたスクリプトを使用して、Spring と同様のさまざまな統合コンポーネントのロジックを提供できます。式言語（SpEL）は Spring Integration で使用されます。JSR223 の詳細については、ドキュメント [Oracle] を参照してください。
この依存関係をプロジェクトに含める必要があります。
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-scripting</artifactId>
<version>5.3.0.RELEASE</version>
</dependency>compile "org.springframework.integration:spring-integration-scripting:5.3.0.RELEASE"さらに、スクリプトエンジンの実装を追加する必要があります。JRuby、Jython。
バージョン 5.2 以降、Spring Integration は Kotlin Jsr223 サポートを提供します。これらの依存関係をプロジェクトに追加して、プロジェクトを機能させる必要があります。
runtime 'org.jetbrains.kotlin:kotlin-script-util'
runtime 'org.jetbrains.kotlin:kotlin-compiler-embeddable'
runtime 'org.jetbrains.kotlin:kotlin-scripting-compiler-embeddable'KotlinScriptExecutor は、提供されている kotlin 言語インジケータによって選択されるか、スクリプトファイルに .kts 拡張子が付いています。
| この機能には Java 6 以上が必要であることに注意してください。 |
JVM スクリプト言語を使用するには、その言語の JSR223 実装をクラスパスに含める必要があります。Java 6 はネイティブに Javascript をサポートしています。Groovy (英語) および JRuby (英語) プロジェクトは、標準配布で JSR233 サポートを提供します。
| さまざまな JSR223 言語実装がサードパーティによって開発されています。特定の実装の Spring Integration との互換性は、仕様と実装者の仕様の解釈にどれだけ準拠しているかによって異なります。 |
| スクリプト言語として Groovy を使用する場合、Groovy に固有の追加機能を提供するため、Spring-Integration の Groovy サポートを使用することをお勧めします。ただし、このセクションも関連しています。 |
スクリプト構成
統合要件の複雑さに応じて、XML 構成の CDATA として、またはスクリプトを含む Spring リソースへの参照として、スクリプトをインラインで提供できます。スクリプトのサポートを有効にするために、Spring Integration は ScriptExecutingMessageProcessor を定義します。ScriptExecutingMessageProcessor は、メッセージペイロードを payload という名前の変数にバインドし、メッセージヘッダーを headers 変数にバインドします。どちらもスクリプト実行コンテキスト内でアクセスできます。必要なのは、これらの変数を使用するスクリプトを書くことだけです。次のペアの例は、フィルターを作成するサンプル構成を示しています。
<int:filter input-channel="referencedScriptInput">
<int-script:script location="some/path/to/ruby/script/RubyFilterTests.rb"/>
</int:filter>
<int:filter input-channel="inlineScriptInput">
<int-script:script lang="groovy">
<![CDATA[
return payload == 'good'
]]>
</int-script:script>
</int:filter> 上記の例が示すように、スクリプトはインラインで含めることも、リソースの場所を参照して含めることもできます（location 属性を使用）。さらに、lang 属性は言語名に対応します (またはその JSR223 エイリアス)
スクリプトをサポートする他の Spring Integration エンドポイント要素には router、service-activator、transformer、splitter が含まれます。それぞれの場合のスクリプト構成は、上記と同じです（エンドポイント要素を除く）。
スクリプトサポートのもう 1 つの便利な機能は、アプリケーションコンテキストを再起動せずにスクリプトを更新（再読み込み）できることです。これを行うには、次の例に示すように、script 要素で refresh-check-delay 属性を指定します。
<int-script:script location="..." refresh-check-delay="5000"/>上記の例では、スクリプトの場所は 5 秒ごとに更新が確認されます。スクリプトが更新された場合、更新から 5 秒より後に行われた呼び出しはすべて、新しいスクリプトを実行します。
次の例を考えてみましょう。
<int-script:script location="..." refresh-check-delay="0"/>前述の例では、スクリプトの変更が発生するとすぐにコンテキストが更新され、「リアルタイム」構成の簡単なメカニズムが提供されます。負の値は、アプリケーションコンテキストの初期化後にスクリプトがリロードされないことを意味します。これがデフォルトの動作です。次の例は、更新しないスクリプトを示しています。
<int-script:script location="..." refresh-check-delay="-1"/>| インラインスクリプトはリロードできません。 |
スクリプト変数のバインド
スクリプトの実行コンテキストに外部から提供される変数をスクリプトが参照できるようにするには、変数バインディングが必要です。デフォルトでは、payload および headers はバインディング変数として使用されます。次の例に示すように、<variable> 要素を使用して、追加の変数をスクリプトにバインドできます。
<script:script lang="js" location="foo/bar/MyScript.js">
<script:variable name="foo" value="thing1"/>
<script:variable name="bar" value="thing2"/>
<script:variable name="date" ref="date"/>
</script:script> 前の例に示すように、スクリプト変数をスカラー値または Spring Bean 参照にバインドできます。payload と headers は、まだバインディング変数として含まれていることに注意してください。
Spring Integration 3.0 では、variable 要素に加えて、variables 属性が導入されています。この属性と variable 要素は相互に排他的ではなく、1 つの script コンポーネント内で組み合わせることができます。ただし、変数は、どこに定義されているかに関係なく、一意である必要があります。また、Spring Integration 3.0 以降、次の例に示すように、インラインスクリプトでも変数バインディングを使用できます。
<service-activator input-channel="input">
<script:script lang="ruby" variables="thing1=THING1, date-ref=dateBean">
<script:variable name="thing2" ref="thing2Bean"/>
<script:variable name="thing3" value="thing2"/>
<![CDATA[
payload.foo = thing1
payload.date = date
payload.bar = thing2
payload.baz = thing3
payload
]]>
</script:script>
</service-activator> 上記の例は、インラインスクリプト、variable 要素、variables 属性の組み合わせを示しています。variables 属性にはコンマ区切り値が含まれ、各セグメントには変数とその値の "=" で区切られたペアが含まれます。前の例の date-ref 変数のように、変数名の末尾に -ref を付けることができます。つまり、バインディング変数の名前は date ですが、値はアプリケーションコンテキストからの dateBean Bean への参照です。これは、プロパティプレースホルダーの構成またはコマンドライン引数を使用する場合に便利です。
変数の生成方法をさらに制御する必要がある場合は、ScriptVariableGenerator 戦略を使用する独自の Java クラスを実装できます。この戦略は、次のインターフェースで定義されます。
public interface ScriptVariableGenerator {
Map<String, Object> generateScriptVariables(Message<?> message);
} このインターフェースでは、generateScriptVariables(Message) メソッドを実装する必要があります。メッセージ引数を使用すると、メッセージペイロードとヘッダーで使用可能なデータにアクセスでき、戻り値はバインドされた変数の Map です。このメソッドは、メッセージに対してスクリプトが実行されるたびに呼び出されます。次の例は、ScriptVariableGenerator の実装を提供し、script-variable-generator 属性でそれを参照する方法を示しています。
<int-script:script location="foo/bar/MyScript.groovy"
script-variable-generator="variableGenerator"/>
<bean id="variableGenerator" class="foo.bar.MyScriptVariableGenerator"/>script-variable-generator が提供されない場合、スクリプトコンポーネントは DefaultScriptVariableGenerator を使用します。DefaultScriptVariableGenerator は、提供された <variable> 要素を、generateScriptVariables(Message) メソッドで Message からの payload および headers 変数とマージします。
script-variable-generator 属性と <variable> 要素の両方を提供することはできません。それらは相互に排他的です。 |
Groovy サポート
Spring Integration 2.0 では、Groovy サポートを追加しました。これにより、Groovy スクリプト言語を使用して、さまざまな統合コンポーネントのロジックを提供できます。これは、ルーティング、変換、その他の統合に関する Spring 式言語（SpEL）のサポート方法と同様です。Groovy の詳細については、プロジェクトの Web サイト (英語) にある Groovy のドキュメントを参照してください。
この依存関係をプロジェクトに含める必要があります。
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-groovy</artifactId>
<version>5.3.0.RELEASE</version>
</dependency>compile "org.springframework.integration:spring-integration-groovy:5.3.0.RELEASE"Groovy の設定
Spring Integration 2.1 では、Groovy サポートの構成名前空間は Spring Integration のスクリプトサポートの拡張であり、スクリプトのサポートセクションで詳細に説明されているコア構成と動作を共有します。Groovy スクリプトは一般的なスクリプトサポートで十分にサポートされていますが、Groovy サポートは、Spring Framework の org.springframework.scripting.groovy.GroovyScriptFactory および関連コンポーネントによってサポートされる Groovy 構成名前空間を提供し、Groovy を使用するための拡張機能を提供します。次のリストは、2 つのサンプル構成を示しています。
<int:filter input-channel="referencedScriptInput">
<int-groovy:script location="some/path/to/groovy/file/GroovyFilterTests.groovy"/>
</int:filter>
<int:filter input-channel="inlineScriptInput">
<int-groovy:script><![CDATA[
return payload == 'good'
]]></int-groovy:script>
</int:filter> 上記の例が示すように、構成は一般的なスクリプトサポート構成と同じに見えます。唯一の違いは、int-groovy 名前空間プレフィックスによって示されるように、Groovy 名前空間の使用です。また、<script> タグの lang 属性は、このネームスペースでは無効であることに注意してください。
Groovy オブジェクトのカスタマイズ
Groovy オブジェクト自体を（変数の設定以外に）カスタマイズする必要がある場合は、customizer 属性を使用して、GroovyObjectCustomizer を実装する Bean を参照できます。例: これは、MetaClass を変更し、スクリプト内で使用できるように関数を登録することにより、ドメイン固有言語（DSL）を実装する場合に役立ちます。次の例は、そのメソッドを示しています。
<int:service-activator input-channel="groovyChannel">
<int-groovy:script location="somewhere/SomeScript.groovy" customizer="groovyCustomizer"/>
</int:service-activator>
<beans:bean id="groovyCustomizer" class="org.something.MyGroovyObjectCustomizer"/> カスタム GroovyObjectCustomizer の設定は、<variable> エレメントまたは script-variable-generator 属性と相互に排他的ではありません。インラインスクリプトを定義するときにも提供できます。
Spring Integration 3.0 には、variable 要素と連動する variables 属性が導入されました。また、groovy スクリプトには、バインディング変数に名前が指定されていない場合、変数を BeanFactory の Bean に解決する機能があります。次の例は、変数（entityManager）の使用方法を示しています。
<int-groovy:script>
<![CDATA[
entityManager.persist(payload)
payload
]]>
</int-groovy:script>entityManager は、アプリケーションコンテキストで適切な Bean でなければなりません。
<variable> 要素、variables 属性、script-variable-generator 属性に関する詳細については、スクリプト変数のバインドを参照してください。
Groovy スクリプトコンパイラーのカスタマイズ
@CompileStatic ヒントは、最も一般的な Groovy コンパイラーカスタマイズオプションです。クラスまたはメソッドレベルで使用できます。詳細については、Groovy 参考マニュアル (英語) 、特に @CompileStatic (英語) を参照してください。（統合シナリオで）短いスクリプトにこの機能を利用するには、単純なスクリプトをより Java に似たコードに変更する必要があります。次の <filter> スクリプトを検討してください。
headers.type == 'good'上記のスクリプトは、Spring Integration で次のメソッドになります。
@groovy.transform.CompileStatic
String filter(Map headers) {
headers.type == 'good'
}
filter(headers) これにより、filter() メソッドは静的な Java コードに変換およびコンパイルされ、getProperty() ファクトリや CallSite プロキシなどの Groovy 呼び出しの動的フェーズをバイパスします。
バージョン 4.3 以降では、@CompileStatic の ASTTransformationCustomizer を内部 CompilerConfiguration に追加するように指定して、compile-static boolean オプションを使用して Spring Integration Groovy コンポーネントを構成できます。これが適切な場所にあると、スクリプトコードで @CompileStatic を使用したメソッド宣言を省略しても、コンパイルされたプレーン Java コードを取得できます。この場合、次の例に示すように、前述のスクリプトは短くてもかまいませんが、解釈されたスクリプトよりも少し冗長である必要があります。
binding.variables.headers.type == 'good'@CompileStatic には動的 GroovyObject.getProperty() 機能がないため、groovy.lang.Script binding プロパティを介して headers および payload （またはその他の）変数にアクセスする必要があります。
さらに、compiler-configuration Bean リファレンスを導入しました。この属性を使用すると、ImportCustomizer など、他の必要な Groovy コンパイラーのカスタマイズを提供できます。この機能の詳細については、Groovy の高度なコンパイラー構成 (英語) に関する資料を参照してください。
compilerConfiguration を使用しても、@CompileStatic アノテーションに ASTTransformationCustomizer は自動的に追加されず、compileStatic オプションをオーバーライドします。それでも CompileStatic が必要な場合は、new ASTTransformationCustomizer(CompileStatic.class) をそのカスタム compilerConfiguration の CompilationCustomizers に手動で追加する必要があります。 |
Groovy コンパイラーのカスタマイズは refresh-check-delay オプションに影響を与えず、再ロード可能なスクリプトも静的にコンパイルできます。 |
制御バス
（エンタープライズ統合パターン (英語) ）に従って、コントロールバスの背景となる考え方は、「アプリケーションレベル」のメッセージングに使用されるのと同じメッセージングシステムをフレームワーク内のコンポーネントの監視および管理に使用できるということです。Spring Integration では、公開された操作を呼び出す手段としてメッセージを送信できるように、前述のアダプターに基づいて構築します。これらの操作の 1 つのオプションは Groovy スクリプトです。次の例では、制御バスの Groovy スクリプトを構成します。
<int-groovy:control-bus input-channel="operationChannel"/>制御バスには、アプリケーションコンテキストで Bean の操作を呼び出すためにアクセスできる入力チャネルがあります。
Groovy 制御バスは、Groovy スクリプトとして入力チャネルでメッセージを実行します。メッセージを受け取り、本文をスクリプトにコンパイルし、GroovyObjectCustomizer でカスタマイズして実行します。制御バスの MessageProcessor は、@ManagedResource アノテーションが付けられ、Spring の Lifecycle インターフェースを実装するか、Spring の CustomizableThreadCreator 基本クラス（たとえば、TaskExecutor および TaskScheduler 実装のいくつか）を継承するアプリケーションコンテキストのすべての Bean を公開します。
特に非同期メッセージフロー内で、コントロールバスコマンドスクリプトでカスタムスコープ（「リクエスト」など）でマネージド Bean を使用する場合は注意してください。制御バスの MessageProcessor がアプリケーションコンテキストから Bean を公開できない場合、コマンドスクリプトの実行中に BeansException が発生する可能性があります。例: カスタムスコープのコンテキストが確立されていない場合、そのスコープ内で Bean を取得しようとすると BeanCreationException がトリガーされます。 |
Groovy オブジェクトをさらにカスタマイズする必要がある場合は、次の例に示すように、customizer 属性を介して GroovyObjectCustomizer を実装する Bean への参照を提供することもできます。
<int-groovy:control-bus input-channel="input"
output-channel="output"
customizer="groovyCustomizer"/>
<beans:bean id="groovyCustomizer" class="org.foo.MyGroovyObjectCustomizer"/>エンドポイントへの動作の追加
Spring Integration 2.2 以前は、ポーラーの <advice-chain/> 要素に AOP アドバイスを追加することで、統合フロー全体に動作を追加できました。しかし、ダウンストリームエンドポイントではなく、REST Web サービスの呼び出しだけを再試行する場合を考えます。
例: 次のフローを検討します。
inbound-adapter->poller->http-gateway1->http-gateway2->jdbc-outbound-adapter ポーラーのアドバイスチェーンに一部の再試行ロジックを構成し、ネットワークグリッチのために http-gateway2 への呼び出しが失敗した場合、再試行により http-gateway1 と http-gateway2 の両方が再度呼び出されます。同様に、jdbc-outbound-adapter で一時的な障害が発生すると、両方の HTTP ゲートウェイがもう一度呼び出されてから、再び jdbc-outbound-adapter が呼び出されます。
Spring Integration 2.2 は、個々のエンドポイントに動作を追加する機能を追加します。これは、多くのエンドポイントに <request-handler-advice-chain/> 要素を追加することで実現されます。次の例は、outbound-gateway 内の <request-handler-advice-chain/> 要素の使用方法を示しています。
<int-http:outbound-gateway id="withAdvice"
url-expression="'http://localhost/test1'"
request-channel="requests"
reply-channel="nextChannel">
<int-http:request-handler-advice-chain>
<ref bean="myRetryAdvice" />
</int-http:request-handler-advice-chain>
</int-http:outbound-gateway> この場合、myRetryAdvice はこのゲートウェイにローカルにのみ適用され、応答が nextChannel に送信された後にダウンストリームで実行されるその他のアクションには適用されません。アドバイスの範囲はエンドポイント自体に限定されます。
現時点では、 ただし、 |
提供されたアドバイスクラス
Spring Integration は、AOP アドバイスクラスを適用する一般的なメカニズムを提供することに加えて、これらのすぐに使えるアドバイスの実装を提供します。
RequestHandlerRetryAdvice( 再試行のアドバイスで説明)RequestHandlerCircuitBreakerAdvice( サーキットブレーカーのアドバイスで説明)ExpressionEvaluatingRequestHandlerAdvice( 式評価アドバイスで説明)RateLimiterRequestHandlerAdvice( レートリミッターのアドバイスで説明)CacheRequestHandlerAdvice( キャッシングのアドバイスで説明)ReactiveRequestHandlerAdvice( リアクティブアドバイスで説明)
再試行のアドバイス
再試行のアドバイス（o.s.i.handler.advice.RequestHandlerRetryAdvice）は、Spring Retry [GitHub] (英語) プロジェクトによって提供される豊富な再試行メカニズムを活用します。spring-retry のコアコンポーネントは RetryTemplate です。これにより、RetryPolicy および BackoffPolicy 戦略（多数の実装を含む）、および再試行が尽きたときに実行するアクションを決定する RecoveryCallback 戦略を含む、洗練された再試行シナリオを構成できます。
- ステートレス再試行
ステートレス再試行は、再試行アクティビティがアドバイス内で完全に処理される場合です。スレッドは一時停止し（そうするように構成されている場合）、アクションを再試行します。
- ステートフルリトライ
ステートフル再試行は、再試行状態がアドバイス内で管理されているが、例外がスローされ、呼び出し元がリクエストを再送信する場合です。ステートフル再試行の例は、現在のスレッドで実行するのではなく、メッセージ呼び出し元（たとえば、JMS）に再送信を行わせたい場合です。ステートフル再試行には、再試行された送信を検出するための何らかのメカニズムが必要です。
spring-retry の詳細については、プロジェクトの Javadoc と、spring-retry の作成元である Spring Batch のリファレンスドキュメントを参照してください。
| デフォルトのバックオフ動作は、バックオフしないことです。再試行はすぐに試行されます。試行間でスレッドを一時停止させるバックオフポリシーを使用すると、過剰なメモリ使用やスレッド不足などのパフォーマンスの問題が発生する可能性があります。大量の環境では、バックオフポリシーを注意して使用する必要があります。 |
再試行アドバイスの構成
このセクションの例では、常に例外をスローする次の <service-activator> を使用しています。
public class FailingService {
public void service(String message) {
throw new RuntimeException("error");
}
}- 単純なステートレス再試行
デフォルトの
RetryTemplateには、3 回試行するSimpleRetryPolicyがあります。BackOffPolicyはないため、3 回の試行は連続して行われ、試行間の遅延はありません。RecoveryCallbackはないため、最終的な失敗した再試行が発生した後、呼び出し元に例外をスローします。Spring Integration 環境では、この最終的な例外は、受信エンドポイントでerror-channelを使用して処理される場合があります。次の例では、RetryTemplateを使用し、そのDEBUG出力を示しています。<int:service-activator input-channel="input" ref="failer" method="service"> <int:request-handler-advice-chain> <bean class="o.s.i.handler.advice.RequestHandlerRetryAdvice"/> </int:request-handler-advice-chain> </int:service-activator> DEBUG [task-scheduler-2]preSend on channel 'input', message: [Payload=...] DEBUG [task-scheduler-2]Retry: count=0 DEBUG [task-scheduler-2]Checking for rethrow: count=1 DEBUG [task-scheduler-2]Retry: count=1 DEBUG [task-scheduler-2]Checking for rethrow: count=2 DEBUG [task-scheduler-2]Retry: count=2 DEBUG [task-scheduler-2]Checking for rethrow: count=3 DEBUG [task-scheduler-2]Retry failed last attempt: count=3- 回復を伴う単純なステートレス再試行
次の例では、前の例に
RecoveryCallbackを追加し、ErrorMessageSendingRecovererを使用してErrorMessageをチャネルに送信します。<int:service-activator input-channel="input" ref="failer" method="service"> <int:request-handler-advice-chain> <bean class="o.s.i.handler.advice.RequestHandlerRetryAdvice"> <property name="recoveryCallback"> <bean class="o.s.i.handler.advice.ErrorMessageSendingRecoverer"> <constructor-arg ref="myErrorChannel" /> </bean> </property> </bean> </int:request-handler-advice-chain> </int:service-activator> DEBUG [task-scheduler-2]preSend on channel 'input', message: [Payload=...] DEBUG [task-scheduler-2]Retry: count=0 DEBUG [task-scheduler-2]Checking for rethrow: count=1 DEBUG [task-scheduler-2]Retry: count=1 DEBUG [task-scheduler-2]Checking for rethrow: count=2 DEBUG [task-scheduler-2]Retry: count=2 DEBUG [task-scheduler-2]Checking for rethrow: count=3 DEBUG [task-scheduler-2]Retry failed last attempt: count=3 DEBUG [task-scheduler-2]Sending ErrorMessage :failedMessage:[Payload=...]- カスタマイズされたポリシーを使用したステートレス再試行とリカバリ
より洗練されたものにするために、カスタマイズされた
RetryTemplateでアドバイスを提供できます。この例では、SimpleRetryPolicyを引き続き使用していますが、試行回数を 4 回に増やしています。また、最初の再試行が 1 秒間待機し、2 番目が 5 秒間待機し、3 番目が 25 待機するExponentialBackoffPolicyを追加します（合計 4 回の試行）。次のリストは、例とそのDEBUG出力を示しています。<int:service-activator input-channel="input" ref="failer" method="service"> <int:request-handler-advice-chain> <bean class="o.s.i.handler.advice.RequestHandlerRetryAdvice"> <property name="recoveryCallback"> <bean class="o.s.i.handler.advice.ErrorMessageSendingRecoverer"> <constructor-arg ref="myErrorChannel" /> </bean> </property> <property name="retryTemplate" ref="retryTemplate" /> </bean> </int:request-handler-advice-chain> </int:service-activator> <bean id="retryTemplate" class="org.springframework.retry.support.RetryTemplate"> <property name="retryPolicy"> <bean class="org.springframework.retry.policy.SimpleRetryPolicy"> <property name="maxAttempts" value="4" /> </bean> </property> <property name="backOffPolicy"> <bean class="org.springframework.retry.backoff.ExponentialBackOffPolicy"> <property name="initialInterval" value="1000" /> <property name="multiplier" value="5.0" /> <property name="maxInterval" value="60000" /> </bean> </property> </bean> 27.058 DEBUG [task-scheduler-1]preSend on channel 'input', message: [Payload=...] 27.071 DEBUG [task-scheduler-1]Retry: count=0 27.080 DEBUG [task-scheduler-1]Sleeping for 1000 28.081 DEBUG [task-scheduler-1]Checking for rethrow: count=1 28.081 DEBUG [task-scheduler-1]Retry: count=1 28.081 DEBUG [task-scheduler-1]Sleeping for 5000 33.082 DEBUG [task-scheduler-1]Checking for rethrow: count=2 33.082 DEBUG [task-scheduler-1]Retry: count=2 33.083 DEBUG [task-scheduler-1]Sleeping for 25000 58.083 DEBUG [task-scheduler-1]Checking for rethrow: count=3 58.083 DEBUG [task-scheduler-1]Retry: count=3 58.084 DEBUG [task-scheduler-1]Checking for rethrow: count=4 58.084 DEBUG [task-scheduler-1]Retry failed last attempt: count=4 58.086 DEBUG [task-scheduler-1]Sending ErrorMessage :failedMessage:[Payload=...]- ステートレス再試行のネームスペースサポート
バージョン 4.0 からは、次の例に示すように、再試行のアドバイスに対する名前空間のサポートのおかげで、前述の構成を大幅に簡素化できます。
<int:service-activator input-channel="input" ref="failer" method="service"> <int:request-handler-advice-chain> <ref bean="retrier" /> </int:request-handler-advice-chain> </int:service-activator> <int:handler-retry-advice id="retrier" max-attempts="4" recovery-channel="myErrorChannel"> <int:exponential-back-off initial="1000" multiplier="5.0" maximum="60000" /> </int:handler-retry-advice>上記の例では、アドバイスはトップレベルの Bean として定義されているため、複数の
request-handler-advice-chainインスタンスで使用できます。次の例に示すように、チェーン内でアドバイスを直接定義することもできます。<int:service-activator input-channel="input" ref="failer" method="service"> <int:request-handler-advice-chain> <int:retry-advice id="retrier" max-attempts="4" recovery-channel="myErrorChannel"> <int:exponential-back-off initial="1000" multiplier="5.0" maximum="60000" /> </int:retry-advice> </int:request-handler-advice-chain> </int:service-activator><handler-retry-advice>は、<fixed-back-off>または<exponential-back-off>の子要素を持つことも、子要素を持たないこともできます。子要素のない<handler-retry-advice>は、バックオフを使用しません。recovery-channelがない場合、再試行が使い果たされると例外がスローされます。名前空間は、ステートレス再試行でのみ使用できます。より複雑な環境（カスタムポリシーなど）では、通常の
<bean>定義を使用します。- 回復を伴う単純なステートフル再試行
再試行をステートフルにするには、
RetryStateGenerator実装でアドバイスを提供する必要があります。このクラスは、RetryTemplateがこのメッセージの再試行の現在の状態を判断できるように、メッセージを再送信として識別するために使用されます。フレームワークは、SpelExpressionRetryStateGeneratorを提供します。SpelExpressionRetryStateGeneratorは、SpEL 式を使用してメッセージ識別子を決定します。この例でもデフォルトのポリシー（バックオフなしの 3 回の試行）を使用しています。ステートレス再試行と同様に、これらのポリシーはカスタマイズできます。次のリストは、例とそのDEBUG出力を示しています。<int:service-activator input-channel="input" ref="failer" method="service"> <int:request-handler-advice-chain> <bean class="o.s.i.handler.advice.RequestHandlerRetryAdvice"> <property name="retryStateGenerator"> <bean class="o.s.i.handler.advice.SpelExpressionRetryStateGenerator"> <constructor-arg value="headers['jms_messageId']" /> </bean> </property> <property name="recoveryCallback"> <bean class="o.s.i.handler.advice.ErrorMessageSendingRecoverer"> <constructor-arg ref="myErrorChannel" /> </bean> </property> </bean> </int:request-handler-advice-chain> </int:service-activator> 24.351 DEBUG [Container#0-1]preSend on channel 'input', message: [Payload=...] 24.368 DEBUG [Container#0-1]Retry: count=0 24.387 DEBUG [Container#0-1]Checking for rethrow: count=1 24.387 DEBUG [Container#0-1]Rethrow in retry for policy: count=1 24.387 WARN [Container#0-1]failure occurred in gateway sendAndReceive org.springframework.integration.MessagingException: Failed to invoke handler ... Caused by: java.lang.RuntimeException: foo ... 24.391 DEBUG [Container#0-1]Initiating transaction rollback on application exception ... 25.412 DEBUG [Container#0-1]preSend on channel 'input', message: [Payload=...] 25.412 DEBUG [Container#0-1]Retry: count=1 25.413 DEBUG [Container#0-1]Checking for rethrow: count=2 25.413 DEBUG [Container#0-1]Rethrow in retry for policy: count=2 25.413 WARN [Container#0-1]failure occurred in gateway sendAndReceive org.springframework.integration.MessagingException: Failed to invoke handler ... Caused by: java.lang.RuntimeException: foo ... 25.414 DEBUG [Container#0-1]Initiating transaction rollback on application exception ... 26.418 DEBUG [Container#0-1]preSend on channel 'input', message: [Payload=...] 26.418 DEBUG [Container#0-1]Retry: count=2 26.419 DEBUG [Container#0-1]Checking for rethrow: count=3 26.419 DEBUG [Container#0-1]Rethrow in retry for policy: count=3 26.419 WARN [Container#0-1]failure occurred in gateway sendAndReceive org.springframework.integration.MessagingException: Failed to invoke handler ... Caused by: java.lang.RuntimeException: foo ... 26.420 DEBUG [Container#0-1]Initiating transaction rollback on application exception ... 27.425 DEBUG [Container#0-1]preSend on channel 'input', message: [Payload=...] 27.426 DEBUG [Container#0-1]Retry failed last attempt: count=3 27.426 DEBUG [Container#0-1]Sending ErrorMessage :failedMessage:[Payload=...]前述の例をステートレスの例と比較すると、ステートフルリトライでは、失敗のたびに呼び出し元に例外がスローされることがわかります。
- 再試行の例外分類
Spring Retry には、どの例外が再試行を呼び出すことができるかを決定するための柔軟性があります。デフォルトの構成ではすべての例外が再試行され、例外分類子は最上位の例外を調べます。たとえば、
MyExceptionでのみ再試行するように構成し、アプリケーションが原因がMyExceptionであるSomeOtherExceptionをスローする場合、再試行は発生しません。Spring Retry 1.0.3 以降、
BinaryExceptionClassifierにはtraverseCauses（デフォルトはfalse）と呼ばれるプロパティがあります。trueの場合、一致が見つかるか、トラバースする原因がなくなるまで、例外の原因をトラバースします。再試行にこの分類子を使用するには、最大試行回数、
ExceptionオブジェクトのMap、traverseCausesブール値を取るコンストラクターで作成されたSimpleRetryPolicyを使用します。その後、このポリシーをRetryTemplateに注入できます。
ユーザー例外は MessagingException にラップされる可能性があるため、この場合は traverseCauses が必要です。 |
サーキットブレーカーのアドバイス
サーキットブレーカーパターンの一般的な考え方は、サービスが現在利用できない場合、それを使用しようとして時間（およびリソース）を浪費しないことです。o.s.i.handler.advice.RequestHandlerCircuitBreakerAdvice はこのパターンを実装しています。サーキットブレーカーが閉じた状態の場合、エンドポイントはサービスの呼び出しを試みます。サーキットブレーカーは、一定回数の連続試行が失敗するとオープン状態になります。オープン状態の場合、新しいリクエストは「高速で失敗」し、時間が経過するまでサービスの呼び出しは試行されません。
その時間が経過すると、サーキットブレーカーは半開状態に設定されます。この状態で、1 回の試行でも失敗すると、ブレーカーはすぐにオープン状態になります。試行が成功すると、ブレーカーは閉じた状態になり、その場合、構成された数の連続した障害が再び発生するまで、再び開いた状態になりません。正常に試行されると、ブレーカーが再びオープン状態になるタイミングを判別するために、状態が障害ゼロにリセットされます。
通常、このアドバイスは外部サービスに使用される可能性があり、失敗するまでに時間がかかる場合があります（ネットワーク接続を試行するタイムアウトなど）。
RequestHandlerCircuitBreakerAdvice には、threshold と halfOpenAfter の 2 つのプロパティがあります。threshold プロパティは、ブレーカーが開くまでに発生する必要のある連続した障害の数を表します。デフォルトは 5 です。halfOpenAfter プロパティは、最後の失敗後、ブレーカーが別のリクエストを試みる前に待機する時間を表します。デフォルトは 1000 ミリ秒です。
次の例では、サーキットブレーカーを構成し、その DEBUG および ERROR 出力を示しています。
<int:service-activator input-channel="input" ref="failer" method="service">
<int:request-handler-advice-chain>
<bean class="o.s.i.handler.advice.RequestHandlerCircuitBreakerAdvice">
<property name="threshold" value="2" />
<property name="halfOpenAfter" value="12000" />
</bean>
</int:request-handler-advice-chain>
</int:service-activator>
05.617 DEBUG [task-scheduler-1]preSend on channel 'input', message: [Payload=...]
05.638 ERROR [task-scheduler-1]org.springframework.messaging.MessageHandlingException: java.lang.RuntimeException: foo
...
10.598 DEBUG [task-scheduler-2]preSend on channel 'input', message: [Payload=...]
10.600 ERROR [task-scheduler-2]org.springframework.messaging.MessageHandlingException: java.lang.RuntimeException: foo
...
15.598 DEBUG [task-scheduler-3]preSend on channel 'input', message: [Payload=...]
15.599 ERROR [task-scheduler-3]org.springframework.messaging.MessagingException: Circuit Breaker is Open for ServiceActivator
...
20.598 DEBUG [task-scheduler-2]preSend on channel 'input', message: [Payload=...]
20.598 ERROR [task-scheduler-2]org.springframework.messaging.MessagingException: Circuit Breaker is Open for ServiceActivator
...
25.598 DEBUG [task-scheduler-5]preSend on channel 'input', message: [Payload=...]
25.601 ERROR [task-scheduler-5]org.springframework.messaging.MessageHandlingException: java.lang.RuntimeException: foo
...
30.598 DEBUG [task-scheduler-1]preSend on channel 'input', message: [Payload=foo...]
30.599 ERROR [task-scheduler-1]org.springframework.messaging.MessagingException: Circuit Breaker is Open for ServiceActivator 上記の例では、しきい値は 2 に設定され、halfOpenAfter は 12 秒に設定されています。5 秒ごとに新しいリクエストが届きます。最初の 2 回の試行でサービスが呼び出されました。3 番目と 4 番目は、サーキットブレーカーが開いていることを示す例外で失敗しました。5 番目のリクエストは、最後の失敗から 15 秒後にリクエストされたために試行されました。ブレーカーがすぐに開いたため、6 回目の試行はすぐに失敗します。
式評価アドバイス
最終的に提供されるアドバイスクラスは o.s.i.handler.advice.ExpressionEvaluatingRequestHandlerAdvice です。このアドバイスは、他の 2 つのアドバイスよりも一般的です。エンドポイントに送信された元の受信メッセージの式を評価するメカニズムを提供します。成功または失敗の後、評価するために個別の式を使用できます。オプションで、入力メッセージとともに評価結果を含むメッセージをメッセージチャネルに送信できます。
このアドバイスの一般的な使用例は、<ftp:outbound-channel-adapter/> を使用することです。おそらく、転送が成功した場合はファイルを 1 つのディレクトリに、失敗した場合は別のディレクトリに移動します。
アドバイスには、成功した場合の式、失敗した場合の式、それぞれに対応するチャネルを設定するプロパティがあります。成功した場合、successChannel に送信されるメッセージは AdviceMessage であり、ペイロードは式の評価の結果です。inputMessage という追加のプロパティには、ハンドラーに送信された元のメッセージが含まれています。failureChannel に送信されるメッセージ（ハンドラーが例外をスローする場合）は、MessageHandlingExpressionEvaluatingAdviceException のペイロードを持つ ErrorMessage です。すべての MessagingException インスタンスと同様に、このペイロードには failedMessage および cause プロパティと、式評価の結果を含む evaluationResult という追加のプロパティがあります。
バージョン 5.1.3 以降、チャネルは構成されているが式が提供されていない場合、メッセージの payload を評価するためにデフォルトの式が使用されます。 |
アドバイスのスコープ内で例外がスローされると、デフォルトでは、failureExpression が評価された後にその例外が呼び出し元にスローされます。例外のスローを抑制したい場合は、trapException プロパティを true に設定します。次のアドバイスは、Java DSL でアドバイスを設定する方法を示しています。
@SpringBootApplication
public class EerhaApplication {
public static void main(String[] args) {
ConfigurableApplicationContext context = SpringApplication.run(EerhaApplication.class, args);
MessageChannel in = context.getBean("advised.input", MessageChannel.class);
in.send(new GenericMessage<>("good"));
in.send(new GenericMessage<>("bad"));
context.close();
}
@Bean
public IntegrationFlow advised() {
return f -> f.handle((GenericHandler<String>) (payload, headers) -> {
if (payload.equals("good")) {
return null;
}
else {
throw new RuntimeException("some failure");
}
}, c -> c.advice(expressionAdvice()));
}
@Bean
public Advice expressionAdvice() {
ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
advice.setSuccessChannelName("success.input");
advice.setOnSuccessExpressionString("payload + ' was successful'");
advice.setFailureChannelName("failure.input");
advice.setOnFailureExpressionString(
"payload + ' was bad, with reason: ' + #exception.cause.message");
advice.setTrapException(true);
return advice;
}
@Bean
public IntegrationFlow success() {
return f -> f.handle(System.out::println);
}
@Bean
public IntegrationFlow failure() {
return f -> f.handle(System.out::println);
}
}レートリミッターのアドバイス
レートリミッターのアドバイス（RateLimiterRequestHandlerAdvice）により、エンドポイントがリクエストでオーバーロードにならないようにすることができます。レート制限に違反すると、リクエストはブロックされた状態になります。
このアドバイスの典型的な使用例は、外部サービスプロバイダーが 1 分あたり n を超える数のリクエストを許可しない場合です。
RateLimiterRequestHandlerAdvice の実装は、Resilience4j [GitHub] (英語) プロジェクトに完全に基づいており、RateLimiter または RateLimiterConfig のいずれかの注入が必要です。デフォルトおよび / またはカスタム名で構成することもできます。
次の例では、1 秒ごとに 1 つのリクエストでレートリミッターのアドバイスを構成しています。
@Bean
public RateLimiterRequestHandlerAdvice rateLimiterRequestHandlerAdvice() {
return new RateLimiterRequestHandlerAdvice(RateLimiterConfig.custom()
.limitRefreshPeriod(Duration.ofSeconds(1))
.limitForPeriod(1)
.build());
}
@ServiceActivator(inputChannel = "requestChannel", outputChannel = "resultChannel",
adviceChain = "rateLimiterRequestHandlerAdvice")
public String handleRequest(String payload) {
...
}キャッシングのアドバイス
バージョン 5.2 から、CacheRequestHandlerAdvice が導入されました。これは、Spring Framework のキャッシング抽象化に基づいており、@Caching アノテーションファミリーによって提供される概念と機能に沿っています。内部のロジックは CacheAspectSupport 拡張に基づいており、キャッシュ操作のプロキシは、リクエスト Message<?> を引数として AbstractReplyProducingMessageHandler.RequestHandler.handleRequestMessage メソッドを中心に実行されます。このアドバイスは、キャッシュキーを評価するために SpEL 式または Function を使用して構成できます。リクエスト Message<?> は、SpEL 評価コンテキストのルートオブジェクトとして、または Function 入力引数として使用できます。デフォルトでは、リクエストメッセージの payload がキャッシュキーに使用されます。デフォルトのキャッシュ操作が CacheableOperation の場合、CacheRequestHandlerAdvice は cacheNames で構成するか、任意の CacheOperation のセットで構成する必要があります。すべての CacheOperation は個別に構成することも、CacheManager、CacheResolver、CacheErrorHandler などの共有オプションを使用して CacheRequestHandlerAdvice 構成から再利用することもできます。この構成機能は、Spring Framework の @CacheConfig と @Caching アノテーションの組み合わせに似ています。CacheManager が提供されていない場合、単一の Bean がデフォルトで CacheAspectSupport の BeanFactory から解決されます。
次の例では、キャッシュ操作の異なるセットで 2 つのアドバイスを構成します。
@Bean
public CacheRequestHandlerAdvice cacheAdvice() {
CacheRequestHandlerAdvice cacheRequestHandlerAdvice = new CacheRequestHandlerAdvice(TEST_CACHE);
cacheRequestHandlerAdvice.setKeyExpressionString("payload");
return cacheRequestHandlerAdvice;
}
@Transformer(inputChannel = "transformerChannel", outputChannel = "nullChannel", adviceChain = "cacheAdvice")
public Object transform(Message<?> message) {
...
}
@Bean
public CacheRequestHandlerAdvice cachePutAndEvictAdvice() {
CacheRequestHandlerAdvice cacheRequestHandlerAdvice = new CacheRequestHandlerAdvice();
cacheRequestHandlerAdvice.setKeyExpressionString("payload");
CachePutOperation.Builder cachePutBuilder = new CachePutOperation.Builder();
cachePutBuilder.setCacheName(TEST_PUT_CACHE);
CacheEvictOperation.Builder cacheEvictBuilder = new CacheEvictOperation.Builder();
cacheEvictBuilder.setCacheName(TEST_CACHE);
cacheRequestHandlerAdvice.setCacheOperations(cachePutBuilder.build(), cacheEvictBuilder.build());
return cacheRequestHandlerAdvice;
}
@ServiceActivator(inputChannel = "serviceChannel", outputChannel = "nullChannel",
adviceChain = "cachePutAndEvictAdvice")
public Message<?> service(Message<?> message) {
...
}リアクティブアドバイス
バージョン 5.3 以降、ReactiveRequestHandlerAdvice は、Mono レスポンスを生成するリクエストメッセージハンドラーに使用できます。このアドバイスには BiFunction<Message<?>, Mono<?>, Publisher<?>> を提供する必要があり、インターセプトされた handleRequestMessage() メソッドの実装によって生成されたレスポンスで Mono.transform() オペレーターから呼び出されます。通常、このような Mono のカスタマイズは、timeout()、retry()、同様のサポートオペレーターを介してネットワークの変動を制御する場合に必要です。たとえば、WebFlux クライアントを介して HTTP リクエストを実行できる場合、以下の構成を使用して、5 秒を超えてレスポンスを待たないようにすることができます。
.handle(WebFlux.outboundGateway("https://somehost/"),
e -> e.customizeMonoReply((message, mono) -> mono.timeout(Duration.ofSeconds(5))));message 引数はメッセージハンドラーのリクエストメッセージであり、リクエストスコープの属性を決定するために使用できます。mono 引数は、このメッセージハンドラーの handleRequestMessage() メソッド実装の結果です。この関数からネストされた Mono.transform() を呼び出して、たとえばリアクティブサーキットブレーカーを適用することもできます。
カスタムアドバイスクラス
前述のアドバイスクラスに加えて、独自のアドバイスクラスを実装できます。org.aopalliance.aop.Advice （通常は org.aopalliance.intercept.MethodInterceptor）の実装を提供できますが、一般的に o.s.i.handler.advice.AbstractRequestHandlerAdvice をサブクラス化することをお勧めします。これには、低レベルのアスペクト指向プログラミングコードの記述を避け、この環境での使用に合わせて特別に調整された開始点を提供するという利点があります。
サブクラスは doInvoke() メソッドを実装する必要があり、その定義は次のとおりです。
/**
* Subclasses implement this method to apply behavior to the {@link MessageHandler} callback.execute()
* invokes the handler method and returns its result, or null).
* @param callback Subclasses invoke the execute() method on this interface to invoke the handler method.
* @param target The target handler.
* @param message The message that will be sent to the handler.
* @return the result after invoking the {@link MessageHandler}.
* @throws Exception
*/
protected abstract Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception; コールバックパラメーターは、AOP を直接扱うサブクラスを回避するのに便利です。callback.execute() メソッドを呼び出すと、メッセージハンドラーが呼び出されます。
target パラメーターは、特定のハンドラーの状態を維持する必要があるサブクラスに提供されます。おそらく、ターゲットによってキー設定された Map でその状態を維持することによってです。この機能により、同じアドバイスを複数のハンドラーに適用できます。RequestHandlerCircuitBreakerAdvice は、アドバイスを使用して、各ハンドラーのサーキットブレーカーの状態を維持します。
message パラメーターは、ハンドラーに送信されるメッセージです。アドバイスは、ハンドラーを呼び出す前にメッセージを変更できませんが、ペイロードを変更できます（可変プロパティがある場合）。通常、アドバイスでは、メッセージをロギングに使用したり、ハンドラーの呼び出しの前後にメッセージのコピーを送信したりします。
通常、戻り値は callback.execute() によって返される値です。ただし、アドバイスには戻り値を変更する機能があります。AbstractReplyProducingMessageHandler インスタンスのみが値を返すことに注意してください。次の例は、AbstractRequestHandlerAdvice を継承するカスタムアドバイスクラスを示しています。
public class MyAdvice extends AbstractRequestHandlerAdvice {
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
// add code before the invocation
Object result = callback.execute();
// add code after the invocation
return result;
}
}
詳細については、ReflectiveMethodInvocation (Javadoc) Javadoc を参照してください。 |
メッセージアドバイスの処理
このセクションの導入部分で説明したように、リクエストハンドラーアドバイスチェーン内のアドバイスオブジェクトは、ダウンストリームフロー (存在する場合) ではなく、現在のエンドポイントにのみ適用されます。応答を生成する MessageHandler オブジェクト ( AbstractReplyProducingMessageHandler を継承するオブジェクトなど) の場合、アドバイスは内部メソッド handleRequestMessage() ( MessageHandler.handleMessage() から呼び出される) に適用されます。他のメッセージハンドラーの場合、アドバイスは MessageHandler.handleMessage() に適用されます。
メッセージハンドラーが AbstractReplyProducingMessageHandler であっても、handleMessage メソッドにアドバイスを適用する必要がある状況がいくつかあります。例: べき等レシーバーは null を返す場合があり、ハンドラーの replyRequired プロパティが true に設定されている場合、例外が発生します。別の例は BoundRabbitChannelAdvice です - 厳密なメッセージ順序を参照してください。
バージョン 4.3.1 から、新しい HandleMessageAdvice インターフェースとその基本実装（AbstractHandleMessageAdvice）が導入されました。HandleMessageAdvice を実装する Advice オブジェクトは、ハンドラー型に関係なく、常に handleMessage() メソッドに適用されます。
HandleMessageAdvice 実装（べき等レシーバーなど）は、レスポンスを返すハンドラーに適用されると、adviceChain から分離され、MessageHandler.handleMessage() メソッドに適切に適用されることを理解することが重要です。
| この関連付けが解除されているため、アドバイスチェーンの順序は尊重されません。 |
次の構成を検討してください。
<some-reply-producing-endpoint ... >
<int:request-handler-advice-chain>
<tx:advice ... />
<ref bean="myHandleMessageAdvice" />
</int:request-handler-advice-chain>
</some-reply-producing-endpoint> 上記の例では、<tx:advice> は AbstractReplyProducingMessageHandler.handleRequestMessage() に適用されます。ただし、myHandleMessageAdvice は MessageHandler.handleMessage() に適用されます。<tx:advice> の前に呼び出されます。順序を保持するには、標準の Spring AOP 構成アプローチに従って、エンドポイント id を .handler サフィックスとともに使用して、ターゲット MessageHandler Bean を取得する必要があります。その場合、ダウンストリームフロー全体がトランザクションスコープ内にあることに注意してください。
レスポンスを返さない MessageHandler の場合、アドバイスチェーンの順序は保持されます。
バージョン 5.3 以降、HandleMessageAdviceAdapter は、MessageHandler.handleMessage() に既存の MethodInterceptor を適用できるようにするため、サブフロー全体に適用されます。たとえば、RetryOperationsInterceptor は、一部のエンドポイントから開始するサブフロー全体に適用できます。これは、コンシューマーエンドポイントが AbstractReplyProducingMessageHandler.RequestHandler.handleRequestMessage() に対してのみアドバイスを適用するため、デフォルトでは不可能です。バージョン 5.3 以降、HandleMessageAdviceAdapter は、MessageHandler.handleMessage() メソッドに MethodInterceptor を適用するために提供されているため、サブフロー全体に適用されます。例: RetryOperationsInterceptor は、いくつかのエンドポイントから始まるサブフロー全体に適用できます。コンシューマーエンドポイントは AbstractReplyProducingMessageHandler.RequestHandler.handleRequestMessage() にのみアドバイスを適用するため、これはデフォルトでは不可能です。
トランザクションサポート
バージョン 5.0 から、HandleMessageAdvice 実装のおかげで、ダウンストリームフロー全体をトランザクション化するために、新しい TransactionHandleMessageAdvice が導入されました。<request-handler-advice-chain> 要素で通常の TransactionInterceptor が使用される場合（たとえば、<tx:advice> の構成を通じて）、開始されたトランザクションは内部 AbstractReplyProducingMessageHandler.handleRequestMessage() にのみ適用され、ダウンストリームフローに伝搬されません。
XML 構成を簡素化するために、<request-handler-advice-chain> とともに、<transactional> 要素がすべての <outbound-gateway> および <service-activator> および関連コンポーネントに追加されました。次の例は、使用中の <transactional> を示しています。
<int-rmi:outbound-gateway remote-channel="foo" host="localhost"
request-channel="good" reply-channel="reply" port="#{@port}">
<int-rmi:transactional/>
</int-rmi:outbound-gateway>
<bean id="transactionManager" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.transaction.TransactionManager"/>
</bean>JPA 統合コンポーネントに精通している場合、このような構成は新しいものではありませんが、<poller> または JMS などのメッセージ駆動型チャネルアダプターだけでなく、フローの任意のポイントからトランザクションを開始できます。
次の例に示すように、Java 構成は TransactionInterceptorBuilder を使用することで簡素化でき、結果の Bean 名をメッセージングアノテーション adviceChain 属性で使用できます。
@Bean
public ConcurrentMetadataStore store() {
return new SimpleMetadataStore(hazelcastInstance()
.getMap("idempotentReceiverMetadataStore"));
}
@Bean
public IdempotentReceiverInterceptor idempotentReceiverInterceptor() {
return new IdempotentReceiverInterceptor(
new MetadataStoreSelector(
message -> message.getPayload().toString(),
message -> message.getPayload().toString().toUpperCase(), store()));
}
@Bean
public TransactionInterceptor transactionInterceptor() {
return new TransactionInterceptorBuilder(true)
.transactionManager(this.transactionManager)
.isolation(Isolation.READ_COMMITTED)
.propagation(Propagation.REQUIRES_NEW)
.build();
}
@Bean
@org.springframework.integration.annotation.Transformer(inputChannel = "input",
outputChannel = "output",
adviceChain = { "idempotentReceiverInterceptor",
"transactionInterceptor" })
public Transformer transformer() {
return message -> message;
}TransactionInterceptorBuilder コンストラクターの true パラメーターに注意してください。通常の TransactionInterceptor ではなく、TransactionHandleMessageAdvice が作成されます。
Java DSL は、次の例に示すように、エンドポイント構成の .transactional() オプションを介して Advice をサポートします。
@Bean
public IntegrationFlow updatingGatewayFlow() {
return f -> f
.handle(Jpa.updatingGateway(this.entityManagerFactory),
e -> e.transactional(true))
.channel(c -> c.queue("persistResults"));
}アドバイスフィルター
Filter アドバイスを助言する際には、追加の考慮事項があります。デフォルトでは、破棄アクション（フィルターが false を返す場合）は、アドバイスチェーンのスコープ内で実行されます。これには、廃棄チャネルの下流のすべてのフローが含まれる場合があります。たとえば、破棄チャネルの下流の要素が例外をスローし、再試行のアドバイスがある場合、プロセスは再試行されます。また、throwExceptionOnRejection が true に設定されている場合（アドバイスの範囲内で例外がスローされます）。
discard-within-advice を false に設定すると、この動作が変更され、破棄（または例外）が発生します。after the アドバイスチェーンが呼び出されます。
アノテーションを使用したエンドポイントへのアドバイス
アノテーション（@Filter、@ServiceActivator、@Splitter、@Transformer）を使用して特定のエンドポイントを構成する場合、adviceChain 属性でアドバイスチェーンの Bean 名を指定できます。さらに、@Filter アノテーションには discardWithinAdvice 属性もあります。これは、アドバイスフィルターに従って、破棄動作を構成するために使用できます。次の例では、廃棄が after the アドバイスで実行されます。
@MessageEndpoint
public class MyAdvisedFilter {
@Filter(inputChannel="input", outputChannel="output",
adviceChain="adviceChain", discardWithinAdvice="false")
public boolean filter(String s) {
return s.contains("good");
}
}アドバイスチェーン内のアドバイスのオーダー
アドバイスクラスは「アラウンド」アドバイスであり、ネストされた方法で適用されます。最初のアドバイスが最も外側で、最後のアドバイスが最も内側（つまり、アドバイスされているハンドラーに最も近い）です。アドバイスクラスを正しい順序で配置して、必要な機能を実現することが重要です。
例: 再試行のアドバイスとトランザクションのアドバイスを追加するとします。再試行アドバイスアドバイスを最初に配置し、その後にトランザクションアドバイスを配置することができます。各再試行は新しいトランザクションで実行されます。一方、すべての試行と回復操作（再試行 RecoveryCallback 内）をトランザクション内でスコープする場合、トランザクションアドバイスを最初に置くことができます。
推奨されるハンドラープロパティ
アドバイス内からハンドラープロパティにアクセスすると便利な場合があります。例: ほとんどのハンドラーは NamedComponent を実装して、コンポーネント名にアクセスできるようにします。
ターゲットオブジェクトには、target 引数（AbstractRequestHandlerAdvice をサブクラス化する場合）または invocation.getThis() （org.aopalliance.intercept.MethodInterceptor を実装する場合）を介してアクセスできます。
ハンドラー全体がアドバイスされる場合（ハンドラーが応答を生成しない場合やアドバイスが HandleMessageAdvice を実装する場合など）、次の例に示すように、ターゲットオブジェクトを NamedComponent などのインターフェースにキャストできます。
String componentName = ((NamedComponent) target).getComponentName();MethodInterceptor を直接実装する場合、次のようにターゲットオブジェクトをキャストできます。
String componentName = ((NamedComponent) invocation.getThis()).getComponentName();handleRequestMessage() メソッドのみが推奨される場合（応答生成ハンドラーで）、ハンドラー全体（AbstractReplyProducingMessageHandler）にアクセスする必要があります。次の例は、その方法を示しています。
AbstractReplyProducingMessageHandler handler =
((AbstractReplyProducingMessageHandler.RequestHandler) target).getAdvisedHandler();
String componentName = handler.getComponentName();べき等レシーバーエンタープライズ統合パターン
バージョン 4.1 から、Spring Integration はべき等レシーバー (英語) エンタープライズ統合パターンの実装を提供します。これは関数パターンであり、べき等性ロジック全体をアプリケーションに実装する必要があります。ただし、意思決定を簡素化するために、IdempotentReceiverInterceptor コンポーネントが提供されています。これは、MessageHandler.handleMessage() メソッドに適用される AOP Advice であり、構成に応じてリクエストメッセージを filter するか、duplicate としてマークすることができます。
以前は、たとえば <filter/> でカスタム MessageSelector を使用してこのパターンを実装できました（フィルターを参照）。ただし、このパターンはエンドポイント自体ではなく、エンドポイントの動作を実際に定義するため、べき等レシーバーの実装はエンドポイントコンポーネントを提供しません。むしろ、アプリケーションで宣言されたエンドポイントに適用されます。
IdempotentReceiverInterceptor のロジックは、提供された MessageSelector に基づいており、メッセージがそのセレクターで受け入れられない場合、true に設定された duplicateMessage ヘッダーで強化されます。ターゲット MessageHandler （またはダウンストリームフロー）は、このヘッダーを参照して正しいべき等性ロジックを実装できます。IdempotentReceiverInterceptor が discardChannel または throwExceptionOnRejection = true で構成されている場合、複製メッセージはターゲット MessageHandler.handleMessage() に送信されません。むしろ、破棄されます。重複したメッセージを破棄する（何もしない）場合は、discardChannel を、デフォルトの nullChannel Bean などの NullChannel で構成する必要があります。
メッセージ間の状態を維持し、べき等性についてメッセージを比較する機能を提供するために、MetadataStoreSelector を提供します。MessageProcessor 実装（Message に基づいてルックアップキーを作成）とオプションの ConcurrentMetadataStore （メタデータストア）を受け入れます。詳細については、MetadataStoreSelector Javadoc を参照してください。追加の MessageProcessor を使用して、ConcurrentMetadataStore 用に value をカスタマイズすることもできます。デフォルトでは、MetadataStoreSelector は timestamp メッセージヘッダーを使用します。
通常、キーに既存の値がない場合、セレクターは受け入れのためにメッセージを選択します。場合によっては、キーの現在の値と新しい値を比較して、メッセージを受け入れる必要があるかどうかを判断すると便利です。バージョン 5.3 以降、BiPredicate<String, String> を参照する compareValues プロパティが提供されています。最初のパラメーターは古い値です。true を返してメッセージを受け入れ、MetadataStore の古い値を新しい値に置き換えます。これは、キーの数を減らすのに役立ちます。たとえば、ファイル内の行を処理する場合、ファイル名をキーに格納し、現在の行番号を値に格納できます。その後、再起動後、すでに処理された行をスキップできます。例については、分割ファイルを処理するべき等べき下流を参照してください。
便宜上、MetadataStoreSelector オプションは <idempotent-receiver> コンポーネントで直接構成可能です。次のリストは、可能なすべての属性を示しています。
<idempotent-receiver
id="" (1)
endpoint="" (2)
selector="" (3)
discard-channel="" (4)
metadata-store="" (5)
key-strategy="" (6)
key-expression="" (7)
value-strategy="" (8)
value-expression="" (9)
compare-values="" (10)
throw-exception-on-rejection="" /> (11)| 1 | IdempotentReceiverInterceptor Bean の ID。オプション。 |
| 2 | このインターセプターが適用されるコンシューマーエンドポイント名またはパターン。endpoint="aaa, bbb*, ccc, *ddd, eee*fff" など、コンマ（,）で名前（パターン）を区切ります。これらのパターンに一致するエンドポイント Bean 名は、ターゲットエンドポイントの MessageHandler Bean を取得するために使用され（.handler サフィックスを使用）、IdempotentReceiverInterceptor がそれらの Bean に適用されます。必須。 |
| 3 | MessageSelector Bean リファレンス。metadata-store および key-strategy (key-expression) と相互に排他的。selector が提供されない場合、key-strategy または key-strategy-expression のいずれかが必要です。 |
| 4 | IdempotentReceiverInterceptor がメッセージを受け入れない場合にメッセージを送信するチャネルを識別します。省略すると、重複したメッセージが duplicateMessage ヘッダーとともにハンドラーに転送されます。オプション。 |
| 5 | ConcurrentMetadataStore リファレンス。基礎となる MetadataStoreSelector によって使用されます。selector と相互に排他的。オプション。デフォルトの MetadataStoreSelector は、アプリケーションの実行中に状態を維持しない内部 SimpleMetadataStore を使用します。 |
| 6 | MessageProcessor リファレンス。基礎となる MetadataStoreSelector によって使用されます。リクエストメッセージから idempotentKey を評価します。selector および key-expression と相互に排他的。selector が提供されない場合、key-strategy または key-strategy-expression のいずれかが必要です。 |
| 7 | ExpressionEvaluatingMessageProcessor に入力する SpEL 式。基礎となる MetadataStoreSelector によって使用されます。リクエストメッセージを評価コンテキストルートオブジェクトとして使用して、idempotentKey を評価します。selector および key-strategy と相互に排他的。selector が提供されない場合、key-strategy または key-strategy-expression のいずれかが必要です。 |
| 8 | MessageProcessor リファレンス。基礎となる MetadataStoreSelector によって使用されます。リクエストメッセージから idempotentKey の value を評価します。selector および value-expression と相互に排他的。デフォルトでは、"MetadataStoreSelector" は "timestamp" メッセージヘッダーをメタデータの「値」として使用します。 |
| 9 | ExpressionEvaluatingMessageProcessor に入力する SpEL 式。基礎となる MetadataStoreSelector によって使用されます。リクエストメッセージを評価コンテキストルートオブジェクトとして使用して、idempotentKey の value を評価します。selector および value-strategy と相互に排他的。デフォルトでは、"MetadataStoreSelector" は "timestamp" メッセージヘッダーをメタデータ "value" として使用します。 |
| 10 | キーの古い値と新しい値を比較することにより、オプションでメッセージを選択できる BiPredicate<String, String> Bean への参照。デフォルトでは null。 |
| 11 | IdempotentReceiverInterceptor がメッセージを拒否した場合に例外をスローするかどうか。デフォルトは false です。discard-channel が提供されているかどうかに関係なく適用されます。 |
Java 構成の場合、Spring Integration はメソッドレベルの @IdempotentReceiver アノテーションを提供します。メッセージングアノテーションを持つ method をマークするために使用されます（@ServiceActivator、@Router, and others) to specify which `IdempotentReceiverInterceptor オブジェクトがこのエンドポイントに適用されます。次の例は、@IdempotentReceiver アノテーションの使用方法を示しています。
@Bean
public IdempotentReceiverInterceptor idempotentReceiverInterceptor() {
return new IdempotentReceiverInterceptor(new MetadataStoreSelector(m ->
m.getHeaders().get(INVOICE_NBR_HEADER)));
}
@Bean
@ServiceActivator(inputChannel = "input", outputChannel = "output")
@IdempotentReceiver("idempotentReceiverInterceptor")
public MessageHandler myService() {
....
}Java DSL を使用する場合、次の例に示すように、インターセプターをエンドポイントのアドバイスチェーンに追加できます。
@Bean
public IntegrationFlow flow() {
...
.handle("someBean", "someMethod",
e -> e.advice(idempotentReceiverInterceptor()))
...
}IdempotentReceiverInterceptor は、MessageHandler.handleMessage(Message<?>) メソッド専用に設計されています。バージョン 4.3.1 以降では、AbstractHandleMessageAdvice を基本クラスとして HandleMessageAdvice を実装し、より良い分離を実現しています。詳細については、メッセージアドバイスの処理を参照してください。 |
ロギングチャネルアダプター
<logging-channel-adapter> は、ワイヤータップに従って、ワイヤタップと組み合わせて使用されることがよくあります。ただし、あらゆるフローの最終的なコンシューマーとしても使用できます。例: 結果を返す <service-activator> で終わるフローを考えますが、その結果を破棄したいとします。そのためには、結果を NullChannel に送信できます。または、INFO レベル <logging-channel-adapter> にルーティングできます。これにより、INFO レベルでログを記録するときに破棄されたメッセージを表示できますが、WARN レベルでログを記録するときには表示されません。NullChannel では、DEBUG レベルでログを記録するときに、破棄されたメッセージのみが表示されます。以下のリストは、logging-channel-adapter エレメントのすべての可能な属性を示しています。
<int:logging-channel-adapter
channel="" (1)
level="INFO" (2)
expression="" (3)
log-full-message="false" (4)
logger-name="" /> (5)| 1 | ロギングアダプターをアップストリームコンポーネントに接続するチャネル。 |
| 2 | このアダプターに送信されたメッセージが記録されるログレベル。デフォルト: INFO |
| 3 | メッセージのどの部分がログに記録されるかを正確に表す SpEL 式。デフォルト: payload — ペイロードのみが記録されます。log-full-message が指定されている場合、この属性は指定できません。 |
| 4 | true の場合、メッセージ全体（ヘッダーを含む）が記録されます。デフォルト: false — ペイロードのみが記録されます。expression が指定されている場合、この属性は指定できません。 |
| 5 | ロガーの name を指定します（log4j では category として知られています）。このアダプターによって作成されたログメッセージを識別するために使用されます。これにより、個々のアダプターのログ名を（ロギングサブシステムで）設定できます。デフォルトでは、すべてのアダプターは org.springframework.integration.handler.LoggingHandler という名前でログを記録します。 |
Java 構成の使用
次の Spring Boot アプリケーションは、Java 構成を使用して LoggingHandler を構成する例を示しています。
@SpringBootApplication
public class LoggingJavaApplication {
public static void main(String[] args) {
ConfigurableApplicationContext context =
new SpringApplicationBuilder(LoggingJavaApplication.class)
.web(false)
.run(args);
MyGateway gateway = context.getBean(MyGateway.class);
gateway.sendToLogger("foo");
}
@Bean
@ServiceActivator(inputChannel = "logChannel")
public LoggingHandler logging() {
LoggingHandler adapter = new LoggingHandler(LoggingHandler.Level.DEBUG);
adapter.setLoggerName("TEST_LOGGER");
adapter.setLogExpressionString("headers.id + ': ' + payload");
return adapter;
}
@MessagingGateway(defaultRequestChannel = "logChannel")
public interface MyGateway {
void sendToLogger(String data);
}
}Java DSL を使用した構成
次の Spring Boot アプリケーションは、Java DSL を使用してロギングチャネルアダプターを構成する例を示しています。
@SpringBootApplication
public class LoggingJavaApplication {
public static void main(String[] args) {
ConfigurableApplicationContext context =
new SpringApplicationBuilder(LoggingJavaApplication.class)
.web(false)
.run(args);
MyGateway gateway = context.getBean(MyGateway.class);
gateway.sendToLogger("foo");
}
@Bean
public IntegrationFlow loggingFlow() {
return IntegrationFlows.from(MyGateway.class)
.log(LoggingHandler.Level.DEBUG, "TEST_LOGGER",
m -> m.getHeaders().getId() + ": " + m.getPayload());
}
@MessagingGateway
public interface MyGateway {
void sendToLogger(String data);
}
}java.util.function インターフェースのサポート
バージョン 5.1 から、Spring Integration は java.util.function パッケージのインターフェースを直接サポートします。すべてのメッセージングエンドポイント（Service Activator、Transformer、Filter など）が Function （または Consumer）Bean を参照できるようになりました。メッセージングアノテーションは、通常の MessageHandler 定義と同様に、これらの Bean に直接適用できます。たとえば、この Function Bean 定義がある場合:
@Configuration
public class FunctionConfiguration {
@Bean
public Function<String, String> functionAsService() {
return String::toUpperCase;
}
}XML 構成ファイルの単純な参照として使用できます。
<service-activator input-channel="processorViaFunctionChannel" ref="functionAsService"/>メッセージングアノテーションを使用してフローを構成する場合、コードは簡単です。
@Bean
@Transformer(inputChannel = "functionServiceChannel")
public Function<String, String> functionAsService() {
return String::toUpperCase;
} 関数が配列を返す場合、Collection （基本的には任意の Iterable）、Stream または Reactor Flux、@Splitter をそのような Bean で使用して、結果コンテンツの反復を実行できます。
java.util.function.Consumer インターフェースは、<int:outbound-channel-adapter> に使用したり、@ServiceActivator アノテーションと一緒に使用して、フローの最終ステップを実行したりできます。
@Bean
@ServiceActivator(inputChannel = "messageConsumerServiceChannel")
public Consumer<Message<?>> messageConsumerAsService() {
// Has to be an anonymous class for proper type inference
return new Consumer<Message<?>>() {
@Override
public void accept(Message<?> e) {
collector().add(e);
}
};
} また、上記のコードスニペットのコメントにも注意してください。Function/Consumer でメッセージ全体を処理する場合は、ラムダ定義を使用できません。Java 型消去のため、apply()/accept() メソッド呼び出しのターゲット型を判別できません。
java.util.function.Supplier インターフェースは、@InboundChannelAdapter アノテーションと一緒に、または <int:inbound-channel-adapter> の ref として単純に使用できます。
@Bean
@InboundChannelAdapter(value = "inputChannel", poller = @Poller(fixedDelay = "1000"))
public Supplier<String> pojoSupplier() {
return () -> "foo";
}Java DSL では、エンドポイント定義で関数 Bean への参照を使用するだけです。一方、Supplier インターフェースの実装は、通常の MessageSource 定義として使用できます。
@Bean
public Function<String, String> toUpperCaseFunction() {
return String::toUpperCase;
}
@Bean
public Supplier<String> stringSupplier() {
return () -> "foo";
}
@Bean
public IntegrationFlow supplierFlow() {
return IntegrationFlows.from(stringSupplier())
.transform(toUpperCaseFunction())
.channel("suppliedChannel")
.get();
}この関数サポートは、Spring Cloud Function (英語) フレームワークと一緒に使用すると便利です。Spring Cloud Function (英語) フレームワークでは、関数カタログがあり、そのメンバー関数を統合フロー定義から参照できます。
Kotlin ラムダ
フレームワークは、関数の Kotlin ラムダをサポートするように改善されたため、Kotlin 言語と Spring Integration フロー定義の組み合わせを使用できるようになりました。
@Bean
@Transformer(inputChannel = "functionServiceChannel")
fun kotlinFunction(): (String) -> String {
return { it.toUpperCase() }
}
@Bean
@ServiceActivator(inputChannel = "messageConsumerServiceChannel")
fun kotlinConsumer(): (Message<Any>) -> Unit {
return { print(it) }
}
@Bean
@InboundChannelAdapter(value = "counterChannel",
poller = [Poller(fixedRate = "10", maxMessagesPerPoll = "1")])
fun kotlinSupplier(): () -> String {
return { "baz" }
}