【実践】RxJavaで書くリアクティブプログラミング：具体的なコード例で学ぶ基本操作

はじめに

これまでの記事では、リアクティブプログラミングの考え方や、Reactive Streams・RxJava・Project Reactorの関係を解説してきました。

本記事では、実際にRxJavaを使って動かせる具体的なコード例を通じて、基本操作を1つずつ確認していきます。

「読めば分かるけど、実際どう書くの」というギャップを埋めることが目的です。

環境準備

Mavenを使う場合、pom.xmlに以下の依存関係を追加します。

<dependency>
    <groupId>io.reactivex.rxjava3</groupId>
    <artifactId>rxjava</artifactId>
    <version>3.1.11</version>
</dependency>

Gradleの場合は以下です。

implementation 'io.reactivex.rxjava3:rxjava:3.1.11'

以降のコード例はすべて、import io.reactivex.rxjava3.core.Observable;が前提です(必要に応じて他のimportは適宜補ってください)。

---

1. Observableを作る・購読する

リアクティブプログラミングの第一歩は、データの発生源であるObservableを作り、それを 購読(subscribe)することです。

1.1 固定の値を流す：Observable.just

import io.reactivex.rxjava3.core.Observable;

public class Example1 {
    public static void main(String[] args) {
        Observable<String> observable = Observable.just("Java", "Kotlin", "Scala");

        observable.subscribe(
            value -> System.out.println("受信: " + value),   // onNext
            error -> System.out.println("エラー: " + error),  // onError
            () -> System.out.println("完了")                  // onComplete
        );
    }
}

実行結果:

受信: Java
受信: Kotlin
受信: Scala
完了

subscribeに渡した3つのラムダ式が、それぞれonNext(値を受け取るたび)・onError(エラー時)・onComplete(完了時)に対応します。

justで渡した値が順番に流れ、最後に自動的に完了通知が送られていることが分かります。

1.2 コレクションから作る：Observable.fromIterable

import java.util.Arrays;
import java.util.List;

List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);

Observable.fromIterable(numbers)
    .subscribe(n -> System.out.println("数値: " + n));

実行結果:

数値: 1
数値: 2
数値: 3
数値: 4
数値: 5

既存のListや配列を、そのままObservableのストリームとして扱いたい場合に使います。

1.3 自分でイベントを発生させる：Observable.create

より細かく制御したい場合は、createを使って自分でいつ値を送出するかを書けます。

Observable<Integer> observable = Observable.create(emitter -> {
    System.out.println("処理を開始します");
    emitter.onNext(10);
    emitter.onNext(20);
    if (true) {
        emitter.onComplete();
    } else {
        emitter.onError(new RuntimeException("何らかのエラー"));
    }
});

observable.subscribe(
    value -> System.out.println("受信: " + value),
    error -> System.out.println("エラー: " + error.getMessage()),
    () -> System.out.println("完了しました")
);

実行結果:

処理を開始します
受信: 10
受信: 20
完了しました

注目してほしいポイント: 「処理を開始します」が表示されるのはsubscribeを呼んだ後です。

Observableはsubscribeされるまで何も実行しません(これを「遅延実行」と呼びます)。

作った時点では何も起きず、購読されて初めて動き出す、という点が命令的なコードとの大きな違いです。

---

2. 基本オペレーターでストリームを加工する

2.1 map：値を変換する

Observable.just(1, 2, 3, 4, 5)
    .map(n -> n * n)  // 各値を2乗する
    .subscribe(n -> System.out.println("2乗: " + n));

実行結果:

2乗: 1
2乗: 4
2乗: 9
2乗: 16
2乗: 25

2.2 filter：条件に合う値だけ通す

Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    .filter(n -> n % 2 == 0)  // 偶数だけを通す
    .subscribe(n -> System.out.println("偶数: " + n));

実行結果:

偶数: 2
偶数: 4
偶数: 6
偶数: 8
偶数: 10

2.3 オペレーターを連結する

mapとfilterのような複数のオペレーターは、メソッドチェーンで連結できます。

上から下に読むだけで、データがどう加工されていくかが分かるのが、宣言的な書き方の利点です。

Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    .filter(n -> n % 2 == 0)     // 偶数だけに絞り込む
    .map(n -> n * 10)            // 10倍する
    .take(3)                     // 最初の3件だけ取る
    .subscribe(n -> System.out.println("結果: " + n));

実行結果:

結果: 20
結果: 40
結果: 60

---

3. 複数の非同期処理を合成する：flatMap

実務でよくあるのが、「あるデータを取得したら、そのデータを使って別の非同期処理を呼び出す」というパターンです。

ここでは、ユーザーIDから注文一覧を非同期に取得する処理をシミュレートしてみます。

import java.util.concurrent.TimeUnit;

// ユーザーIDから注文一覧を返す「非同期処理」を模したメソッド
Observable<String> fetchOrders(int userId) {
    return Observable.just("注文A", "注文B")
        .delay(500, TimeUnit.MILLISECONDS)  // 500ms待ってから値が流れる想定
        .map(order -> "ユーザー" + userId + "の" + order);
}

Observable.just(1, 2, 3)               // 3人分のユーザーID
    .flatMap(userId -> fetchOrders(userId))  // 各ユーザーIDに対して非同期処理を呼び出す
    .subscribe(order -> System.out.println(order));

実行結果(実行タイミングにより順序は変わる場合があります):

ユーザー1の注文A
ユーザー2の注文A
ユーザー3の注文A
ユーザー1の注文B
ユーザー2の注文B
ユーザー3の注文B

flatMapは、1つの値を受け取って新しいObservableに変換し、それを平坦化して(ネストさせずに)結果のストリームへ合流させるオペレーターです。

「Observableの中でさらにObservableを呼ぶ」という、非同期処理の連鎖にありがちな入れ子構造を、フラットなメソッドチェーンとして書けます。

---

4. 複数のストリームを組み合わせる

4.1 merge：複数のストリームを1つに合流させる

Observable<String> stream1 = Observable.just("A1", "A2", "A3");
Observable<String> stream2 = Observable.just("B1", "B2", "B3");

Observable.merge(stream1, stream2)
    .subscribe(value -> System.out.println(value));

実行結果(発生順に混ざって流れる):

A1
A2
A3
B1
B2
B3

4.2 zip：複数のストリームの値を1件ずつペアにする

Observable<String> names = Observable.just("Alice", "Bob", "Carol");
Observable<Integer> ages = Observable.just(25, 30, 35);

Observable.zip(names, ages, (name, age) -> name + "(" + age + "歳)")
    .subscribe(result -> System.out.println(result));

実行結果:

Alice(25歳)
Bob(30歳)
Carol(35歳)

zipは、両方のストリームから同じ順番目の値同士を組み合わせます。

値の数が異なる場合は、少ない方に合わせて打ち切られます。

---

5. エラーハンドリング

5.1 onErrorReturn：エラー時に代わりの値を返す

Observable.just(4, 2, 0, 8)
    .map(n -> 100 / n)  // 0で割るとArithmeticExceptionが発生する
    .onErrorReturn(error -> -1)  // エラー時は-1を代わりに流す
    .subscribe(
        n -> System.out.println("結果: " + n),
        error -> System.out.println("エラー: " + error)
    );

実行結果:

結果: 25
結果: 50
結果: -1
完了(以降の8は処理されない)

onErrorReturnが呼ばれた時点でストリームは終了扱いになるため、その後に控えていた8は処理されません。

エラー発生後もストリームを継続したい場合は、要素単位でのエラー処理を検討する必要があります。

5.2 retry：エラー時に再試行する

import java.util.concurrent.atomic.AtomicInteger;

AtomicInteger attempt = new AtomicInteger(0);

Observable.<String>create(emitter -> {
        int current = attempt.incrementAndGet();
        System.out.println(current + "回目の試行");
        if (current < 3) {
            emitter.onError(new RuntimeException("失敗"));
        } else {
            emitter.onNext("成功");
            emitter.onComplete();
        }
    })
    .retry(2)  // 最大2回まで再試行する
    .subscribe(
        value -> System.out.println("結果: " + value),
        error -> System.out.println("最終的に失敗: " + error)
    );

実行結果:

1回目の試行
2回目の試行
3回目の試行
結果: 成功

retry(2)は、エラー発生時に最大2回まで購読処理そのものをやり直します。
ネットワーク越しの通信など、一時的な失敗が起こりうる処理に有効です。

---

6. スレッドを制御する：subscribeOnとobserveOn

RxJavaでは、「どのスレッドで処理を実行するか」を明示的に指定できます。

GUIアプリケーションでは特に、「重い処理はバックグラウンドスレッドで行い、結果の表示だけメインスレッドに戻す」という使い方が重要になります。

import io.reactivex.rxjava3.schedulers.Schedulers;
import io.reactivex.rxjava3.core.Observable;

Observable.just("データ取得開始")
    .map(msg -> {
        System.out.println(Thread.currentThread().getName() + "で実行: " + msg);
        // 重い処理(DB問い合わせやファイル読み込みなどを想定)
        return "処理結果";
    })
    .subscribeOn(Schedulers.io())          // ここから上の処理をIO用スレッドで実行
    .observeOn(Schedulers.single())        // ここから下の処理を別スレッドで受け取る
    .subscribe(result -> {
        System.out.println(Thread.currentThread().getName() + "で受信: " + result);
    });

Thread.sleep(500); // mainスレッドがすぐ終了してしまうのを防ぐための待機(サンプル用)

実行結果の例(スレッド名は環境により変わります):

RxCachedThreadScheduler-1で実行: データ取得開始
RxSingleScheduler-1で受信: 処理結果

• subscribeOn: Observableが値を生成する側の処理をどのスレッドで行うかを指定する(通常はストリームのどこに書いても1箇所だけ有効)
• observeOn: それ以降の(このオペレーターより下流の)処理を、指定したスレッドに切り替えて実行する

---

7. 実践例：検索ボックスのオートコンプリート

最後に、これまでのオペレーターを組み合わせた、実務でよくある例を見てみましょう。

ユーザーが検索ボックスに文字を入力するたびに、API的な検索処理を呼び出す場面を想定します。

素朴に実装すると「1文字打つたびに毎回リクエストが飛ぶ」「前のリクエストの結果が後から返ってきて表示が乱れる」といった問題が起きがちです。

import io.reactivex.rxjava3.subjects.PublishSubject;
import java.util.concurrent.TimeUnit;

// 検索キーワードの入力を表すストリーム(実際はUIのテキスト変更イベント等に相当)
PublishSubject<String> searchInput = PublishSubject.create();

// 検索APIを呼び出す処理を模したメソッド
Observable<String> search(String keyword) {
    System.out.println("[API呼び出し] キーワード: " + keyword);
    return Observable.just(keyword + "の検索結果1", keyword + "の検索結果2")
        .delay(300, TimeUnit.MILLISECONDS);
}

searchInput
    .debounce(300, TimeUnit.MILLISECONDS)  // 300ms入力が止まるまで待つ(連続入力の間引き)
    .distinctUntilChanged()                // 直前と同じキーワードなら無視する
    .filter(keyword -> keyword.length() >= 2)  // 2文字未満は検索しない
    .switchMap(keyword -> search(keyword))     // 新しい入力が来たら古い検索は打ち切る
    .subscribe(result -> System.out.println("表示: " + result));

// ユーザーの入力をシミュレート
searchInput.onNext("J");
searchInput.onNext("Ja");
searchInput.onNext("Jav");
searchInput.onNext("Java");
Thread.sleep(1000); // debounce(300ms)より後に発火させる

実行結果の例:

[API呼び出し] キーワード: Java
表示: Javaの検索結果1
表示: Javaの検索結果2

高速に連続入力されたJ・Ja・Jav・Javaのうち、実際にAPIが呼ばれたのは最後のJavaだけです。

これはdebounce(入力が一定時間止まるまで待つ)とswitchMap(新しい値が来たら古い処理を打ち切って切り替える)の組み合わせによるものです。

この2つのオペレーターの組み合わせは、検索オートコンプリートの定番パターンとしてよく使われます。

PublishSubjectは、Observable(データの発生源)とObserver(受け手)の両方の性質を持つ特殊なクラスで、「外部からのイベントを、Observableのストリームとして流し込む入り口」を作りたい場合によく使われます。

まとめ

セクション	使ったオペレーター・API	できること
Observableの作成	just・fromIterable・create	固定値・コレクション・自作ロジックからストリームを作る
基本の加工	map・filter・take	値の変換・絞り込み・件数制限
非同期処理の連鎖	flatMap	1つの値から新しい非同期処理を呼び出し、平坦化して合成する
複数ストリームの合成	merge・zip	複数の発生源を1つに合流、またはペアリングする
エラー処理	onErrorReturn・retry	エラー時の代替値・再試行
スレッド制御	subscribeOn・observeOn	処理を実行するスレッドを指定する
実践例	debounce・distinctUntilChanged・switchMap・PublishSubject	検索オートコンプリートのような、入力に応じた非同期処理の典型パターン

参考

• GitHub - ReactiveX/RxJava
• RxJava Getting Started - GitHub Wiki
• ReactiveX - Operators
• RxJava Javadoc