はじめに

レスポンスが遅い API をループで呼んでいるため、レスポンスが遅くなってしまっている API がありました。レスポンス改善のため API を非同期で呼ぶというタスクをやることになったのですが、API を非同期で呼んだことがなかったので、色々確認してみました。

WebClient を使えば API を非同期で呼び出せるくらいの知識はあったのですが、実装したことはなかったですし、そもそも元のコードは RestTemplate で API を呼んでいます。
そこで、以下の 3 つの選択肢が思い浮かんだので、それぞれ実装してみて方針を決めたいなと思いました。

WebClient (ノンブロッキングな HTTP クライアント) を使う
ブロッキングな HTTP クライアントをスレッドで呼ぶ
ブロッキングな HTTP クライアントをスレッドで呼ぶ (Spring の Async アノテーション使用)

また、Request スコープの Bean を API 呼び出し時に使用しており、子スレッドでもその Bean を呼べるのかも気になったので、検証しました。

サンプルの概要

webflux-sample というよりも webclient-sample だということに後から気づきました。webflux を使った API のサンプルはもう書けないのかと思いましたが、それはそれとして。

レスポンスが遅い API (request-server) と、その API を呼ぶ API (call-api-app) の 2 つの Spring Boot アプリがあります。

レスポンスが遅い API

5 秒後にレスポンスを返すことで、重い処理をする API に見せかけています。

API を呼ぶアプリ

これも API になってます。

/web-client -> WebClient で request-server の API を呼ぶ
/completable-future -> RestClient で request-server の API を呼ぶ
/async -> RestClient・Async によって request-server の API を呼ぶ

という 3 つの API を実装しています。

Request スコープの Bean が子スレッドでも使えるのかを確かめたかったので、リクエスト単位で更新される RequestId という Bean も実装しています。

API を非同期で呼ぶ実装

API を呼ぶクラスのインターフェースを定義して、3 パターンの実装をしました。

3 回 API を呼んで (インターフェースでは表現できていない)、3 回のレスポンスをリストにして返しています。

リストの順序にもこだわり、レスポンスの順番がコード上の呼び出し順になるように実装しました。ここは、要件によって変わってきますね。順序を気にしなくて良い場合もあると思います。

1. WebClient による実装

@Override
public List<String> fetch() {
    return Flux.mergeSequential( // merge では順番保証されない
            Flux.fromStream(
                    IntStream.rangeClosed(1, 3)
                            .mapToObj(i -> webClient.get()
                                    .uri("/thread-sleep")
                                    .header("X-Request-Id", "%s-%s".formatted(requestId.getRequestId(), i))
                                    .retrieve()
                                    .bodyToFlux(String.class)
                            )
            )
    )
            .collectList()
            .block();
}

WebClient で API を呼び出した返り値を Flux として受け取り、3 つの Flux を 1 つにしてリストにして呼び出す、というイメージです。

複数の Flux を 1 つに

複数の Flux を 1 つにする方法は何通りかありますが、ノンブロッキングにしたい場合は merge 系を使います。Flux.mergeSequential は、順番を保証した状態で複数の Flux を 1 つにしてくれます。

順番が気にならない場合は mergeWith ですね。パフォーマンスに影響を与えるのかはわからないです。

感想

スレッドとか気にせず書けていいですね。Request スコープの Bean も問題なく使えています。ただ、Flux をしっかり勉強しないといけないのがつらいですね。人によっては、WebClient の勉強も必要でしょう。

2. RestClient による実装

いつの間にかブロッキングなクライアント代表になった RestClient。RestTemplate を新しく書く気になれませんでした。

こんなことを書いていると、仕事で JdbcTemplate を使っている部分を JdbcClient に書き直したくなってきました。新しく Spring を勉強した人にもメンテしてもらいやすくなると思うんですよね。いや、JdbcTemplate もそんな古くないですね。

@Override
public List<String> fetch() {
    String requestId = this.requestId.getRequestId();  // CompletableFuture 内では呼び出せないので事前に呼び出しておく

    List<CompletableFuture<String>> futures = IntStream.rangeClosed(1, 3)
            .mapToObj(
                    i -> CompletableFuture.supplyAsync(
                            () -> callApi(requestId, String.valueOf(i))
                    )
            )
            .toList();  // List ではないと非同期で実行されない

    return futures.stream()
            .map(CompletableFuture::join)
            .toList();
}

private String callApi(String requestId, String count) {
    return restClient.get()
            .uri("/thread-sleep")
            .header("X-Request-Id", "%s-%s".formatted(requestId, count))
            .retrieve()
            .body(String.class);
}

CompletableFuture.supplyAsync で RestClient を呼び出します。

futures の初期化と return を分けないと、スレッドで実行されないですよ。

感想

ブロッキングなクライアントでも API を非同期に呼び出せるようにはなりますが、スレッドが実行されるタイミングを考えながらコードを書く必要がありますね。コメントにもありますが、正直 List でないと非同期で実行されない理由はわかっていないです。Stream のままだと順番に API を呼んでしまうんですよね。

Request スコープの Bean は使えなかったです。ただ、なんとかする方法はあって、次の実装では試してみました。

3. RestClient + Async による実装

先ほどは純粋な Java のスレッドで RestClient を呼び出しましたが、Spring の機能を使ってスレッドで呼び出してみます。

実装はほぼ 2 つ目と同じなのでリンクのみでお茶を濁します。
Async をつけたメソッドは他のクラスから呼び出さないと非同期にならないという点に注意です。Cline に教えてもらって知りました。

あと、EnableAsync をつけたクラスをコンポーネントにしてください。

そのままでは、やはり Request スコープの Bean は使えなかったので、Executor の設定を書くことで解決してみました。

Request スコープの Bean を子スレッドで使う

@Bean
public Executor taskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(3);
    executor.setQueueCapacity(10);
    executor.setMaxPoolSize(500);
    executor.setThreadNamePrefix("async-");

    // Request スコープの Bean を子スレッドで使用できるようになる
    executor.setTaskDecorator(task -> {
        RequestAttributes attributes = RequestContextHolder.getRequestAttributes();
        return () -> {
            try {
                if (attributes != null) {
                    RequestContextHolder.setRequestAttributes(attributes);
                }
                task.run();
            } finally {
                RequestContextHolder.resetRequestAttributes();
            }
        };
    });

    executor.initialize();
    return executor;
}

よくわかんないですが、コメントしている部分を書くことで Request スコープの Bean が子スレッドにも引き継がれます。あんまり書く気にはなれないですね......

感想

クラスが増える時点でつらいです。2 つ目と比較してもメリットがわからないし。

まとめ

WebClient を学習できるのであれば WebClient が良いと思いました。一番シンプルな気がします。RequestScope の Bean も扱えるので、コーディングも直感的に行えます。0 からコードを書くならこれでしょう。

今回の自分のように既存システムの改修の場合は、ブロッキングなクライアントをスレッドで呼び出すのもありなのかなと思います。みんながみんな WebClient・Flux を使いこなせる訳ではないですし。RequestScope の Bean が呼び出せない問題も、メソッドの設計でなんとかなります。3 の実装で試した Request スコープの Bean を呼び出す方法は、2 でも適応できると思います。が、複雑なのでオススメではないです。API を呼ぶメソッドの引数にするなどして対応可能なことがほとんどかと思います。

Spring の Async は...... 良い使い方を教えて下さい。

以上、雑な検証ではありますが GitHub でコードを共有しているので是非見ていってください。