Spring WebFluxのリアクティブプログラミング実装サンプル（基本編）

はじめに

最近のWebサービスの開発では、バックエンドのデータの再利用性を高めるため、マイクロサービス化してWebAPIとして提供することで、いろんなクライアントで活用する事例が増えています。
これらバックエンドのAPIや外部サービスのAPIをフロントエンドで扱いやすいようにまとめたり、コントロールする中間層（いわゆるBFF。Backends For Frontends）をSpring Boot2のWebFluxでリアクティブに実装する方法を紹介します。

リアクティブ…の何がいいの？

「リアクティブプログラミング」について詳細は、他の記事に詳しいのでここでは詳述しませんが、それはノンブロッキングな処理を簡単に実装することができます。

これまでのSpring MVCでは１リクエストに対し１スレッドを割り当てて処理を行うため、今回説明するBFF(APIアグリゲーター層)においては、外部APIを呼び出して応答待ち中の"何もしない無駄な待機時間"もスレッドをブロックし続けてしまいます。
そのため、処理中に別のリクエストを受ける場合は、都度スレッドを生成する必要があります。

それに対しWebFluxでは、API呼び出し中の待機時間のスレッドをブロックせず（＝ノンブロッキングに）使いまわして別の処理を行えるため、少ないスレッドで効率よくリクエストを捌くことができます。

ただ、リアクティブな実装はクセがあるため、慣れるまでは難しいと感じるかもしれません。
本記事では、よく使いそうな実装パターンをサンプルとして紹介していきますので、これから始めてみようと思う方の手助けになれば幸いです。

サンプルで作るもの

今回のサンプルでは、バックエンドのWebAPIを取りまとめて返すBFFを作っていきます。
そのため、サクッとモックデータを返す簡単なバックエンドを用意したので、それを使います。

• Mock Blog's Resources (backends) : GitHub

上記からcloneしてmaven install後、アプリケーションを起動してください。
http://localhost:8081/categoriesにアクセスしてJSONが返ってくれば成功です。

サンプルで扱うモデルとAPI

先に、BFFで扱うバックエンドのモデルについて簡単に触れておきます。
バックエンドはシンプルなブログアプリのリソースをWebAPIで提供し、以下のモデル構成となっています。

ブログのモデル構成

ちょっと強引¹ですが、これらのエンティティ単位でAPIが用意されているので、組み合わせてフロントエンドに返すためのリアクティブな実装を見ていきましょう。

開発環境

• Windows 10
• IntelliJ IDEA ultimate 2019.1
• Java 8
• Spring Boot 2.1.6

プロジェクトの作成

Intellijなら Spring initializr に沿って、

• Developer Tools -> Lombok
• Web -> Spring Reactive Web

を選択して作成すればOKです。
(lombokは必須ではないですが、POJOの定義や生成が簡単になるのでオススメです。lombok pluginのインストールも忘れずに)

生成されたmavenビルドファイルはこちら。

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.6.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.example</groupId>
    <artifactId>reactor</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>reactor</name>
    <description>Demo project for WebFlux</description>

    <properties>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

今回作るこのBFFのサンプルも実行可能な状態でGitHubにあるで、併せて参考にしてみてください。

• API Aggregator Sample (reactor) : GitHub

それでは、いよいよ実装の説明に入っていきます。

リアクティブなAPI呼び出し

これまでSpring MVCでWebAPIをコールするには RestTemplate を使うのが一般でした。
WebFluxでは WebClient を使うことでノンブロッキングなAPIコールに対応します。

以下はWebClientの実装サンプルになります。
（簡略化して@Repositoryではなく@ServiceでWebClientを直接利用してます）

com.example.reactor.service.CategoryService.java

@Service
public class CategoryService {

    private static final String URL_BACKENDS_ROOT = "http://localhost:8081";

    private final WebClient webClient = WebClient.create();

    public Flux<Category> findAll() {
        return webClient.get()
                .uri(URL_BACKENDS_ROOT + "/categories")
                .retrieve()
                .bodyToFlux(Category.class);
    }

    public Mono<Category> read(String categoryId) {
        return webClient.get()
                .uri(URL_BACKENDS_ROOT + "/categories/{categoryId}", categoryId)
                .retrieve()
                .bodyToMono(Category.class);
    }
}

注目すべきは、findAllやreadのWebClientの処理結果がFlux<~>やMono<~> でラッピングされている点です。

• Flux<~>　…　List<~>のような複数の戻り値を表します。²
• Mono<~>　…　１つの戻り値を表します。

コントローラでこれらFlux<~>とMono<~>をどのように扱うかを、パターン別に見ていきましょう。

単発のAPIコール

まず最初に、１つのAPIをコールする実装を見てみます。

処理イメージ

実装サンプル

com.example.reactor.controller.BlogController.java

@RestController
@RequestMapping("blog")
public class BlogController {

    @Autowired
    private CategoryService categoryService;

    @GetMapping("/categories")
    public Flux<Category> findCategories() {
        return categoryService.findAll();
    }
}

一見すると、何の変哲もないRestControllerですが、コントローラの戻り値がFlux<~>のため、ノンブロッキングでカテゴリ一覧を検索しています。

複数の順次APIコール

次に、２つのAPIを順番にコールする実装を見てみます。

処理イメージ

ユーザーIDからユーザーを取得し、そのユーザーの記事ヘッダー一覧を取得する流れです。

実装サンプル

com.example.reactor.controller.BlogController.java

@RestController
@RequestMapping("blog")
public class BlogController {

    @Autowired
    private UserService userService;

    @Autowired
    private HeaderService headerService;

    @GetMapping("/headers/find-by-user/{userId}")
    public Mono<HeadersByUser> findHeadersByUserId(@NotNull @PathVariable final String userId) {
        return userService.read(userId)
                .flatMap(user -> headerService.findByUserId(user.getUserId())
                        .collectList()
                        .map(headers -> HeadersByUser.builder()
                                .user(user)
                                .headers(headers)
                                .build()));
    }
}

UserService#read()およびHeaderService#findByUserId()は先のCategoryServiceと同様にWebClientでAPIを叩く実装になっています。

ここでのポイントは、ユーザー取得後に記事ヘッダー一覧を検索する際にflatMap()で順次処理を繋げている点です。

return userService.read(userId)
    .flatMap(user -> /* 次のリアクティブな処理(Flux/Mono) */ )

このように、次の処理がまたFlux/Mono を返すリアクティブな処理の場合 flatMap() で繋げます。

逆に、Flux/Monoではなく、通常の事後処理（例ではレスポンス用のモデルに詰め替え）を行う場合は map() で繋げます。

return headerService.findByUserId(user.getUserId())
    .collectList()
    .map(headers -> /* 次の通常処理(非Flux/Mono) */ )

サンプルでは、取得したユーザーと記事ヘッダーをひとつのオブジェクトにまとめて返すため、Flux<Header>をcollectList()にてMono<List<Header>>に変換して、以下のHeadersByUserモデルにセットしています。

com.example.reactor.resource.HeadersByUser.java

@Data
@Builder
public class HeadersByUser implements Serializable {
    private User user;
    private List<Header> headers;
}

こうしてMono<HeadersByUser>をコントローラの戻り値まで繋げることで、初めて順次処理が実行されます。

余談ですが、
lombokの@Dataアノテーションでgetter/setterが自動生成されるため、上記の通りPOJOの実装はとてもシンプルになります。
@Builderは、インスタンス生成時のプロパティの初期設定をメソッドチェーンで書けるため、ラムダ式で扱いやすくなります。

lombokのBuilderを使い、１行で値を初期化しつつインスタンス生成

HeadersByUser model =
    HeadersByUser.builder().user(user).headers(headers).build();

この他にも便利な機能が揃っています。
詳しくはこの記事などが参考になります。

複数の並列APIコール

今度は、２つのAPIを並列にコールして待ち合わせする実装を見てみます。

処理イメージ

カテゴリIDからカテゴリ情報と記事ヘッダー一覧を並列に取得します。

実装サンプル

com.example.reactor.controller.BlogController.java

@RestController
@RequestMapping("blog")
public class BlogController {

    @Autowired
    private CategoryService categoryService;

    @Autowired
    private HeaderService headerService;

    @GetMapping("/headers/find-by-category/{categoryId}")
    public Mono<HeadersWithCategory> findHeadersByCategoryId(@NotNull @PathVariable final String categoryId) {
        return Mono.zip(
                    categoryService.read(categoryId), // T1
                    headerService.findByCategoryId(categoryId).collectList() // T2
                )
                .map(tuple2 -> {
                    final Category category = tuple2.getT1();
                    final List<Header> headers = tuple2.getT2();
                    return HeadersWithCategory.builder()
                            .category(category)
                            .headers(headers)
                            .build();
                });
    }
}

CategoryService#read()とHeaderService#findByCategoryId()は例によってWebClientを使いMono<Category>とFlux<Header>を返す実装です。

並列に処理する場合には、

Mono.zip( Mono処理1, Mono処理2,...)

を使います。
そして、それぞれのリアクティブな処理結果はTuple2オブジェクトから取得できます。

    .map(tuple2 -> {
        final Category category = tuple2.getT1();
        final List<Header> headers = tuple2.getT2();
        ...

並列処理の数が３つ４つと増えると、対応するTupleもTuple3、Tuple4…となりますが、使い方は同じです。
Mono.zip()に指定した順番に対応したgetT1()...T5()を使って処理結果を取得できます。

サンプルでは、取得したカテゴリと記事ヘッダー一覧をまとめるモデルHeadersWithCategoryを定義して、並列処理の最後にまとめて返却しています。

com.example.reactor.resource.HeadersWithCategory.java

@Data
@Builder
public class HeadersWithCategory implements Serializable {
    private Category category;
    private List<Header> headers;
}

順次と並列の組み合わせAPIコール

最後に、順次と並列の組み合わせたAPIコールの実装も見てみましょう。

記事IDから記事ヘッダーを取得し、続けて記事内容とコメント一覧を取得する流れです。

com.example.reactor.controller.BlogController.java

@RestController
@RequestMapping("blog")
public class BlogController {

    @Autowired
    private HeaderService headerService;

    @Autowired
    private BodyService bodyService;

    @Autowired
    private CommentService commentService;

    @GetMapping("/entries/{entryId}")
    public Mono<Entry> getEntry(@NotNull @PathVariable Long entryId) {
        return headerService.read(entryId)
                .flatMap(header -> Mono.zip(
                            bodyService.read(header.getEntryId()), // T1
                            commentService.findByEntryId(header.getEntryId()).collectList() // T2
                        )
                        .map(tuple2 -> {
                            final Body body = tuple2.getT1();
                            final List<Comment> comments = tuple2.getT2();
                            return Entry.builder()
                                    .header(header)
                                    .body(body)
                                    .comments(comments)
                                    .build();
                        })
                );
    }
}

特に補足はありません。
これまでの順次と並列の処理を素直に組み合わせた形になります。

レスポンスのEntryモデルの定義は以下になります。

com.example.reactor.resource.Entry.java

@Data
@Builder
public class Entry implements Serializable {
    public Header header;
    public Body body;
    public List<Comment> comments;
}

作ったサンプルの動作確認

今回のBFFサンプルは API Aggregator Sample (reactor) : GitHub に置いてあります。
cloneしてmaven install後、アプリケーションを起動すれば以下のエンドポイントで動作を確認できます。
（もちろん backends も起動しておいてください）

サンプル	エンドポイント
単発のAPIコール	http://localhost:8080/blog/categories
複数の順次APIコール	http://localhost:8080/blog/headers/find-by-user/qiitaro
複数の並列APIコール	http://localhost:8080/blog/headers/find-by-category/java
順次と並列の組み合わせAPIコール	http://localhost:8080/blog/entries/1

注意すべきポイント

最後に実装時に気を付けるポイントをまとめます。

block()は使わない

block()を使うと、以下のような見慣れた同期的な実装ができてしまいます。
ですがこの実装は非同期処理を、ブロックして処理が返ってくるまで待つことになります。

    @GetMapping("/headers/find-by-user/{userId}")
    public Mono<HeadersByUser> findHeadersByUserId(@NotNull @PathVariable final String userId) {

        User user = userService.read(userId).block();

        List<Header> headers =
                headerService.findByUserId(user.getUserId())
                        .collectList()
                        .block();

        HeadersByUser response =
                HeadersByUser.builder()
                        .user(user)
                        .headers(headers)
                        .build();

        return Mono.just(response);
    }

WebFlux最大の利点を捨てることになるので使わないようにしましょう。

ちなみに、サーバがNettyの場合はblock()を使った処理は未サポートのためそもそもエラーになりますが、
Tomcatの場合は利用可能なので注意してください。
（本記事の手順で構成するとNettyになります）

リアクティブな処理は最後(Controllerの戻り値)まで繋げる

順次や並列で処理を繋げて書いていたつもりが、いざ動かしてみるとまったく動かないことがあります。
そんな時はFlux/Monoを返すリアクティブな処理が、適切に後続処理へ繋がっているか確認してください。

戻り値のない処理であっても、Mono<Void>という型をControllerの戻り値に返して繋げる必要があります。

---

以上で、今回の基本編は終わりになります。
ここまでの方法でAPI AggregatorとしてのBFFはある程度作れますが、エラーハンドリングやリトライ、セッションの扱いなど、実運用に向けてはもう少し考慮が必要でしょう。

次回以降のエントリではその辺りを説明していきます。
ありがとうございました。

---

1. 本来であればドメインとして関連性の高いこれらのエンティティは、バックエンドからも一つの記事APIとして提供されるべきです。 ↩

2. Flux<~>は単に複数の戻り値を表すだけでなく、event-streamにも対応しています。 ↩