Spring WebFlux 入門

こんにちは。ソフトウェアエンジニアの大北です。
フライウィールでは、プロダクトの一部で、Spring WebFluxというWeb Frameworkを使っています。Spring WebFluxは、Springが提供しているWeb Frameworkで、Nettyのようなサーバー上で、ノンブロッキングな処理を実行できます。ノンブロッキングに処理を行うことで、スレッドを節約することができ、スケーラブルなプロダクトにできるため採用しました。（詳しくは、ノンブロッキングI/Oについて調べてみてください。）
しかしながら、Spring WebFluxを用いたプログラミングは、Spring MVCなどで使われるブロッキングな処理と比べるととっつきづらく、また情報も少ないように感じます。そこで、今回は、まだSpring WebFluxを触ったことのない方や、触り始めたばかりの人に向けて参考になる情報をいくつか紹介します。
まず、SpringのWebサイトに、Spring WebFluxのチュートリアルがあります。
Building a Reactive RESTful Web Service
指定したURLにリクエストを投げると”Hello, Spring!”と返ってくるだけの簡単な例ですが、どうやってWebFluxを使ってWebClientを実装するのか、そして実装したものをどうやってテストするのかといったことを学べます。
上のチュートリアルでは、”Mono”という型が登場します。Spring WebFluxでは、Reactor（JVM上で動くノンブロッキングなアプリケーションを作るためのリアクティブライブラリー）が採用されていて、値を扱う時にMonoとFluxという型を利用します。Monoでは、0もしくは1個の値を扱うことができ、Fluxでは、N個の値を扱うことができます。また、MonoやFluxでは、Publish-subscribe pattern¹ が使われていて、subscribeするまでは、処理が実行されません。

@Test
public void test_mono() {
  // 1を持ったMonoを作る
  Mono<Integer> number = Mono.just(1);
  // numberが持っている値に1を足す
  number.map(i -> i + 1)
        // 処理を実行し、結果を出力する
        .subscribe(i -> System.out.println("number: " + i));
}

number: 2

@Test
public void test_flux() {
  // 1, 2, 3, 4を持ったFluxを作る
  Flux<Integer> numbers = Flux.just(1, 2, 3, 4);
  // numbersが持っている全ての値に1を足す
  numbers.map(i -> i + 1)
         // 処理を実行し、結果を出力する
         .subscribe(i -> System.out.println("number: " + i));
}

number: 2
number: 3
number: 4
number: 5

ところで、上の具体例ではjustやmap、subscribeといったメソッドを呼び出しましたが、どうやって自分のしたい処理に合ったメソッドを探せばいいでしょうか？
MonoやFluxには、とても多くのメソッドが用意されているため、メソッド一覧を見て、欲しいメソッドを探し出すのはとても大変です。
そこで、役に立つのが、Reactorのサイトにある Which operator do I need? です。例えば、上の具体例で使ったmapというメソッドにたどりつくまでの説明には、
A.2. Transforming an Existing Sequence

• I want to transform existing data:
  • on a 1-to-1 basis (eg. strings to their length): map

と書いてあります。これは、日本語にすると
A.2. 既に存在しているシーケンスを変形したい

• 既に存在しているデータを変形したい
  • 1対1の対応で変形したい（例. 複数の文字列を各々の文字列の長さに変形させる）

といった具合になります。上の具体例では、全ての値に1を足したかったので、このmapを選びました。
また、WebFluxを使ったプログラミングのとっつきづらさの原因の1つとして、いつ何が起きているのかわかりづらく、デバッグするのが大変ということがあると思います。私は、このデバッグのしづらさを軽減するために、MonoやFluxに実装されているlogメソッドを利用しています。（Logging a sequenceで詳しく説明されています。）このメソッドを使うと、どのように処理が行われているのかがログに出力されます。先程の例に対して、logを呼んでみると、以下のようになります。
logを使った例

@Test
 public void test_log() {
  Flux<Integer> numbers = Flux.just(1, 2, 3, 4);
  numbers.log()
         .map(i -> i + 1)
         .subscribe(i -> System.out.println("number: " + i));
 }

出力結果

2018-12-06 12:13:34.733  INFO [-,,,] 23583 --- [    Test worker] reactor.Flux.LiftFuseable.1              : | onSubscribe([Fuseable] ScopePassingSpanSubscriber)
2018-12-06 12:13:34.734  INFO [-,,,] 23583 --- [    Test worker] reactor.Flux.LiftFuseable.1              : | request(unbounded)
2018-12-06 12:13:34.735  INFO [-,,,] 23583 --- [    Test worker] reactor.Flux.LiftFuseable.1              : | onNext(1)
number: 2
2018-12-06 12:13:34.736  INFO [-,,,] 23583 --- [    Test worker] reactor.Flux.LiftFuseable.1              : | onNext(2)
number: 3
2018-12-06 12:13:34.736  INFO [-,,,] 23583 --- [    Test worker] reactor.Flux.LiftFuseable.1              : | onNext(3)
number: 4
2018-12-06 12:13:34.737  INFO [-,,,] 23583 --- [    Test worker] reactor.Flux.LiftFuseable.1              : | onNext(4)
number: 5
2018-12-06 12:13:34.738  INFO [-,,,] 23583 --- [    Test worker] reactor.Flux.LiftFuseable.1              : | onComplete()

見やすく加工すると、以下のようなログが出力されているのがわかります。

INFO: | onSubscribe([Fuseable] ScopePassingSpanSubscriber)
INFO: | request(unbounded)
INFO: | onNext(1)
number: 2
INFO: | onNext(2)
number: 3
INFO: | onNext(3)
number: 4
INFO: | onNext(4)
number: 5
INFO: | onComplete()

これによって、どのように処理が進んでいったのかがわかります。
例えば、この例だと、

1. numbersがサブスクライブされる（onSubscirbe）
2. numbersがリクエストを受け取る（request）
3. リクエストを処理する
   1. 1に対してリクエストを処理する（onNext(1)）
   2. 2に対してリクエストを処理する（onNext(2)）
      …
4. 全ての処理が正常に終了した（onComplete）

という風に進んでいます。
今回の例では使用しませんでしたが、onSubscribeやonNextのタイミングで何か処理を行いたい場合は、doOnSubscribeやdoOnNextを使うことができます。
以上が、私の考えるSpring WebFluxを始める時にあると便利な情報です。次の私のターン（12月14日を予定しています）では、これらを使った応用として、リアクティブにRedisを操作する方法を紹介したいと思っています。

Notes