Rxjava2最全面的解析

时间 2019-11-30

原文原文链接

前言

因为公司从新规划的部门，我调到了另一个部门，因此负责的项目也换了，仔细看了下总体的项目，rxjava+retrofit。总体的一套。众所周知，rxjava+retrofit是目前网上最流行的网络解析框架。而目前网络上的文章大多仍是关于rxjava1的。关于RxJava2的少之又少，因而，便有了此文。java

此文的目的有三个： 1. 给对 RxJava2感兴趣的人一些入门的指引 2. 给正在使用 RxJava2但仍然心存疑惑的人一些更深刻的解析 3.给想从RxJava1替换成RxJava2的人给出直接的对比。react

RxJava概念介绍

RxJava=reactive+extension。那么接下来我会分别对这两点进行总体的介绍。reactive又称reactive programming。也就是响应式编程。在往简单的说，rxjava能够很方便的处理线程切换的问题。说到这个，咱们就会想到异步操做。handler？AsyncTask？但你要知道，随着请求的数量愈来愈多，代码逻辑将会变得愈来愈复杂。而rxjava却仍能够保持清晰的逻辑。它的原理就是建立一个Observable对象来搞事情。而后使用各类操做符经过建造者模式创建成一系列的链式操做。就如流水线同样，把事情搞完。而后发射给Observer进行处理。android

观察者模式

rxjava的实现主要是经过观察者模式实现的。那么什么是观察者模式，我这边作一个简单的介绍。数据库

栗子：观察者对被观察者进行一个简单，当被观察者被改变时，要当即作出反应。好比，你认为隔壁老王和你媳妇有一腿，但却没证据，此时，只要当隔壁老王进了你媳妇房门的时候，你就要去捕获他。在这个例子中，你是观察者，老王是被观察者。（记得当初我常常搞反了）。那么，观察者模式是不是一对一呢？很明显不是的，就上面的例子，你能够叫三千城管监听着老王。只要他有不轨之心。就打断他的第三条腿。也就是说多个观察者对应一个被观察者。字看累了来看图：编程

其实在android中也有不少自带的观察者模式。最明显的莫过于点击事件。说个最简单的例子，点击按钮后弹一个吐司。那么，咱们在点击按钮的时候，告知系统，此时，我须要弹一个吐司。那么就这么弹出来了。那么，这个时候问题来了。我是否须要实时去监听这个按钮呢？答案是不须要的。这就和前面的举例有的差距了。换句话说。我只要在此按钮进行点击时进行监听就能够了。这种操做被称为订阅。也就是说Button经过setOnClickListener对OnclickListener进行了订阅了操做，来监听onclick方法。json

extension

不只支持事件序列，还支持数据流。事件-->动态的，没法预知，例如：事件点击，服务器的推送等等数据流-->静态的，可预知的，例如：读取本地文件，播放音视频等等。数组
经过操做符对中间事件的处理。缓存
线程操做的便捷。关于这些具体的实现。我会在后面一一举例。bash

RxJava1与RxJava2的区别

说到区别，可能有的小伙伴会问，我没看过rxjava1。能够直接看rxjava2么。我的以为没必要要，由于 rxjava2.x 是按照 Reactive-Streams specification 规范彻底的重写的，彻底独立于 rxjava1.x 而存在，它改变了以往 rxjava1的用法。换句话说，我学java需不须要先学C语言同样。服务器

那么二者的区别体如今哪呢？主要是以下几个方面：

空指针问题这应该是一个很大的变化，用过rxjava1的人都知道，咱们能够在发射事件的时候传入NULL。但这在rxjava2中是不存在的。不信你试试？分分钟给你来一个NullPointerExpection。
Function相关的在rxjava1中，咱们有各类Func1，Func2......，但在rxjava2中只有Function了。依旧记得看凯哥的文章的时候把我整蒙了。愣是没发现，后来才注意到被替换了。而且，他们都增长了throw exception。
背压—backpressure 关于backpressure，这个就厉害了。厉害到我都不懂了。好了，开个玩笑，咱们继续说。咱们知道在Rxjava1中Observable对backpressure是支持的。但在Rxjava2中Observable取消了对backpressure的支持。而且引进了一个叫作Flowable的来支持backpressure。

那么什么是背压：听不懂的含义：上游的生产速度大于下游的处理速度，致使下游处理不急，这种操做被称为backpressure。

这种状况看似很常见，但实际上，这种状况并不常见，或者能够说成很是罕见。那么遇到了怎么办？若是它出现了，直接丢弃。what the fuck？你tm在逗我？但事实就是这样，若是咱们在开发过程当中，遇到了backpressure，咱们就应该丢弃它。

听得懂的含义：对于可丢弃的事件，上游生产速度过快致使事件堆积，当堆积到超出buffer上限的时候，就叫作backpressure。

处理方案是什么：一、丢弃新事件；二、不丢弃，继续堆积。（忽略了backpressure，至关于Observable）。

适合backpressure的状况：在线直播流：好比说，正在直播的时候，忽然网络出现了卡顿，页面卡住了。那么当网络好了以后确定不会是在接着以前的页面继续的，就至关于，你网络卡了多久，他就丢弃了多长时间的数据。

backpressure的关键点是什么：不可控，可丢弃。

基本使用

讲了一大堆理念知识，接下来就是开工干活了。那么关于Rxjava2的基本实现主要是三点：建立Observable，建立Observer，进行绑定。那么咱们一个个的看。

建立Observable

Observable是什么？观察者仍是被观察者？我又忘了。哈哈。开个玩笑，固然是后者了。为何是先建立Observable而不是Observer？固然了，前后顺序的无所谓的。可是考虑到后面的链式调用。因此我这边就先写了先建立Observable了。

Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                emitter.onNext("Hello");
                emitter.onNext("Rxjava2");
                emitter.onNext("My name is Silence");
                emitter.onNext("What's your name");
                //一旦调用onComplete,下面将不在接受事件
                emitter.onComplete();
            }
        });复制代码

如今我来解释一下上面的ObservableEmitter究竟是什么。字面意思是可观察的发射器。没错，这个就是被观察者用来发送事件的。它能够发出三种类型的事件，经过调用emitter的onNext(T value)、onError(Throwable error)和onComplete()就能够分别发出next事件、error事件和complete事件。至于这三个事件到底什么意思。不急，咱们后面说。

建立Observer

如今咱们来建立一个观察者，它决定了在观察中到底应该有着什么样的行为操做。

Observer<String> observer = new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "onSubscribe: " + d);
                result += "onSubscribe: " + d + "\n";
            }

            @Override
            public void onNext(String string) {
                Log.i(TAG, "onNext: " + string);
                result += "onNext: " + string + "\n";
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "onError: " + e);
                result += "onError: " + e + "\n";
            }

            @Override
            public void onComplete() {
                Log.i(TAG, "onComplete: ");
                result += "onComplete: " + "\n";
            }
        };复制代码

其中onSubscribe、onNext、onError和onComplete是必要的实现方法，其含义以下：

onSubscribe：它会在事件还未发送以前被调用，能够用来作一些准备操做。而里面的Disposable则是用来切断上下游的关系的。
onNext：普通的事件。将要处理的事件添加到队列中。
onError：事件队列异常，在事件处理过程当中出现异常状况时，此方法会被调用。同时队列将会终止，也就是不容许在有事件发出。
onComplete：事件队列完成。rxjava不只把每一个事件单独处理。并且会把他们当成一个队列。当再也不有onNext事件发出时，须要触发onComplete方法做为完成标识。

进行Subscribe

订阅其实只须要一行代码就够了：

observerable.subscribe(Observer);复制代码

运行一个看看效果先：

和以前介绍的同样，先调用onSubscribe，而后走了onNext，最后以onComplete收尾。

神奇的操做符

对于rxjava来讲，有一句话，我以为说的很对，叫作：若是你天天研究一个操做符，最少一个半月，若是你想理解原理。最少半年。换句话说，有关rxjava的知识彻底能够写一本书。那么本文确定不会讲那么细。在这边我会给大家介绍一些经常使用的操做符。保证平常开发的流程足矣。

建立操做符

通常建立操做符是指，刚开始建立观察者的时候调用的。在基本使用中我已经介绍了create操做符，那么这边咱们就要说到just，fromarray和interval了。

just

此操做符是将传入的参数依次发出来。

Observable observable = Observable.just("Hello", "Rxjava2", "My name is Silence","What's your name");
// 将会依次调用：
// onNext("Hello");
// onNext("Rxjava2");
// onNext("My name is Silence");
// onNext("What's your name");
// onCompleted();复制代码

fromarray

将传入的数组经过坐标一次发送出去。

String[] words = {"Hello", "Rxjava2", "My name is Silence","What's your name"};
Observable observable = Observable.from(words);
// 将会依次调用：
// onNext("Hello");
// onNext("Rxjava2");
// onNext("My name is Silence");
// onNext("What's your name");
// onCompleted();复制代码

interval

这个其实就是定时器，用了它你能够抛弃CountDownTimer了。如今咱们看看怎么用：

Observable.interval(2, TimeUnit.SECONDS).subscribe(
                new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Log.i(TAG, "accept: "+aLong.intValue());
                    }
                }
        );复制代码

咱们看看结果：

上面就是咱们每隔2s打印一次long的值。

变换操做符

变换操做符的做用是对Observable发射的数据按照必定规则作一些变换操做，而后讲变换后的数据发射出去。变换操做符有map，flatMap，concatMap，switchMap，buffer，groupBy等等。这里咱们会讲解最经常使用的map，flatMap、concatMap以及compose。

map

map操做符经过指定一个Function对象，将Observable转换为一个新的Observable对象并发射，观察者将收到新的Observable处理。直接上代码：

Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onNext(4);
            }
        }).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                return "This is result " + integer + "\n";
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String str) throws Exception {
                Log.i("--->", "accept: "+str);
                string += str;
            }
        });
        tv_first.setText(string);复制代码

输入结果以下：

仔细看，map()方法中，咱们把一个integer对象转换成了一个String对象。而后当map()调用结束时，事件的参数类型也从integer转换成了String。这就是最多见的变换操做。

flatMap

flatmap的操做符是将Observable发射的数据集合变成一个Observable集合。也就是说它能够讲一个观察对象变换成多个观察对象，可是并不能保证事件的顺序。想保证事件的顺序？那你过会看下面降到的concatMap。

那么什么叫做数据集合变成一个Observable集合呢？仍是用上面的例子，我有一组integer集合。我想转换成string集合怎么办？那就继续看代码：

Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }
        }).flatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                final List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("I am value " + integer + "\n");
                }
                return Observable.fromIterable(list);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.i("--->", "accept: "+s);
                string += s;
            }
        });
        tv_first.setText(string);复制代码

咱们来看结果：

打住打住，是否是有问题？WTF？有啥问题？还记不记得我上面说过flatMap不能保证事件执行顺序。那么这边事件为何都是按顺序执行的？不急，咱们在发射事件的时候给他加一个延迟在看看结果：

Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }
        }).flatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                final List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("I am value " + integer + "\n");
                }
                return Observable.fromIterable(list).delay(100,TimeUnit.MILLISECONDS);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.i("--->", "accept: "+s);
                string += s;
            }
        });
        tv_first.setText(string);复制代码

咱们在当他发射事件的时候给他加一个100ms的延迟看看结果：

看到没有，我说啥的？不能保证执行顺序。因此万事容我慢慢道来。先喝杯茶压压惊。咱们在接着往下讲。

concatMap

上面我也介绍了concatMap。除了保证了执行顺序，其余都和concatMap一毛同样。你说保证就保证啊。您先喝杯茶，接着往下看：

Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }
        }).concatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                final List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("I am value " + integer + "\n");
                }
                return Observable.fromIterable(list).delay(1000,TimeUnit.MILLISECONDS);
//                return Observable.fromIterable(list);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.i("--->", "accept: "+s);
                string += s;
            }
        });
        tv_first.setText(string);复制代码

为了咱们能看的更明显一点，我这边直接设置了一秒钟的延迟。下面咱们来看效果图：

能够从执行顺序和打印时间看出，的的确确是延迟了一秒钟。

compose

这个操做符就很厉害了。他的变换是怎么作的呢？咱们知道rxjava是经过建造者的模式经过链式来调用起来的。那么多个链式就须要多个Observable。而这个操做符就是把多个Observable转化成一个Observable。听起来是否是很厉害~。具体如何操做，咱们接着看：


    public  <T> ObservableTransformer<T, T> applyObservableAsync() {
        return new ObservableTransformer<T, T>() {
            @Override
            public ObservableSource<T> apply(Observable<T> upstream) {
                return upstream.subscribeOn(Schedulers.io())
                        .observeOn(AndroidSchedulers.mainThread());
            }
        };
    }复制代码

上面代码能够看出，我把子线程和主线程进行了一个封装，而后返回了一个ObservableTransformer对象。那么咱们只要这边作就能够了：

Observable.just(1, 2, 3, 4, 5, 6)
                .compose(this.<Integer>applyObservableAsync())
                .subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer strings) throws Exception {
                Log.i("-->", "accept: " + strings);
                string += strings;
            }
        });
        tv_first.setText(string);复制代码

过滤操做符

过滤操做符用于过滤和选择Observable发射的数据序列。让Observable只返回知足咱们条件的数据。过滤操做符有buffer，filter，skip，take，skipLast，takeLast等等，这边我会介绍到filter，buffer，skip，take，distinct。

filter

filter操做符是对源Observable产生的结果进行有规则的过滤。只有知足规则的结果才会提交到观察者手中。例如：

Observable.just(1,2,3).filter(new Predicate<Integer>() {
            @Override
            public boolean test(Integer integer) throws Exception {
                return integer < 3;
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer s) throws Exception {
                Log.i("--->", "accept: " + s);
                string += s;
            }
        });
        tv_first.setText(string);
    }复制代码

代码很简单，咱们发送1，2，3；可是咱们加上一个filter操做符，让它只返回小于3的的内容。那么咱们来看一下结果：

distinct

这个操做符其实就更简单了。好比说，我要在一组数据中去掉重复的内容，就要用到它。也就是去重。它只容许尚未发射的数据项经过。发射过的数据项直接pass。

Observable.just(1,2,3,4,2,3,5,6,1,3)
                .distinct().subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer s) throws Exception {
                Log.i("--->", "accept: " + s);
                string += s;
            }
        });
        tv_first.setText(string);复制代码

那么输出结果就很简单了：

buffer

这个其实也不难，主要是缓存，把源Observable转换成一个新的Observable。这个新的Observable每次发射的是一组List，而不是单独的一个个的发送数据源。

Observable.just(1,2,3,4,5,6)
                .buffer(2).subscribe(new Consumer<List<Integer>>() {
            @Override
            public void accept(List<Integer> strings) throws Exception {
                for (Integer integer : strings) {
                    Log.i("-->", "accept: "+integer);
                    string+=strings;
                }
                Log.i("-->", "accept: ----------------------->");
            }
        });
        tv_first.setText(string);复制代码

咱们让他每次缓存2个，下面咱们来看结果：

skip 、take

skip操做符将源Observable发射过的数据过滤掉前n项，而take操做则只取前n项；另外还有skipLast和takeLast则是从后往前进行过滤。先来看看skip操做符。

Observable.just(1, 2, 3, 4, 5, 6)
                .skip(2).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer strings) throws Exception {
                Log.i("-->", "accept: " + strings);
                string += strings;
            }
        });
        tv_first.setText(string);复制代码

结果以下：

接下来咱们把skip换成take看看。

Observable.just(1, 2, 3, 4, 5, 6)
                .take(3).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer strings) throws Exception {
                Log.i("-->", "accept: " + strings);
                string += strings;
            }
        });
        tv_first.setText(string);复制代码

结果以下：

组合操做符

merge

merge是将多个操做符合并到一个Observable中进行发射，merge可能让合并到Observable的数据发生错乱。（并行无序）

Observable<Integer> observable1=Observable.just(1,2,3);
        Observable<Integer> observable2=Observable.just(1,2,3);
        Observable.merge(observable1,observable2).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i(TAG, "accept: "+integer);
            }
        });复制代码

结果以下：

concat

将多个Observable发射的数据进行合并而且发射，和merge不一样的是，merge是无序的，而concat是有序的。（串行有序）没有发射完前一个它必定不会发送后一个。

Observable<Integer> observable1=Observable.just(1,2,3);
        Observable<Integer> observable2=Observable.just(4,5,6);
        Observable.concat(observable1,observable2).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i(TAG, "accept: "+integer);
            }
        });复制代码

结果以下：

zip

此操做符和合并多个Observable发送的数据项，根据他们的类型就行从新变换，并发射一个新的值。

Observable<Integer> observable1=Observable.just(1,2,3);
        Observable<String> observable2=Observable.just("a","b","c");
        Observable.zip(observable1, observable2, new BiFunction<Integer, String, String>() {

            @Override
            public String apply(Integer integer, String s) throws Exception {
                return integer+s;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.i(TAG, "apply: "+s);
            }
        });复制代码

结果以下：

concatEager

前面说道串行有序，而concatEager则是并行且有序。咱们来看看若是修改：

Observable<Integer> observable1=Observable.just(1,2,3);
        Observable<String> observable2=Observable.just("a","b","c");
        Observable.concatEager(Observable.fromArray(observable1,observable2)).subscribe(new Consumer<Serializable>() {
            @Override
            public void accept(Serializable serializable) throws Exception {
                Log.i(TAG, "accept: "+serializable);
            }
        });复制代码

结果以下：

线程控制

其实线程控制也是一种操做符。但它不属于建立、变换、过滤。因此我这边把它单独拉出来说讲。

subscribeOn是指上游发送事件的线程。说白了也就是子线程。屡次指定上游的线程只有第一次指定的有效, 也就是说屡次调用subscribeOn() 只有第一次的有效, 其他的会被忽略。

observerOn是指下游接受事件的线程。也就是主线程。屡次指定下游的线程是能够的, 也就是说每调用一次observeOn() , 下游的线程就会切换一次。

举个栗子：

Observable.just(1, 2, 3, 4) // IO 线程，由 subscribeOn() 指定
    .subscribeOn(Schedulers.io())
    .observeOn(Schedulers.newThread())
    .map(mapOperator) // 新线程，由 observeOn() 指定
    .observeOn(Schedulers.io())
    .map(mapOperator2) // IO 线程，由 observeOn() 指定
    .observeOn(AndroidSchedulers.mainThread) 
    .subscribe(subscriber);  // Android 主线程，由 observeOn() 指定复制代码

在RxJava中, 已经内置了不少线程选项供咱们选择, 例若有

Schedulers.io() ：I/O操做（读写文件、数据库，及网络交互等）所使用的Scheduler。行为模式和newThread()差很少。区别在于io()的内部实现是用一个无数量上限的线程池。能够重用空闲的线程。所以多数状况下io()比newThread()更有效率。
Schedulers.immediate()：直接在当前线程运行。
Schedulers.computation() ：计算所使用的Scheduler，例如图形的计算。这个Scheduler使用固定线程池，大小为CPU核数。不要把I/O操做放在computation中。不然I/O操做的等待会浪费CPU。
Schedulers.newThread()：表明一个常规的新线程
Schedulers.trampoline()：当咱们想在线程执行一个任务时（不是当即执行），能够用此方法将它加入队列。这个调度器将会处理它的队列而且按序执行队列中的每个任务。
AndroidSchedulers.mainThread() ：表明Android的主线程

这些内置的Scheduler已经足够知足咱们开发的需求, 所以咱们应该使用内置的这些选项, 在RxJava内部使用的是线程池来维护这些线程, 全部效率也比较高。

与Retrofit结合

就目前开发角度而言，retrofit能够说是最火的网络框架。其缘由我认为有两点，第一：能够和okhttp结合。第二：能够和rxjava结合。也就是说Retrofit 除了提供了传统的 Callback 形式的 API，还有 RxJava 版本的 Observable 形式 API。

若是须要使用retrofit，咱们须要在gradle的配置加上这句：

compile 'com.squareup.retrofit2:retrofit:2.0.1'复制代码

话很少说，直接上例子：


    private static OkHttpClient mOkHttpClient;
    private static Converter.Factory gsonConverterFactory = GsonConverterFactory.create();
    private static CallAdapter.Factory rxJavaCallAdapterFactory = RxJavaCallAdapterFactory.create();
     public static BaseHttpApi getObserve() {

        if (baseHttpApi == null) {
            Retrofit retrofit = new Retrofit.Builder()
                    .addConverterFactory(gsonConverterFactory)
                    .addCallAdapterFactory(rxJavaCallAdapterFactory)
                    .client(mOkHttpClient)
                    .baseUrl(BaseUrl.WEB_BASE)
                    .build();
            baseHttpApi = retrofit.create(BaseHttpApi.class);
       }
        return baseHttpApi;

    }复制代码

如上代码，能够很清晰的看出，它经过2个工厂模式建立了gson和rxjava。而且经过了链式调用将他们进行了绑定。那么怎么经过链式调用实现网络请求呢？不急，咱们喝杯茶，接着往下看。

好比，一个post请求，咱们能够这么写：

public interface BaseHttpApi{  
    @FormUrlEncoded
    @POST("seller/cash_flow_log_detail.json")
    Observable<ServiceReward> serviceReward(@Field("requestmodel") String model);
}复制代码

敲黑板了。注意，我这边是interface而不是一个class。接下来就是平常调用了，代码以下：

Network.getObserve()
                .serviceReward(new Gson().toJson(map))
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<ServiceReward>() {
                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(ServiceReward serviceReward) {
                        parseOrderDetail(serviceReward);
                    }
                });复制代码

看第二行，这就是为何刚开始为何要用工厂模式建立gson的缘由。如今咱们只要在parseOrderDetail方法中处理正常的逻辑就能够了。是否是看起来代码有点多？那么咱们能够这样：

Network.getObserve()
                .serviceReward(new Gson().toJson(map))
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(serviceReward ->{
                        parseOrderDetail(serviceReward);
                 });复制代码

一个lamada表达式，是否是感受瞬间代码少了不少，不过有人要说，我加载的时候是一个弹窗显示的，若是加载失败了我这个弹窗岂不是影藏不了？不存在的，若是真有这种状况怎么作？咱们接着看：

Network.getObserve()
                .serviceReward(new Gson().toJson(map))
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(serviceReward ->{
                        parseOrderDetail(serviceReward);
                 },throwable ->{do something when net error...});复制代码

这么处理岂不是快哉。对于lamada，刚开始可能都是各类不习惯，不过用习惯了就会发现代码各类简洁（我最近也在适应中）。

最后

关于rxjava其实对咱们来讲很难上手。或者不能这么说，应该是rxjava的东西太深了，咱们很难掌握透彻。因此我前面也说了若是你天天研究一个操做符，最少一个半月，若是你想理解原理。最少半年。换句话说，有关rxjava的知识彻底能够写一本书。但平常开发中，此文中的内容基本能够解决大部分的平常需求。固然，若是你有心的话，你能够去尝试着了解rxjava底层的实现原理。