1. RxJava2 : 什么是观察者模式
2. RxJava2 : 创建操作符(无关时间)
3. Rxjava2 : 创建操作符(有关时间)
4. Rxjava2 : 变换操作符
5. Rxjava2 : 判断操作符
6. Rxjava2 : 筛选操作符
7. Rxjava2 : 合并操作符
8. Rxjava2 : do操作符
9. Rxjava2 : error处理
10. Rxjava2 : 重试
11. Rxjava2 : 线程切换

流程

Observable
->
doOnSubscribe
元素1
->
doOnEach
->
doOnNext
->
onNext(Observer)
->
doAfterNext

元素2
->
doOnEach
->
doOnNext
->
onNext(Observer)
->
doAfterNext

-> doOnEach
->{
  doOnError -> onError(Observer)
  doOnComplete -> onComplete(Observer)
}

->
doOnFinally

Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onError(new NullPointerException());
                e.onNext(3);
            }
        })
                .doOnSubscribe(new Consumer<Disposable>() {
                    @Override
                    public void accept(Disposable disposable) throws Exception {
                        Log.d(TAG, "doOnSubscribe");
                    }
                })
                .doOnEach(new Consumer<Notification<Integer>>() {
                    @Override
                    public void accept(Notification<Integer> integerNotification) throws Exception {
                        Log.d(TAG, "doOnEach");
                    }
                })
                .doOnNext(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "doOnNext");
                    }
                })
                .doAfterNext(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.d(TAG, "doAfterNext");
                    }
                })
                .doOnError(new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        Log.d(TAG, "doOnError");
                    }
                })
                .doFinally(new Action() {
                    @Override
                    public void run() throws Exception {
                        Log.d(TAG, "doFinally");
                    }
                })
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.d(TAG, "observe : onNext : " + integer);
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d(TAG, "observe : onError");
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "observe : onComplete");
                    }
                });

log

02-13 11:08:46.821 19940-19940/... D/SplashActivity: doOnSubscribe
02-13 11:08:46.821 19940-19940/... D/SplashActivity: doOnEach
02-13 11:08:46.821 19940-19940/... D/SplashActivity: doOnNext
02-13 11:08:46.821 19940-19940/... D/SplashActivity: observe : onNext : 1
02-13 11:08:46.821 19940-19940/... D/SplashActivity: doAfterNext
02-13 11:08:46.821 19940-19940/... D/SplashActivity: doOnEach
02-13 11:08:46.821 19940-19940/... D/SplashActivity: doOnNext
02-13 11:08:46.821 19940-19940/... D/SplashActivity: observe : onNext : 2
02-13 11:08:46.821 19940-19940/... D/SplashActivity: doAfterNext
02-13 11:08:46.831 19940-19940/... D/SplashActivity: doOnEach
02-13 11:08:46.831 19940-19940/... D/SplashActivity: doOnError
02-13 11:08:46.831 19940-19940/... D/SplashActivity: observe : onError
02-13 11:08:46.831 19940-19940/... D/SplashActivity: doFinally

doOnEach

• doOnEach
  只要有元素发射,均会经历doOnEach
  元素类型:
  1.正常的元素
  2.error
  3.complete