Javaにおけるリアクティブプログラミング：Reactive Streams・RxJava・Project Reactorの関係を整理する

はじめに

前回の記事「リアクティブプログラミングとは」では、Observable/Observerやオペレーターといった一般的な考え方を整理しました。
本記事では、Javaエコシステムに特化して、Reactive Streams仕様・RxJava・Project Reactor(Spring WebFlux) という3つの主要な概念・ライブラリの関係を整理します。
名前が似ていて混同しやすいので、まず全体像を掴むことを目標にします。

全体像：仕様とその実装

Javaのリアクティブ界隈で最初に押さえておきたいのは、「仕様(インターフェースの取り決め)」と「その実装ライブラリ」が別物として存在している、という構造です。

Reactive Streams(仕様)
    ├── java.util.concurrent.Flow(JDK標準、Java 9〜)
    ├── RxJava(ReactiveX系の実装ライブラリ)
    └── Project Reactor(Spring系の実装ライブラリ)

Reactive Streams：ライブラリ間の相互運用のための仕様

Reactive Streamsは、特定のライブラリではなく、「非同期のストリーム処理と、ノンブロッキングなバックプレッシャーをどう扱うか」を定めた 仕様(インターフェースの取り決め) です。Publisher・Subscriber・Subscription・Processorという4つのインターフェースが中核をなします。

この仕様が存在することで、RxJavaで作ったストリームをProject Reactorのコードに渡す、といった異なる実装ライブラリ間の相互運用が可能になります。各ライブラリが独自のインターフェースだけを使っていたら、こうした連携は困難になります。

Java 9からは、この仕様に準拠したインターフェース群がjava.util.concurrent.Flowとして標準ライブラリに取り込まれました。Flow.Publisher・Flow.Subscriber等は、Reactive Streams仕様のインターフェースと意味的に1対1で対応しています。

バックプレッシャーとは

Reactive Streams仕様が特に重視しているのが バックプレッシャー(Backpressure) という考え方です。

データの発生源(Publisher)が高速に大量のデータを送り出す一方で、受け取る側(Subscriber)の処理が追いつかない場合、素朴に実装すると受信側にデータが溜まり続け、メモリを圧迫してしまいます。バックプレッシャーは、受信側が「今はこれだけの件数なら処理できる」と発生源に伝え、発生源側もそれに応じて送出量を調整する、双方向の流量制御の仕組みです。受信側が要求した分だけ送出されるため、無制限にデータが溜まり続ける事態を防げます。

RxJava：ReactiveXのJava実装

RxJavaは、リアクティブプログラミングの代表的な実装群である「ReactiveX」のJava(JVM)版です。前回の記事で触れたObservable・オペレーター(map・filter等)の概念をJavaで使えるようにしたライブラリです。

RxJavaには、バックプレッシャーの有無で使い分ける2種類のストリーム型があります。

型	バックプレッシャー	主な用途
Observable	非対応	GUIのイベント処理など、短時間・少量のデータストリーム
Flowable	対応	大量データ・ファイルI/O等、送出量の制御が必要な場面

ObservableからFlowableへ変換する際は、あふれた分をどう扱うか(バッファリングする・捨てる・最新のものだけ残す等)をBackpressureStrategyとして明示的に指定する必要があります。

Project Reactor：Spring系のリアクティブライブラリ

Project Reactorは、Reactive Streams仕様を実装した、Spring系フレームワーク(特にSpring WebFlux)で標準的に使われるリアクティブライブラリです。RxJavaと役割は似ていますが、開発元・エコシステムが異なる別のライブラリです。

Project Reactorでは、扱うデータの件数に応じて2種類の型を使い分けます。

型	送出する要素数
Mono<T>	0件または1件
Flux<T>	0件からN件(複数件の連続したストリーム)

例えば「IDを指定して1件のユーザー情報を取得するAPI」の戻り値はMono<User>、「条件に合う複数件のユーザー一覧を取得するAPI」の戻り値はFlux<User>、というように、返ってくるデータの件数に応じて型を選びます。

Flux<String> names = Flux.just("Alice", "Bob", "Carol")
    .filter(name -> name.startsWith("A") || name.startsWith("B"))
    .map(String::toUpperCase);

names.subscribe(System.out::println);
// ALICE
// BOB

Spring BootでWebFluxを使う場合、spring-boot-starter-webfluxを依存関係に追加すると、内部的にProject Reactor(reactor-core)が組み込まれ、コントローラーのメソッドの戻り値としてMono/Fluxをそのまま返せるようになります。

RxJavaとProject Reactor、どちらを使うべきか

どちらもReactive Streams仕様に準拠しており、考え方(Observable的なストリーム・オペレーターによる宣言的な変換)は共通しています。実務上の選び方の目安は以下の通りです。

• Spring Boot(特にWebFlux)を使うプロジェクト: Project Reactorが標準で組み込まれているため、素直にReactorを使うのが自然
• Spring以外のプロジェクト、あるいはAndroid開発: RxJavaの方が情報・実績が豊富で、Android開発では特に広く使われてきた歴史がある
• 両者ともReactive Streams準拠のため、必要であれば相互に変換して連携させることも可能

まとめ

項目	内容
Reactive Streams	非同期ストリーム処理とバックプレッシャーに関する「仕様」。特定ライブラリではない
java.util.concurrent.Flow	Java 9から標準ライブラリに取り込まれた、Reactive Streams準拠のインターフェース群
バックプレッシャー	受信側の処理能力に応じて、送出側がデータ量を調整する双方向の流量制御
RxJava	ReactiveXのJava実装。Observable(バックプレッシャー非対応)とFlowable(対応)を使い分ける
Project Reactor	Spring系で標準的に使われるリアクティブライブラリ。Mono(0〜1件)とFlux(0〜N件)を使い分ける

参考

• Reactive Streams
• GitHub - reactive-streams/reactive-streams-jvm
• GitHub - ReactiveX/RxJava
• Project Reactor - Reference Documentation