RxJava で Observable の並列処理を直列化する

#RxJava#ReactiveX#Java
Posted on 2016/02/08 00:00:00(+9:00)

rx.Observable<T> のオペレータは、通常は非同期で、並列に処理されます。

例えば以下のような場合:

public void start() {
    Observable.range(1, 5)
        .flatMap(x -> fatTask(x))
        .subscribe(x -> Log.d(TAG, "onNext - " + x));
}
 
private final Random rand = new Random();
private final ScheduledExecutorService executor = Executors.newScheduledThreadPool(5);

// ランダムにスリープした後 x を onNext する
private Observable<Integer> fatTask(final int x) {
    return Observable.create(subscriber -> {
        long sleep = (long) (rand.nextDouble() * 10000L);
        Log.d(TAG, "fatTask(" + x + ") - start.");

        executor.schedule(() -> {
            subscriber.onNext(x);
            subscriber.onCompleted();
        }, sleep, TimeUnit.MILLISECONDS);
    });
}

このプログラムの出力はこうなります。

出力: fatTask(1) - start.

fatTask(2) - start. fatTask(3) - start. fatTask(4) - start. fatTask(5) - start. onNext - 3 onNext - 5 onNext - 4 onNext - 2 onNext - 1

fatTask は 1,2,3,4,5 の順で 完了を待たずに 呼びだされます。 が、それぞれ処理にかかる時間が異なるので、 onNext が呼ばれる順は 1〜 とは限りません。

ソースとなる Stream の順番を崩したくない場合は、 fatTask(1) が完了してから fatTask(2) を開始する、というように直列化しなければなりません。

Observable.Concat(concatWith)

これを行うのが Observable.Concat です(RxJava では Observable.concatWith のようですね)。 複数の Observable を順に（完了してから次へ）処理していきます。

使い方

toList で一旦ただの List にしてから、concatWith で数珠つなぎにします。

public void start() {
    Observable.range(1, 5)
        .toList()
        .flatMap(list -> {
            // fatTask(1).contat(fatTask(2)).contat(fatTask(3))... 
            // にする（fold 使えれば…)
            Observable<Integer> task = null;
            for (int x : list) {
                if (task == null) {
                    task = fatTask(x);
                } else {
                    task = task.concatWith(fatTask(x));
                }
            }
            return task;
        })
        .subscribe(x -> Log.d(TAG, "onNext - " + x));
}

このプログラムの出力はこうなります。

出力

fatTask(1) - start. onNext - 1 fatTask(2) - start. onNext - 2 fatTask(3) - start. onNext - 3 fatTask(4) - start. onNext - 4 fatTask(5) - start. onNext - 5

fatTask(1) の完了を待ってから、次の fatTask(2) が実行されています。

※ Rx.NET では、

static IObservable<T> Concat<T>(IEnumerable<IObservable<T>> sources)

で、複数の IObservable を一括で渡せるのですが、 RxJava にはないようで、、、。

static <T> Observable<T> concatEager(Iterable<? extends Observable<? extends T>> sources)

というのがあったんですが、期待通りうごいてくれず、 Eager? なんでしょう？

ソースが無限リストだったら？

toList で一旦ただの List にしているのが非常に気に入らないですね。 range(1, 5) が interval(1, TimeUnit.SECONDS) のように無限の Stream だったら使えません。

そこで、 concat には、こんな overload もあります。

static <T> Observable<T> concat(Observable<? extends Observable<? extends T>> observables)

Observable を通知する Observable？ ややこしいですがこう使います。

public void start() {
    // 2. を concat する
    Observable.concat( 
        // 1. Observable<Long>
        Observable.interval(1, TimeUnit.SECONDS) 
            // 2. Long を Observable<Integer> に変換 
            //    → Observable<Observable<Integer>> になる
            .map(x -> fatTask(x.intValue()))) 
        .subscribe(x -> Log.d(TAG, "onNext - " + x));
}

このプログラムの出力はこうなります。

出力

fatTask(0) - start. onNext - 0 fatTask(1) - start. onNext - 1 fatTask(2) - start. onNext - 2 …つづく

無限リストながら、並列処理せずに順序通り動いてくれます。

interval の値を単純に map で Observable<Integer> に変換してやります。するとこれは Observable<Observable<Integer>> になり、concat 可能になります。 flatMap だと平坦化されちゃうのでただの map です。

まとめ

Observable は普通は非同期で並列処理。 非同期ながら直列化したい場合は Observable.concat でできます。

  1. GPS から緯度経度を取得
  2. なんか重い計算を行う
  3. 結果をテキストファイルに書き出す

みたいな処理をするとき 3. を 1. の順序と同じにしたいのでこれを使います。

はじめ自分は flatMap で繋いでいくだけですべて直列化されているのかなーと勘違いしていたので、これを知った時は目からウロコでした。

参考