最新の安定バージョンについては、Spring Integration 7.1.0 を使用してください!

提供されたアドバイスクラス

Spring Integration は、AOP アドバイスクラスを適用する一般的なメカニズムを提供することに加えて、これらのすぐに使えるアドバイスの実装を提供します。

再試行のアドバイス

再試行アドバイス(o.s.i.handler.advice.RequestHandlerRetryAdvice)は、Spring Framework の再試行サポートが提供する豊富な再試行メカニズムを活用します。このアドバイスの中核となるコンポーネントは RetryTemplate であり、これにより、RetryPolicy や、再試行回数が上限に達した場合に実行するアクションを決定する RecoveryCallback 戦略など、高度な再試行シナリオを設定できます。

ステートレス再試行

ステートレス再試行は、再試行アクティビティがアドバイス内で完全に処理される場合です。スレッドは一時停止し(そうするように構成されている場合)、アクションを再試行します。

ステートフルリトライ

ステートフル再試行は、再試行状態がアドバイス内で管理されるが、例外がスローされ、呼び出し元がリクエストを再送信する場合です。ステートフル再試行の例としては、現在のスレッドで実行するのではなく、メッセージ発信元 (JMS など) に再送信の責任を負わせる場合が挙げられます。ステートフル再試行には、再試行された送信を検出するための何らかのメカニズムが必要です。この目的のために、RequestHandlerRetryAdvice は stateKeyFunctionnewMessagePredicatestateCacheSize プロパティを公開します。後の 2 つは、最初のものが提供されている場合にのみ意味を持ちます。基本的に、stateKeyFunction は、RequestHandlerRetryAdvice ロジックをステートレスからステートフルに切り替えるインジケーターです。newMessagePredicate の意味は、処理するメッセージに基づいて、キーの既存の再試行状態を 2 回リフレッシュすることです。stateCacheSize はデフォルトでは 100 であり、新しい再試行状態がさらに発生すると、古いエントリがキャッシュから削除されます。おそらく、それらの古いメッセージは上流フローからもう再配信されません。たとえば、メッセージブローカーが再配信ポリシーに従ってそれらのメッセージを配信不能にした可能性があります。

デフォルトのバックオフ動作は、バックオフしないことです。再試行はすぐに試行されます。試行間でスレッドを一時停止させるバックオフポリシーを使用すると、過剰なメモリ使用やスレッド不足などのパフォーマンスの問題が発生する可能性があります。大量の環境では、バックオフポリシーを注意して使用する必要があります。

再試行アドバイスの構成

このセクションの例では、常に例外をスローする次の @ServiceActivator を使用しています。

public class FailingService {

	@ServiceActivator(inputChannel = "input", adviceChain = "retryAdvice")
    public void service(String message) {
        throw new RuntimeException("error");
    }
}
単純なステートレス再試行

デフォルトの RetryPolicy は 3 回試行し、さらに対象の MessageHandler に対する最初の呼び出しを行います。デフォルトではバックオフはないため、3 回の試行は連続して行われ、試行間に遅延はありません。RecoveryCallback はないため、最後の再試行が失敗した後、呼び出し元に例外がスローされます。Spring Integration 環境では、この最後の例外は、受信エンドポイントで error-channel を使用して処理される可能性があります。次の例では、RequestHandlerRetryAdvice のデフォルト構成を使用しています。

@Bean
RequestHandlerRetryAdvice retryAdvice() {
    return new RequestHandlerRetryAdvice();
}
回復を伴う単純なステートレス再試行

次の例では、前の例に RecoveryCallback を追加し、ErrorMessageSendingRecoverer を使用して ErrorMessage をチャネルに送信します。

@Bean
RequestHandlerRetryAdvice retryAdvice(MessageChannel recoveryChannel) {
    RequestHandlerRetryAdvice requestHandlerRetryAdvice = new RequestHandlerRetryAdvice();
    requestHandlerRetryAdvice.setRecoveryCallback(new ErrorMessageSendingRecoverer(recoveryChannel));
    return requestHandlerRetryAdvice;
}
カスタマイズされたポリシーを使用したステートレス再試行とリカバリ

より高度な設定が必要な場合は、RequestHandlerRetryAdvice にカスタマイズされた RetryPolicy を使用することができます。この例では、シンプルな RetryPolicy を引き続き使用しますが、試行回数を 4 回に増やしています。また、最初の再試行は 1 秒、2 回目は 5 秒、3 回目は 25 秒(合計 4 回)待つ ExponentialBackoff も追加します。以下のリストは、このような設定の例を示しています。

@Bean
RequestHandlerRetryAdvice retryAdvice() {
    RequestHandlerRetryAdvice requestHandlerRetryAdvice = new RequestHandlerRetryAdvice();
    requestHandlerRetryAdvice.setRecoveryCallback(new ErrorMessageSendingRecoverer(recoveryChannel()));
    RetryPolicy retryPolicy = RetryPolicy.builder()
            .maxRetries(4)
            .delay(Duration.ofSeconds(1))
            .multiplier(5.0)
            .maxDelay(Duration.ofMinutes(1))
            .build();
    requestHandlerRetryAdvice.setRetryPolicy(retryPolicy);
    return requestHandlerRetryAdvice;
}
ステートレス再試行のネームスペースサポート

次の例は、カスタムタグを含む Spring Integration XML 名前空間を使用して RequestHandlerRetryAdvice を構成する方法を示しています。

<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-retries="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-retries="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> 定義を使用します。

回復を伴う単純なステートフル再試行

再試行をステートフルにするには、RequestHandlerRetryAdvice インスタンスに Function<Message<?>, Object> stateKeyFunction を指定する必要があります。この関数は、メッセージを再送信として識別し、RequestHandlerRetryAdvice がこのメッセージの再試行の現在の状態を判断できるようにするために使用されます。ステートフル再試行の背景となる考え方は、現在のスレッドをブロックするのではなく、このメッセージの再試行状態をキャッシュし、MessageHandler の失敗を呼び出し元に再度スローすることです。通常、これは、イベントを再送信 (または再配信) できるメッセージ発信元、たとえば、nack を備えた RabbitMQ や、シーク機能を備えた Apache Kafka などのメッセージブローカー、または消費時にロールバック後の JMS で適切に機能します。キャッシュされた状態がまだない場合 (または Predicate<Message<?>> newMessagePredicate がメッセージに対して true を返す場合)、MessageHandler 呼び出しは最初の呼び出しとして扱われ、失敗すると、BackOffExecution に基づく内部 RetryState が、前述のキーでキャッシュされます。次のメッセージ到着時に、キャッシュされた状態は、MessageHandler の呼び出しを試みる前に、Thread.sleep() のバックオフ間隔を提供します。このバックオフ間隔が BackOffExecution.STOP と等しい場合(たとえば、maxAttempts に到達している場合)、このメッセージに対する再試行はそれ以上行われません。再試行サイクル全体は使い果たされたとみなされ、対応する RetryException は呼び出し元に返されるか、提供されている場合は RecoveryCallback の呼び出しに使用されます。一般に、例外処理ロジックとバックオフ実行はステートレス動作と同様ですが、スレッドがすべての maxAttempts の間ブロックされないという点のみが異なります。次の再試行呼び出しのためにメッセージを再配信するのは、メッセージ呼び出し元の責任です。

サーキットブレーカーのアドバイス

サーキットブレーカーパターンの一般的な考え方は、サービスが現在利用できない場合、それを使用しようとして時間(およびリソース)を浪費しないことです。o.s.i.handler.advice.RequestHandlerCircuitBreakerAdvice はこのパターンを実装しています。サーキットブレーカーが閉じた状態の場合、エンドポイントはサービスの呼び出しを試みます。サーキットブレーカーは、一定回数の連続試行が失敗するとオープン状態になります。オープン状態の場合、新しいリクエストは「高速で失敗」し、時間が経過するまでサービスの呼び出しは試行されません。

指定された時間が経過すると、サーキットブレーカーは半開状態に設定されます。この状態で、1 回でも試行が失敗すると、ブレーカーは直ちに開状態に移行します。試行が成功すると、ブレーカーは閉状態に移行し、設定された回数の連続失敗が発生するまで再び開状態に移行しません。試行が成功すると、ブレーカーは再び開状態に移行するタイミングを判断するために、失敗回数が 0 にリセットされます。

通常、このアドバイスは、ネットワーク接続の試行のタイムアウトなど、失敗するまでに時間がかかる可能性がある外部サービスに使用される可能性があります。

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 を使用して advice を構成する方法を示しています。

@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.<String>handle((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 は個別に構成することも、CacheManagerCacheResolverCacheErrorHandler などの共有オプションを使用して 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) {
    ...
}