RxJava：基礎、ReactiveX、データストリームの操作

RxJavaとReactiveXとは

RxJavaはReactiveX仕様のJava実装であり、観測可能なストリーム（Observable）を使用した非同期プログラミングのためのライブラリです。RxJava 2は2016年にReactive Streams（Flowable）のサポートと、rx.Observableおよびio.reactivex.Observableへの分割とともにリリースされました。RxJava 3（2019）は、RxJava 2との下位互換性を持つ現在のメジャーバージョンです。

RxJavaの中核となる考え方は、すべてがストリームであるということです：データストリーム、イベントストリーム、状態ストリーム。任意の非同期操作は、データ、エラー、または完了シグナルを発行するObservableとして表現できます。ObserverがObservableにサブスクライブし、リアルタイムで通知を受け取ります。

Badoo（2024）によると、コルーチンへの移行前は、Google Playトップ200のAndroidアプリの76％が非同期操作にRxJavaを使用していました。現在はコルーチンにシェアを奪われていますが、RxJavaは何千ものアプリのプロダクションコードに残り、成熟した実績のある技術と見なされています。ReactiveXはクロスプラットフォーム仕様であり、JavaScript（RxJS）、.NET（Rx.NET）、Swift（RxSwift）などにも実装されています。

RxJavaのObserverパターン

ReactiveXは、古典的なObserverパターンを2つのメカニズムで拡張します：演算子チェーンとSchedulerベースのスレッド管理です。ObservableはObserverがサブスクライブするまでデータの発行を開始しません（遅延評価）。これにより、サブスクリプションが存在する場合にのみアクティブになるデータパイプラインを構築できます。

Observableの種類：Observable、Flowable、Single、Maybe、Completable

Observable — onErrorまたはonCompleteで0..N個の要素を発行する基本型。無制限のデータストリームに適しています。例えば、クリックイベントや位置情報の更新などです。Observableはバックプレッシャーをサポートしません。

Flowable — バックプレッシャーをサポートするObservableのReactive Streams版。データソースがObserverの処理速度よりも速く要素を生成する可能性がある場合に使用します。FlowableはBACKPRESSURE_BUFFER、DROP、LATEST、ERRORの各戦略をサポートしています。

Single、Maybe、Completable

Singleは正確に1つの要素またはエラーを発行します — ネットワークリクエストに最適です。Maybeは0または1個の要素を発行し、データが存在しない可能性のあるキャッシュに適しています。CompletableはデータなしでonCompleteまたはonErrorのみを発行し、書き込みや削除操作に便利です。これらの型は、契約を特定のケースに限定することでAPIを簡素化します。Retrofit（Android向けの一般的なHTTPクライアント）は5つのRxJava型すべてを直接サポートしており、余分なボイラープレートコードなしで各エンドポイントに最適な戻り値の型を選択できます。

RxJava演算子：ストリーム変換とフィルタリング

演算子は、あるObservableを別のObservableに変換する関数です。演算子チェーンはデータパイプラインを記述します。各演算子は前の演算子からストリームを受け取り、変換して次の演算子に渡します。RxJavaにはカテゴリに分類された200以上の演算子が含まれています。

• map — 各要素を変換します（Integer → String）
• flatMap — 要素をObservableに変換し、すべてを1つのストリームにマージします
• filter — 条件を満たす要素のみを通過させます
• zip — N個のObservableからインデックスごとに要素を結合します
• merge — 複数のObservableを1つにマージし、時系列順を保持します
• debounce — 指定された時間間隔内に別の発行がない場合にのみ要素を発行します

flatMapは最も強力なRxJava演算子の1つです。各要素に対して非同期リクエストを実行し、結果を共通のストリームに収集できます。例えば、flatMapはIDのリストから詳細をロードするために使用されます：各ID → ネットワークリクエスト → 結果のマージ。mapが単に要素を変換するのとは異なり、flatMapは複数の要素を発行したり、別のObservableに切り替えたりできるため、非同期パイプライン構築の基盤となります。

演算子によるエラー処理

onErrorResumeNext — エラー時にバックアップのObservableに切り替えます。retry — エラー時にN回再サブスクライブします。onErrorReturn — エラーの代わりにデフォルト値を返します。doOnError — ストリームを変更せずにエラー時に副作用を実行します（ロギングや分析）。これらの演算子を組み合わせることで、手動のtry/catchなしで明確なエラー処理戦略を持つ堅牢なパイプラインを構築できます。

Schedulers：RxJavaのスレッド管理

Schedulersは、ObservableとObserverが実行されるスレッドを決定します。subscribeOnはソースのスレッドを設定し、observeOnはObserverと後続の演算子のスレッドを設定します。この分離はRxJavaの重要な利点です：ソースはIOスレッド、処理はcomputation、UIはメインスレッドで実行されます。

主なSchedulers：Schedulers.io() — I/O操作用（ネットワーク、ディスク）、無制限プール。Schedulers.computation() — 計算用、コア数に応じた固定プール。Schedulers.newThread() — タスクごとに新しいスレッド。AndroidSchedulers.mainThread() — Androidメインスレッド（RxAndroid）。Schedulers.trampoline()もあり、FIFOキューで現在のスレッドでタスクを実行するため、テストに便利です。

Google（2025）によると、Schedulersの適切な使用は初心者にとってRxJavaで最も難しい部分です。典型的な間違いは、observeOnの後にsubscribeOnを呼び出すことで、ソースに影響を与えません。subscribeOnはソースのチェーンで最初に、observeOnはUIサブスクリプションの前に配置する必要があります。ルール：subscribeOnは上流（ソース）のみに影響し、observeOnは下流（サブスクライバとその後のすべての演算子）を切り替えます。

AndroidでのRxJavaコード例

3つのシナリオを考えてみましょう：Singleを使用したネットワークリクエスト、zipを使用した並行リクエスト、debounceを使用した検索フィールドのデバウンスです。

Singleを使用したネットワークリクエスト

SingleはRetrofitリクエストに最適です：1リクエスト — 1応答。UI更新のためにメインスレッドでサブスクライブします。

api.getUser(id)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new SingleObserver<User>() {
        @Override
        public void onSuccess(User user) { showUser(user); }
        @Override
        public void onError(Throwable e) { showError(e); }
    })

zipを使用した並行リクエスト

zipは2つの独立したSingleの結果を1つに結合します。これらは並行して実行され、両方が完了した後に結果が生成されます。

Single.zip(
    api.getProfile(),
    api.getSettings(),
    (profile, settings) -> new Dashboard(profile, settings)
)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(dashboard -> showDashboard(dashboard), e -> logError(e))

検索フィールドのデバウンス

debounceは急速なテキスト変更を無視し、400ミリ秒の休止後にのみリクエストを送信します。distinctUntilChangedはテキストが変更されていない場合にリクエストをキャンセルします。

RxTextView.textChanges(searchView)
    .debounce(400, TimeUnit.MILLISECONDS)
    .filter(text -> text.length() >= 3)
    .distinctUntilChanged()
    .switchMap(query -> api.search(query))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(results -> showResults(results))

RxJava vs Kotlin Coroutines：アプローチの比較

RxJavaとKotlin Coroutinesは同じ問題 — 非同期プログラミング — を解決しますが、根本的に異なるアプローチを取ります。RxJavaはObserverパターンに基づいて構築されており、プッシュベースです：ソースがデータを送信し、Observerが反応します。コルーチンはプルベースです：コードがawaitを介して順次データを要求します。

• RxJava — リアクティブ、データストリーム、200以上の演算子、プッシュベース、急な学習曲線
• Coroutines — 逐次、suspend/await、約40関数、プルベース、シンプルな構文
• RxJava — 成熟（2016）、巨大なエコシステム、ただし急な学習曲線
• Coroutines — 最新（2018）、Googleが新しいコードに推奨
• RxJava — Flowableによる組み込みバックプレッシャー、十分にテストされたバッファリング戦略
• Coroutines — バックプレッシャー対応のFlowは最近だが、JetBrainsが活発に開発中

Google I/O 2024によると、Kotlin CoroutinesはAndroidで新しい非同期コードを書くための推奨アプローチです。RxJavaは既存プロジェクト向けにサポートされています。Googleは段階的な移行のためのブリッジライブラリ（kotlinx-coroutines-rx3）を提供しています。AndroidX（LiveData、Room、Paging 3）は両方のアプローチをサポートしており、依存関係の競合なしに古いモジュールでRxJava、新しいモジュールでコルーチンを使用できます。

RxJavaからコルーチンへの移行戦略

段階的な移行：新しいコンポーネントはコルーチンで書き、古いRxJavaコードはそのまま残します。RxJava → コルーチンはawaitSingle()またはawaitFirst()を介して。コルーチン → RxJavaはfuture()またはasFlowable()を介して。大規模プロジェクトの場合、完全な移行には6〜18か月かかります。