RxJavaのAndroidバインディングであるRxAndroidを使い始めてBackpressureではまったので整理
retrolambdaでlambda式使っていますが、使っていなければ適宜newするコードに置き換えてください

Backpressure

通信処理にRxJava使った時に、

System.err﹕ rx.exceptions.MissingBackpressureException

みたいなエラーを吐いてしまうことがあったのでその対策

問題のあるコード

通信部分は例えばこんな感じ

Client.java

public Observable<Item> get() {
    return Observable.create(subscriber -> {
        subscriber.onStart();
        List<Object> items = // 通信
        for (int i = 0; i < items.size(); i++) {
            Item item = // parseする
            subscriber.onNext(Item);
        }
        subscriber.onCompleted();
    });
}

何か通信して取ってきてパースしてItemオブジェクトにしてからsubscriber#onNextでsubscriberに返す、という流れ
これをsubscribeする側はこんな感じ

MainActivity.java

private void fetchItems(String url) {
    new Client(url).get()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(
                    this::setResult,
                    Throwable::printStackTrace,
                    () -> Log.d(TAG, "complete"));
}

private void setResult(Item item) {
    // 通信結果を何かする
}

ioスレッドで通信してメインスレッドでsetResultメソッドを呼ぶ、という流れ
これだと上にも上げたようなエラーのstacktraceが表示された

W/System.err﹕ rx.exceptions.MissingBackpressureException
W/System.err﹕ at rx.internal.util.RxRingBuffer.onNext(RxRingBuffer.java:338)
W/System.err﹕ at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.onNext(OperatorObserveOn.java:115)
W/System.err﹕ at rx.internal.operators.OperatorSubscribeOn$1$1$1.onNext(OperatorSubscribeOn.java:76)

ワーカースレッドから送られてくるデータにメインスレッドでの処理が間に合ってないようなイメージだと思う

解決策

RxJava Wikiにもあるようにいくつか解決策があった

new Client(url).get()
        .onBackpressureBuffer() // ここ
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        // .onBackpressureBuffer() ここだとダメ
        .subscribe(/** なにか処理 **/);

この例のようにonBackpressureBufferやその他onBackpressureDrop、onBackpressureBlockなどをメソッドチェーンでcallするといい感じに完了できる

onBackpressureBuffer
- ワーカースレッドからの通知をバッファリングする
onBackpressureDrop
- メインスレッドでの処理中にはワーカースレッドから通知せずに捨てる
onBackpressureBlock
- メインスレッドの処理が終わるまでワーカースレッド処理を一時停止させる

理解がやや曖昧ですが、こんな感じのはず
observeOnより前にcallしなければいけない点に注意しなければいけない

RxJava(RxAndroid)でのBackpressureについて

Backpressure

問題のあるコード

解決策