Android RxJava 2.x入门例子详解（一）

前言

关于RxJava的介绍这里就不多说了，网上有很多相关的资料。
但有一点需要说明一下，很多同学可能在网上找到很多RxJava 1.X的教程，那么1.X和2.X有什么区别？学习2.X前需不需要先学习1.X？
其实1.X和2.X有很大的改变，如果你已学习过1.X，那么恭喜你，你只需要看看2.X有什么更新就可以了。如果你没学习过1.X，那么也不需要担心，你可以直接跳过1.X，来学习2.X。
所以本教程是直接使用2.X，概念性的东西这里也不多说，本教程直接使用例子一步一步带大家入门。

先在Android Studio 项目添加Gradle配置：

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

可能你想添加最新的版本，那在哪里可以找到最新的版本呢？答案在这里：
https://github.com/ReactiveX/RxAndroid

入门例子：

//create创建一个上游Observable（被观察者）
Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
                Log.d(TAG, "Observable发出：1");
                e.onNext(1);//向下游（观察者）发射事件
                Log.d(TAG, "Observable发出：2");
                e.onNext(2);
                Log.d(TAG, "Observable发出：3");
                e.onNext(3);
                Log.d(TAG, "Observable发出：Complete");
                e.onComplete();
                Log.d(TAG, "Observable发出：4");
                e.onNext(4);//虽然无法接收事件，但发送事件还是继续的
            }

        });
        //创建一个下游Observer（观察者）
        Observer<Integer> observer = new Observer<Integer>() {

            @Override
            public void onSubscribe(@NonNull Disposable d) {
                Log.d(TAG, "subscribe");
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                Log.d(TAG, "onNext收到：" + integer);
            }

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

            @Override
            public void onComplete() {
                Log.d(TAG, "Observer complete");
            }
        };
        //建立连接（订阅），开始发送事件
        observable.subscribe(observer);

ObservableEmitter是发射器，就是用来发出事件的，它可以发出三种类型的事件。通过调用emitter的onNext(T value)、onComplete()和onError(Throwable error)就可以分别发出next事件、complete事件和error事件。

写成RxJava引以为傲的链式操作：

//create创建一个上游Observable（被观察者）
Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
                Log.d(TAG, "Observable发出：1");
                e.onNext(1);//向下游（观察者）发射内容1
                Log.d(TAG, "Observable发出：2");
                e.onNext(2);
                Log.d(TAG, "Observable发出：3");
                e.onNext(3);
                Log.d(TAG, "Observable发出：Complete");
                e.onComplete();
                Log.d(TAG, "Observable发出：4");
                e.onNext(4);//虽然无法接收事件，但发送事件还是继续的
            }
        }).subscribe(new Observer<Integer>() {
            //创建一个下游Observer（观察者）
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                Log.d(TAG, "subscribe");
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                Log.d(TAG, "onNext收到：" + integer);
            }

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

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

1、上游可以发送无限个onNext，下游也可以接收无限个onNext。
2、 当上游发送了一个onComplete后, 上游onComplete之后的事件将会继续发送，而下游收到onComplete事件之后将不再继续接收事件。
3、当上游发送了一个onError后，上游onError之后的事件将继续发送，而下游收到onError事件之后将不再继续接收事件。
4、上游可以不发送onComplete或onError。
5、最为关键的是onComplete和onError必须唯一并且互斥, 即不能发多个onComplete, 也不能发多个onError，也不能先发一个onComplete, 然后再发一个onError，反之亦然。

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

入门例子：

//create创建一个上游 Observable（被观察者）
Observable.create(new ObservableOnSubscribe<Integer>() {

            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
                Log.d(TAG, "Observable发出：1");
                e.onNext(1);//向下游（观察者）发射内容1
                Log.d(TAG, "Observable发出：2");
                e.onNext(2);
                Log.d(TAG, "Observable发出：3");
                e.onNext(3);
                Log.d(TAG, "Observable发出：4");
                e.onNext(4);
                Log.d(TAG, "Observable发出：Complete");
                e.onComplete();
                Log.d(TAG, "Observable发出：5");
                e.onNext(5);//虽然无法接收事件，但发送事件还是继续的
            }
        }).subscribe(new Observer<Integer>() {
            //创建一个下游 Observer（观察者）

            private Disposable mDisposable;
            private int i;
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                Log.d(TAG, "subscribe");
                mDisposable=d;
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                Log.d(TAG, "onNext收到：" + integer);
                i++;
                if (i==2){
                    Log.d(TAG, "onNext：dispose");
                    mDisposable.dispose();//取消订阅，不再接收事件
                    Log.d(TAG, "onNext isDisposed : " + mDisposable.isDisposed());
                }
            }

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

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

下游调用dispose()，取消收收事件，但并不会导致上游不再继续发送事件, 上游可以继续发送剩余的事件。

subscribe()有多个重载：

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);

带有一个Consumer参数的方法，表示下游只关心onNext事件，其他的事件不管。
因此，如果只需要onNext事件可以这么写：

//create创建一个上游 Observable（被观察者）
Observable.create(new ObservableOnSubscribe<Integer>() {

            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
                Log.d(TAG, "Observable发出：1");
                e.onNext(1);//向下游（观察者）发射内容1
                Log.d(TAG, "Observable发出：2");
                e.onNext(2);
                Log.d(TAG, "Observable发出：3");
                e.onNext(3);
                Log.d(TAG, "Observable发出：4");
                e.onNext(4);
                Log.d(TAG, "Observable发出：Complete");
                e.onComplete();
                Log.d(TAG, "Observable发出：5");
                e.onNext(5);//虽然无法接收事件，但发送事件还是继续的
            }
        }).subscribe(new Consumer<Integer>() {

            @Override
            public void accept(Integer integer) throws Exception {
                Log.d(TAG, "onNext收到：" + integer);
            }
        });
最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。

推荐阅读更多精彩内容

  • RxJava 2.x 学习2（转载）
    怎么如此平静, 感觉像是走错了片场.为什么呢, 因为上下游工作在同一个线程呀骚年们! 这个时候上游每次调用emit...
    Young1657阅读 5,446评论 2赞 1
  • RxJava 2.x 学习（转载）
    转载自：https://xiaobailong24.me/2017/03/18/Android-RxJava2.x...
    Young1657阅读 6,094评论 1赞 9
  • RxJava 知识梳理(3) - RxJava2 基础知识小结
    前言 首先要感谢 Season_zlc 的一系列RxJava2的教程，关于上游、下游、水缸的类比，让我对于整个Rx...
    泽毛阅读 8,735评论 0赞 11
  • RxJava2.0初探(一)
    前言上个月RxJava2正式版发布了，但目前国内的资料还比较少，以前使用过RxJava1只需要看看更新文档就知道怎...
    加油码农阅读 3,477评论 0赞 1
  • 酒香
    雨滴落下，霓虹暗淡 雨滴抚摸着她的脸庞 却用利刃刺向她的心 在她脸上的雨水啊 却是炽热的 一点一滴 在他嘴中 散发...
    citron阅读 1,423评论 0赞 0