こちらの記事は、アソビュー！ Advent Calendar 2024 の13日目シリーズ1です。

はじめに

こんにちは！アソビューでバックエンドエンジニアをしております、佐藤です。
今回はSpring WebFluxを用いたAPIのリトライ処理を実装した話をご紹介したいと思います。
過去に、下記の記事でSpring WebFluxについて紹介しているので、Spring WebFluxをよくご存知ない方は併せて見ていただけたらと思います。

tech.asoview.co.jp

背景

9月にアソビューはモバイルアプリ版に「すべて見る」機能を追加しました。
この機能により、TOP画面に表示されるランキングにおいて、以前は最大10件までしか表示されなかったものが、「すべて見る」のリンクまたはアイコンをタップすることで全件を表示するランキング画面へ遷移できるようになっています。

こちらの実現にあたり、以下の設計要件のもと実装を行いました。

設計要件

ページネーション

ランキング10位分のプランを1ページ単位とし、ページネーションで順番にプランを取得。

条件チェック

1ページ分のランキングのプランの詳細を取得し、
取得したプランが「在庫あり」「予約受付中」などの条件を満たしているかを確認。

リトライ処理

条件チェックのために様々なAPIを呼んでいる関係上、
アクセスの都度、ランキング全件分の条件チェックを行うのは負荷観点で懸念があるため、
条件チェックの際、1ページ分の全てのプランが条件を満たさない場合、次のページを取得し、条件を再確認する。

今回はリトライ処理において、Fluxを使用して実装を行いました。

対応方針案

Fluxで実装を行う他に幾つかの方針案がありましたのでご紹介します。
前提として、今回実装を行ったアプリケーションはAPI Gatewayであり、Spring WebFluxを利用しています。

方針1: for文で解決

まずは単純にfor文で実現できないかということで下記のように実装してみました。
最初に①でランキングの全順位分のプランコードを取得し、②で1ページ分のプランコードのリストのリストに変換します。
こちらをfor文で回しながら③にて詳細を取得、条件にあった場合は結果を返し、合わない場合は次ページ分のプランコードのリストをもとに再度詳細を取得する。というような実装です。

Mono<List<PlanDetail>> fetchRankingPlans(
    RankingPlanRequest request) {
    // ①ランキング対象のプランコードを取得
    return rankingServiceStub.fetchRankingPlanCodes(request) 
        .flatMap(rankingPlanCodes -> {
            // ②取得したプランコードをページ単位（例：10件ずつ）に分割
            List<List<RankingPlanCode>> paginatedPlanCodes = paginatePlanCodes(rankingPlanCodes);
            int retryCount = 0; // リトライ回数を追跡

            for (var currentPagePlanCodes : paginatedPlanCodes) {
                // ③各ページのプランコードに対してプラン情報を取得、在庫や予約受付有無を確認
                // 条件に当てはまるプランが一件もない場合空のリストを返却する
                Mono<List<PlanDetail>> details = fetchPlanDetails(currentPagePlanCodes);

                // ランキング詳細情報が存在する場合はその結果を返す
                if (!details.isEmpty()) { // ④
                    return details;
                }

                // リトライの最大回数に達した場合はループを抜ける
                if (retryCount >= MAX_RETRY_COUNT) {
                    break;
                }
                retryCount++;
            }

            // すべての試行が失敗した場合は空のリストを返す
            return Mono.just(new ArrayList<>());
        });
}

問題点

上記実装は④の箇所でコンパイルエラーとなりました。
④ではdetails.isEmpty()を直接呼び出しています。fetchPlanDetails()の返り値はMonoでラップされており、結果をすぐに取得することができないためコンパイルエラーになりました。

こちらは下記のように書き換えることでコンパイルエラーは消えました。

...
if (details.block().isEmpty()) { // ④
    return details;
}
...

しかし、block()を使うと同期的に処理が行われてしまうため、今回のようにノンブロッキングで処理を行いたい場合には、リアクティブプログラミングの良さを損なうことになります。
特に、今回実装したAPI GatewayではWebFluxをすでに使用しており、同期的な処理を含ませてしまうとパフォーマンスの低下を招くリスクがあります。そのため、block()を避け、ノンブロッキングな処理を維持する別の方法を検討する必要がありました。

方針2: webFluxのretryWhen()で実現

WebFluxでの非同期を保ったままリトライ処理をかけないか、少し調べてみたところ、retryWhen()というものがヒットしました。
flatMapでの処理内で、特定のエラーを検知した際、flatMapの処理を指定回数分、リトライすることができるようです。

こちらを使用し実装してみました。 ①でランキングの全順位分のプランコードを取得し、②で詳細の取得、条件に当てはまるプランが1件もない場合、③でRuntimeExceptionを発生させます。リトライ処理を実行する条件は④のように実装します。

Mono<List<PlanDetail>> fetchRankingPlans(
    RankingPlanRequest request) {
    // ①ランキング対象のプランコードを取得
    return rankingServiceStub.fetchRankingPlanCodes(request) 
        .flatMap(rankingPlanCodes -> {
            // ②各ページのプランコードに対してプラン情報を取得、在庫や予約受付有無を確認
            // 条件に当てはまるプランが一件もない場合空のリストを返却する
            fetchPlanDetails(rankingPlanCodes).flatMap(
                planDetails -> {
                    if (planDetail.isEmpty()) {
                        // ③条件に当てはまるプランが一件もない場合例外をスロー
                        Mono.error(new RuntimeException("error"));
                    }
                    // ランキング詳細情報が存在する場合はその結果を返す
                    return Mono.just(planDetails);
                }
            )
            // ④リトライ処理の条件を定義
            .retryWhen(
                Retry.max(MAX_RETRY_COUNT)
                .filter(throwable -> throwable instanceof RuntimeException)
            );

        })
}

問題点

要件では、取得した10位分のプランのうちすべてが条件に当てはまらなかった場合、次の10位分のプランを追加で取得する、となっています。
しかし、リトライ処理を行う際に必要となる「次の10位分のプラン」を取得するためには、対応するプランコードを動的に更新する必要があります。retryWhenを使用したリトライ処理では変数の更新が難しいため、要件に沿うような処理は実現できませんでした。

方針3: Fluxで実現

方針1、2と試して行き着いたのがFluxの使用でした。 Monoが0から1つのオブジェクトの非同期操作に対応するのに対して、Fluxは0以上のオブジェクトの非同期操作に対応しているようです。上記を踏まえ実装してみます。

Mono<List<PlanDetail>> fetchRankingPlans(
    RankingPlanRequest request) {
    // ①ランキング対象のプランコードを取得
    return rankingServiceStub.fetchRankingPlanCodes(request) 
        .flatMap(rankingPlanCodes -> {
            // ②取得したプランコードをページ単位（10件ずつ）に分割
            List<List<RankingPlanCode>> paginatedPlanCodes = paginatePlanCodes(rankingPlanCodes);
            // ③10件ずつのリストをFluxに変換
            return Flux.fromIterable(paginatedPlanCodes)
                    // リトライ処理の回数上限を指定
                    .take(MAX_RETRY_COUNT)
                    // 各ページのプランコードに対してプラン情報を取得、在庫や予約受付有無を確認
                    // 条件に当てはまるプランが一件もない場合空のリストを返却する
                    .concatMap(planCodes -> fetchPlanDetails(planCodes))
                    // 条件に当てはまる場合リトライ処理を中断する
                    .takeUntil(planDetails -> !planDetails.isEmpty())
                    // 中断した条件に当てはまるFluxを取得する
                    .filter(planDetails -> !planDetails.isEmpty())
                    .next();
        });
}

非常に簡潔に書けました・・！
方針1で実現できなかった非同期処理と、方針2で実現できなかった次ページ分のプランコードを用いたリトライ処理が実現できています。
まず①では方針1と同様、ランキングの全順位分のプランコードを取得し、②で1ページ分のプランコードのリストのリストに変換します。
そしてそれらを③でFluxに変換し、非同期で処理していきます。今回使用したFluxの機能を順番に見ていきます。

fromIterable()

ListやSet、Queueなど、コレクションをFluxに変換します。

take(n)

Fluxの要素のうち最初のn分のみ取得します。
今回は負荷の観点からリトライ回数を制限しているためMAX_RETRY_COUNTとして指定しています。

concatMap()

Fluxの要素を順番に処理するために使用します。
処理が順番通りに行われることを保証し、次の要素を処理する前に、現在の要素の処理が完了するまで待ちます。
今回はFluxの要素として、ランキングの順位をもとにしたプランコードを格納しています。これらが並列に処理されると順位がバラバラになってしまうため、こちらを使用しました。
順序の考慮が不要な場合はflatMap()を使用すれば良さそうです。

takeUntil()

takeUntilに記載した条件がtrueになるまで順次処理を続けるというものです。trueになった時点で処理を終了します。
今回は条件に当てはまる結果が取得できた段階で、それ以降の処理は不要になるため使用しました。

filter()

filterに記載した条件がtrueであるFluxを取得します。
条件に合うものを全て取得しますが、前段でtakeUntilしているため、こちらでは条件にあった1つの要素を持つFluxを取得しています。

next()

.next()はFluxの最初の要素を取り出してMonoに変換する際使用します。
今回はリトライ処理のためにFluxを使用しましたが、最終的には対象のページ分のランキングを返却します。
対象のページ分のランキングはFluxの要素の単位としているため、Monoへ変換し、対象のページ分のみ返すことを明示的にしています。