Android開発でRxJava2を使うと、非同期と同期処理の切り替えがもっと簡単にできるため、使ってみたのですが、新たな問題に直面しました。これまで直面したエラーをまとめてみました。

1) RecyclerViewによるjava.lang.IndexOutOfBoundsException

エラー内容

Inconsistency detected. Invalid item position 6(offset:6).state:10 android.support.v7.widget.RecyclerView{bac38b8 VFED..... .F...... 0,0-1360,2030 #7f08008f ... layout:android.support.v7.widget.GridLayoutManager@34761f7,

インターネットからfetchしたデータをRecyclerViewに反映するコードです。インターネットの処理を行うfetchListObservableがSchedulers.io()がbackgroundで処理するように subscribeOnで定義し、observeOnでmainThreadに戻し、UI処理のnotifyDataSetChangedを行います。

実行は１回だけですとエラーが出ないコードですが、同時に２回実行する（ボタンの連打など）となるとIndexOutOfBoundsExceptionのエラーが起きます。理由はmainThreadとbackgroundが共通のlistを使っているため、notifyDataSetChangedがlistを参照してrecyclerViewを更新している際に、backgroundがadd, clear, removeなどでlistの構造を変えています。

// ×の例
fetchListObservable.doOnSubscribe(disposable -> list.clear())
                .doOnNext(item -> list.add(item))
                .subscribeOn(Schedulers.io()) // ↑これまではbackground処理
                .observeOn(AndroidSchedulers.mainThread()) // ↓ここからはUI処理
                .doOnComplete(() -> adapter.notifyDataSetChanged())
                .subscribe()

listの構造を変える処理とnotifyDataSetChangedが同じmainThreadで処理するように

// ◯の例
fetchListObservable.subscribeOn(Schedulers.io()) // ↑これまではbackground処理
                .observeOn(AndroidSchedulers.mainThread()) // ↓ここからはUI処理
                .doOnSubscribe(disposable -> list.clear())
                .doOnNext(item -> list.add(item))
                .doOnComplete(() -> adapter.notifyDataSetChanged())
                .subscribe()

2) java.util.ConcurrentModificationException

（１）の例と同じ原因で、backgroundでlistの構造を変えながら、mainThreadが同一のlistを参照している際に起きる問題です。

// ×の例
fetchListObservable.filter(vm -> {
        // すでに追加されたアイテムを追加しないようにfilterする処理
        Iterator<MyObject> _list = list.iterator();
        while(_list.hasNext()) {
            MyObject obj = _list.next();
            if(obj.id == vm.id) {
                return false;
            }
        }
        
        return true;
    }
    .subscribeOn(schedulers) // ↑これまではbackground処理
    .observeOn(AndroidSchedulers.mainThread()) // ↓ここからはUI処理
    .doOnNext(vm -> list.add(vm))
    .doOnComplete(this::notifyDataSetChanged)
    .subscribe();

listの構造を変える処理をmainThreadにすると問題が解消されます。

// ×の例
fetchListObservable
    .subscribeOn(schedulers) // ↑これまではbackground処理
    .observeOn(AndroidSchedulers.mainThread()) // ↓ここからはUI処理
    .filter(vm -> {
        // mainThreadで処理する
        Iterator<MyObject> _list = list.iterator();
        while(_list.hasNext()) {
            MyObject obj = _list.next();
            if(obj.id == vm.id) {
                return false;
            }
        }
        
        return true;
    }
    .doOnNext(vm -> list.add(vm))
    .doOnComplete(this::notifyDataSetChanged)
    .subscribe();

3) io.reactivex.exceptions.UndeliverableException: java.lang.InterruptedException

AsyncTaskのcancel() と isCancelled() をイメージして、disposeとisDisposedを組むとなると、InterruptedExceptionの例外処理が起きます。イメージコードになりますが、

// ×の例
Disposable disposable = Completable.create(emitter -> {
            for (int i = 0; i < 10; i ++) {
                if(emitter.isDisposed()) {　
                    emitter.onComplete();
                }
                // disposeはcancelと違い、途中の処理に対して、割り込み処理を行うため、interruptExceptionが実行されます。
                Thread.sleep(500);
            }

            emitter.onComplete();
        })
        .subscribeOn(Schedulers.io())
        .subscribe();
                
Thread.sleep(1000);
disposable.dispose();　 // disposeはcancelと違い、すぐinterruptして実行されます。

disposeをcancel処理として使うよりも、activityがonDestroyする際に、全てのメモリを強制的に削除する役割に向いています。