RxJava学习之旅二 2.0.1版本和1.0简单实用的对比讲解

第一篇用了1.0版本的RxJava 从现在开始来使用2.0.1，还是使用新的好一些

compile 'io.reactivex.rxjava2:rxjava:2.0.1'
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'

解释一下其中两个陌生的玩意：ObservableEmitter和Disposable.
ObservableEmitter： Emitter是发射器的意思，那就很好猜了，这个就是用来发出事件的，它可以发出三种类型的事件，通过调用emitter的onNext(T value)、onComplete()和onError(Throwable error)就可以分别发出next事件、complete事件和error事件。

这里使用要注意有使用规则
ObservableEmitter 可以发送无限个onNext,但是complete和onError时间只能发送其中一个

Observer<String> observer = new Observer<String>() {
        @Override
        public void onSubscribe(Disposable d) {

        }

        @Override
        public void onNext(String value) {

        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onComplete() {

        }
    };

但是只要在收到onError或者onComplete后就不会再接收onNext,这时候会出现的情况就是被观察的ObservableEmitter反射器可能还会继续发送onNext但是接收这Observer不会再接收onNext时间。

注: 关于onComplete和onError唯一并且互斥这一点, 是需要自行在代码中进行控制, 如果你的代码逻辑中违背了这个规则, 并不一定会导致程序崩溃. 比如发送多个onComplete是可以正常运行的, 依然是收到第一个onComplete就不再接收了, 但若是发送多个onError, 则收到第二个onError事件会导致程序会崩溃.

作者：Season_zlc
链接：http://www.jianshu.com/p/464fa025229e
來源：简书
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

那么Disposable用来干什么的呢，它的作用就是让Observer不在接收onNext onComplete和onError事件
比如在一连串业务操作中 如果中途出现什么状况需要停掉正在进行的业务 就可以使用这个,比如像下面这样用

    Observer<String> observer = new Observer<String>() {
        private Disposable mDisposable = null;
        @Override
        public void onSubscribe(Disposable d) {
            mDisposable = d;
        }

        @Override
        public void onNext(String value) {
            if ("111".equals(value)) {
                if (mDisposable != null && !mDisposable.isDisposed()) {
                    mDisposable.dispose();
                    mDisposable = null;
                }
            }
        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onComplete() {

        }
    };

另外还有一个不同的地方是

public final Disposable subscribe() {}
public final Disposable subscribe(Consumer<? super T> onNext) {}
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {} 
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) {}
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe) {}
public final void subscribe(Observer<? super T> observer) {}

observable.subscribe();被重载了很多个 而且和1.0的也是有一些区别 那具体怎么用呢

1、没有参数的 调用后Observer不会任何事件，而observable的subscribe中的事件调用会依次执行

Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
        @Override
        public void subscribe(ObservableEmitter<String> e) throws Exception {
            // 这里相当于1.0中的call函数  调用的顺序还是和1.0的时候一样的 只是使用的类和函数名有了少许改变
            //ObservableEmitter 时间发射器  其实可以理解为一个观察者对象
            log("Observable   CurrentThread : " + Thread.currentThread().getName());
            e.onNext("Hello");
            e.onComplete();// 1.0中函数名为onCompleted()
        }
    });
    
打印日志结果
08-08 11:18:39.713 24454-24454/com.example.rxjavademo I/MainActivity----Test: Observable   CurrentThread : main

2、subscribe(Consumer<? super T> onNext) 参数为一个Consumer对象的 从下面的代码中可以看出，这里调用Observer只关心收到的onNext()不关系onComplete和onError 其实这里和上一篇讲的1.0中的Action1这个类的作用很相似就是对onNext的函数封装成了一个对于对象来使用，因为Consumer的内部函数accept(T t)就是一个有参无返回的函数 而onNext(T t) 刚好也是同样的，所以这样你是不是发现onError(T t) 也是同样的是这样的。请继续看第三条

public interface Consumer<T> {
    /**
     * Consume the given value.
     * @param t the value
     * @throws Exception on error
     */
    void accept(T t) throws Exception;
}


Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
        @Override
        public void subscribe(ObservableEmitter<String> e) throws Exception {
            // 这里相当于1.0中的call函数  调用的顺序还是和1.0的时候一样的 只是使用的类和函数名有了少许改变
            //ObservableEmitter 时间发射器  其实可以理解为一个观察者对象
            log("Observable   CurrentThread : " + Thread.currentThread().getName());
            e.onNext("Hello");
            e.onComplete();// 1.0中函数名为onCompleted()
        }
    });

    Consumer<String> consumer = new Consumer<String>() {
        @Override
        public void accept(String s) throws Exception {
            log("Consumer   CurrentThread : " + Thread.currentThread().getName());
            log("onNext :" + s);
        }
    };

    observable.subscribe(consumer);
日志打印结果
08-08 11:23:37.163 28780-28780/com.example.baiguosong.rxjavademo I/MainActivity----Test: Observable   CurrentThread : main
08-08 11:23:37.163 28780-28780/com.example.baiguosong.rxjavademo I/MainActivity----Test: Consumer   CurrentThread : main
08-08 11:23:37.163 28780-28780/com.example.baiguosong.rxjavademo I/MainActivity----Test: onNext :Hello

3、subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError)。上面第二条中讲到Consumer就是对onNext(T t) 和onError(Throwabe t)的封装，看果不其然两个参数的时候就关系的是onNext和onError看日志就知道onNext和onError都执行了

Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
        @Override
        public void subscribe(ObservableEmitter<String> e) throws Exception {
            // 这里相当于1.0中的call函数  调用的顺序还是和1.0的时候一样的 只是使用的类和函数名有了少许改变
            //ObservableEmitter 时间发射器  其实可以理解为一个观察者对象
            log("Observable   CurrentThread : " + Thread.currentThread().getName());
            e.onNext("Hello");

// e.onComplete();// 1.0中函数名为onCompleted()

            e.onError(new NullPointerException());
        }
    });

    Consumer<String> onNextConsumer = new Consumer<String>() {
        @Override
        public void accept(String s) throws Exception {
            log("Consumer   CurrentThread : " + Thread.currentThread().getName());
            log("onNext :" + s);
        }
    };

    Consumer<Throwable> onErrorConsumer = new Consumer<Throwable>() {
        @Override
        public void accept(Throwable e) throws Exception {
            log("onError : " + e.toString());
        }
    };

    observable.subscribe(onNextConsumer,onErrorConsumer);
    
    日志打印结果
    08-08 11:35:49.043 7751-7751/com.example.baiguosong.rxjavademo I/MainActivity----Test: Observable   CurrentThread : main
    08-08 11:35:49.043 7751-7751/com.example.baiguosong.rxjavademo I/MainActivity----Test: Consumer   CurrentThread : main
    08-08 11:35:49.043 7751-7751/com.example.baiguosong.rxjavademo I/MainActivity----Test: onNext :Hello
    08-08 11:35:49.043 7751-7751/com.example.baiguosong.rxjavademo I/MainActivity----Test: onError : java.lang.NullPointerException

4、subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete)经过前面的讲解那么三个参数就一目了然了 第三个参数Action肯定就是对应的onComplete()的函数的封装了 无参无返回值，老规矩还是直接上代码

public interface Action {
    /**
     * Runs the action and optionally throws a checked exception.
     * @throws Exception if the implementation wishes to throw a checked exception
     */
    void run() throws Exception;
}

 Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
        @Override
        public void subscribe(ObservableEmitter<String> e) throws Exception {
            // 这里相当于1.0中的call函数  调用的顺序还是和1.0的时候一样的 只是使用的类和函数名有了少许改变
            //ObservableEmitter 时间发射器  其实可以理解为一个观察者对象
            log("Observable   CurrentThread : " + Thread.currentThread().getName());
            e.onNext("Hello");

// e.onComplete();// 1.0中函数名为onCompleted()

// e.onError(new NullPointerException());
e.onComplete();
}
});

    Consumer<String> onNextConsumer = new Consumer<String>() {
        @Override
        public void accept(String s) throws Exception {
            log("Consumer   CurrentThread : " + Thread.currentThread().getName());
            log("onNext :" + s);
        }
    };

    Consumer<Throwable> onErrorConsumer = new Consumer<Throwable>() {
        @Override
        public void accept(Throwable e) throws Exception {
            log("onError : " + e.toString());
        }
    };

    Action onCompleteAction = new Action() {
        @Override
        public void run() throws Exception {
            log("onComplete : run");
        }
    };

    observable.subscribe(onNextConsumer,onErrorConsumer,onCompleteAction);
    
    日志打印结果
    08-08 11:43:57.733 14894-14894/com.example.baiguosong.rxjavademo I/MainActivity----Test: Observable   CurrentThread : main
    08-08 11:43:57.733 14894-14894/com.example.baiguosong.rxjavademo I/MainActivity----Test: Consumer   CurrentThread : main
    08-08 11:43:57.733 14894-14894/com.example.baiguosong.rxjavademo I/MainActivity----Test: onNext :Hello
    08-08 11:43:57.733 14894-14894/com.example.baiguosong.rxjavademo I/MainActivity----Test: onComplete : run

5、subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe) 这里又多了一个参数Consumer<? super Disposable> 通过前面的举例说明这里的第四个参数就没什么难度了吧 Disposable这个不就是前面讲到的那个让Observer不再接收事件的函数嘛

6、subscribe(Observer<? super T> observer) 这个也很好理解了 直接看代码

Observer<String> observer = new Observer<String>() {
        private Disposable mDisposable = null;
        @Override
        public void onSubscribe(Disposable d) {
            mDisposable = d;
        }

        @Override
        public void onNext(String value) {
            log("onNext :" + value);

        }

        @Override
        public void onError(Throwable e) {
            log("onError :" + e.toString());
        }

        @Override
        public void onComplete() {
            log("onComplete :");
        }
    };

    observable.subscribe(observer);
    
    日志打印结果
    08-08 11:50:43.103 20994-20994/com.example.baiguosong.rxjavademo I/MainActivity----Test: Observable   CurrentThread : main
    08-08 11:50:43.103 20994-20994/com.example.baiguosong.rxjavademo I/MainActivity----Test: onNext :Hello
    08-08 11:50:43.103 20994-20994/com.example.baiguosong.rxjavademo I/MainActivity----Test: onComplete :

以上内容就是2.0.1版本和1.0版本的对比讲解后面的内容都会基于2.0.1版本来学习研究，欢迎拍砖指教。

注：部分内容是看了这位大佬的 传送门
http://www.jianshu.com/p/464fa025229e