RxJava学习笔记

作者: Mr槑 | 来源:发表于2017-12-04 11:23 被阅读50次

RxJava

RxJava是响应式程序设计的一种实现。在响应式程序设计中，当数据到达的时候，消费者做出响应。响应式编程可以将事件传递给注册了的观察者Observer。RxJava为观察者模式提供了一种通用的实现，并且提供了丰富的操作符来处理数据流。它同时支持不同线程之间切换。

什么是响应式编程

响应式编程（Reactive）是一种基于异步数据流概念的编程模式。数据流就像一条河：它可以被观测，被过滤，被操作，或者为新的消费者与另外一条流合并为一条新的流。
响应式编程的一个关键概念是事件。事件可以被等待，可以触发过程，也可以触发其它事件。事件是唯一的以合适的方式将我们的现实世界映射到我们的软件中：如果屋里太热了我们就打开一扇窗户。同样的，当我们的天气app从服务端获取到新的天气数据后，我们需要更新app上展示天气信息的UI；汽车上的车道偏移系统探测到车辆偏移了正常路线就会提醒驾驶者纠正，就是是响应事件。
今天，响应式编程最通用的一个场景是UI：我们的移动App必须做出对网络调用、用户触摸输入和系统弹框的响应。在这个世界上，软件之所以是事件驱动并响应的是因为现实生活也是如此。

观察者模式

大家可能都知道， RxJava 以观察者模式为骨架，在 2.0 中依旧如此。不过此次更新中，出现了两种观察者模式：

Observable ( 被观察者 ) / Observer ( 观察者 )
Flowable （被观察者）/ Subscriber （观察者）

在 RxJava 2.x 中，Observable 用于订阅 Observer，不再支持背压（1.x 中可以使用背压策略），而 Flowable 用于订阅 Subscriber ，是支持背压（Backpressure）的。

接口变化

RxJava 2.x 拥有了新的特性，其依赖于4个基础接口，它们分别是：

Publisher
Subscriber
Subscription
Processor

其中最核心的莫过于 Publisher 和 Subscriber。Publisher 可以发出一系列的事件，而 Subscriber 负责和处理这些事件。其中用的比较多的自然是 Publisher 的 Flowable，它支持背压。

Base Reactive Interfaces

可订阅的对象在RxJava1中只有Observable一种，之前我们经常会直接把数据源称作Observable。而在RxJava2中扩充成了4种 Observable、Flowable、Singles、Completable，因此在之后还是把他们统称为数据源为宜。

public abstract class Flowable<T> implements Publisher<T> {

public abstract class Observable<T> implements ObservableSource<T> {

public abstract class Single<T> implements SingleSource<T> {

public abstract class Completable implements CompletableSource {

Single

和Observable，Flowable一样会发送数据，不同的是订阅后只能接受到一次。普通Observable可以使用toSingle转换：Observable.just(1).toSingle()

Completable

与Single类似，只能接受到完成(onComplete)和错误(onError)。同样也可以由普通的Observable转换而来：Observable.just(1).toCompletable()

和Flowable的接口Publisher类似，Observable、Single、Completable也有类似的基类

public interface Publisher<T> {
    void subscribe(Subscriber<? super T> s);
}

interface ObservableSource<T> {
    void subscribe(Observer<? super T> observer);
}

interface SingleSource<T> {
    void subscribe(SingleObserver<? super T> observer);
}

interface CompletableSource {
    void subscribe(CompletableObserver observer);
}

Observable的创建

Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                //执行一些其他操作
                //.............
                //执行完毕，触发回调，通知观察者
                e.onNext("我来发射数据");
            }
        });

使用create( )创建Observable最基本的创建方式。可以看到，这里传入了一个 ObservableOnSubscribe对象作为参数，它的作用相当于一个计划表，当 Observable被订阅的时候，ObservableOnSubscribe的subscribe()方法会自动被调用，事件序列就会依照设定依次触发（对于上面的代码，就是观察者Observer 将会被调用一次 onNext()）。这样，由被观察者调用了观察者的回调方法，就实现了由被观察者向观察者的事件传递，即观察者模式。

Observer的创建

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

    }

    @Override
    //观察者接收到通知,进行相关操作
    public void onNext(String aLong) {
        System.out.println("我接收到数据了");
    }

    @Override
    public void onError(Throwable e) {

    }

    @Override
    public void onComplete() {

    }
};

public interface Observer<T> {
    void onSubscribe(@NonNull Disposable d);

    void onNext(@NonNull T t);
 
    void onError(@NonNull Throwable e);

    void onComplete();
}


public interface Subscriber<T> {
    public void onSubscribe(Subscription s);

    public void onNext(T t);

    public void onError(Throwable t);

    public void onComplete();
}

public interface SingleObserver<T> {

    void onSubscribe(@NonNull Disposable d);

    void onSuccess(@NonNull T t);

    void onError(@NonNull Throwable e);
}

public interface CompletableObserver {

    void onSubscribe(@NonNull Disposable d);

    void onComplete();

    void onError(@NonNull Throwable e);
}

Subscribe订阅

 observable.subscribe(observer);

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) {}

不带任何参数的subscribe() 表示下游不关心任何事件,你上游尽管发你的数据去吧, 老子可不管你发什么.
带有一个Consumer参数的方法表示下游只关心onNext事件, 其他的事件我假装没看见

Observable的其他创建方式

just()方式

Observable<String> observable = Observable.just("Hello");

使用just( )，将为你创建一个Observable并自动为你调用onNext( )发射数据。通过just( )方式直接触发onNext()，just中传递的参数将直接在Observer的onNext()方法中接收到。

fromIterable()方式

List<String> list = new ArrayList<String>();
for(int i =0;i<10;i++){
    list.add("Hello"+i);
}
Observable<String> observable = Observable.fromIterable((Iterable<String>) list);

使用fromIterable()，遍历集合，发送每个item。相当于多次回调onNext()方法，每次传入一个item。
注意：Collection接口是Iterable接口的子接口，所以所有Collection接口的实现类都可以作为Iterable对象直接传入fromIterable()方法。

简便的观察者模式

除了Observable(被观察者)的创建之外，RxJava 2.x 还提供了多个函数式接口 ，用于实现简便式的观察者模式。具体的函数式接口包括以下：

Observable.just("hello").subscribe(new Consumer<String>() {
    @Override
    public void accept(String s) throws Exception {
        System.out.println(s);
    }
});

其中Consumer中的accept()方法接收一个来自Observable的单个值。Consumer就是一个观察者。其他函数式接口可以类似应用。
注意：Observable (被观察者)只有在被Observer (观察者)订阅后才能执行其内部的相关逻辑

Cold Observable 和 Hot Observable

Hot Observable 无论有没有 Subscriber 订阅，事件始终都会发生。当 Hot Observable 有多个订阅者时，Hot Observable 与订阅者们的关系是一对多的关系，可以与多个订阅者共享信息。

然而，Cold Observable 只有 Subscriber 订阅时，才开始执行发射数据流的代码。并且 Cold Observable 和 Subscriber 只能是一对一的关系，当有多个不同的订阅者时，消息是重新完整发送的。也就是说对 Cold Observable 而言，有多个Subscriber的时候，他们各自的事件是独立的。

尽管 Cold Observable 很好，但是对于某些事件不确定何时发生以及不确定 Observable 发射的元素数量，那还得使用 Hot Observable。比如：UI交互的事件、网络环境的变化、地理位置的变化、服务器推送消息的到达等等。

Cold Observable

只有当有订阅者订阅的时候， Cold Observable 才开始执行发射数据流的代码。并且每个订阅者订阅的时候都独立的执行一遍数据流代码。 Observable.interval 就是一个 Cold Observable。每一个订阅者都会独立的收到他们的数据流。

Observable 的 just、creat、range、fromXXX 等操作符都能生成Cold Observable。

Hot Observable

Hot observable 不管有没有订阅者订阅，他们创建后就开发发射数据流。一个比较好的示例就是鼠标事件。不管系统有没有订阅者监听鼠标事件，鼠标事件一直在发生，当有订阅者订阅后，从订阅后的事件开始发送给这个订阅者，之前的事件这个订阅者是接受不到的；如果订阅者取消订阅了，鼠标事件依然继续发射。

Cold Observable 如何转换成 Hot Observable？

使用publish，生成 ConnectableObservable

使用 publish 操作符，可以让 Cold Observable 转换成 Hot Observable。它将原先的 Observable 转换成 ConnectableObservable。

使用Subject/Processor

Subject 和 Processor 的作用是相同的。Processor 是 RxJava2.x 新增的类，继承自 Flowable 支持背压控制。而 Subject 则不支持背压控制。

public abstract class Subject<T> extends Observable<T> implements Observer<T> {

当 Subject 作为 Subscriber 时，它可以订阅目标 Cold Observable 使对方开始发送事件。同时它又作为Observable 转发或者发送新的事件，让 Cold Observable 借助 Subject 转换为 Hot Observable。

注意，Subject 并不是线程安全的，如果想要其线程安全需要调用toSerialized()方法。(在RxJava1.x的时代还可以用 SerializedSubject 代替 Subject，但是在RxJava2.x以后SerializedSubject不再是一个public class）
然而，很多基于 EventBus 改造的 RxBus 并没有这么做，这样的做法是非常危险的，因为会遇到并发的情况。

Hot Observable 如何转换成 Cold Observable？

ConnectableObservable的refCount操作符

RefCount操作符把从一个可连接的 Observable 连接和断开的过程自动化了。它操作一个可连接的Observable，返回一个普通的Observable。当第一个订阅者订阅这个Observable时，RefCount连接到下层的可连接Observable。RefCount跟踪有多少个观察者订阅它，直到最后一个观察者完成才断开与下层可连接Observable的连接。

Observable的share操作符

share操作符封装了publish().refCount()调用

Flowable的产生

在原来的RxJava 1.x版本中并没有Flowable的存在，Backpressure问题是由Observable来处理的。在RxJava 2.x中对于backpressure的处理进行了改动，为此将原来的Observable拆分成了新的Observable和Flowable，同时其他相关部分也同时进行了拆分。原先的Observable已经不具备背压处理能力。

被观察者发送消息十分迅速以至于观察者不能及时的响应这些消息。

背压是指在异步场景中，被观察者发送事件速度远快于观察者的处理速度的情况下，一种告诉上游的被观察者降低发送速度的策略。简而言之，背压是流速控制的一种策略。

处理Backpressure的策略

处理Backpressure的策略仅仅是处理Subscriber接收事件的方式，并不影响Flowable发送事件的方法。即使采用了处理Backpressure的策略，Flowable原来以什么样的速度产生事件，现在还是什么样的速度不会变化，主要处理的是Subscriber接收事件的方式。

什么情况下才会产生Backpressure问题？

如果生产者和消费者在一个线程的情况下，无论生产者的生产速度有多快，每生产一个事件都会通知消费者，等待消费者消费完毕，再生产下一个事件。所以在这种情况下，根本不存在Backpressure问题。即同步情况下，Backpressure问题不存在。
如果生产者和消费者不在同一线程的情况下，如果生产者的速度大于消费者的速度，就会产生Backpressure问题。即异步情况下，Backpressure问题才会存在。

ERROR

这种方式会在产生Backpressure问题的时候直接抛出一个异常,这个异常就是著名的MissingBackpressureException。

Flowable<Integer> flowable = Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                Log.d(TAG, "emit 1");
                emitter.onNext(1);
                Log.d(TAG, "emit 2");
                emitter.onNext(2);
                Log.d(TAG, "emit 3");
                emitter.onNext(3);
                Log.d(TAG, "emit complete");
                emitter.onComplete();
            }
}, BackpressureStrategy.ERROR); //增加了一个参数

Subscriber<Integer> subscriber = new Subscriber<Integer>() {
            @Override
            public void onSubscribe(Subscription s) {
                Log.d(TAG, "onSubscribe");
                s.request(Long.MAX_VALUE);  
            }
            @Override
            public void onNext(Integer integer) {
                Log.d(TAG, "onNext: " + integer);

            }
            @Override
            public void onError(Throwable t) {
                Log.w(TAG, "onError: ", t);
            }
            @Override
            public void onComplete() {
                Log.d(TAG, "onComplete");
            }
};

flowable.subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(subscriber);

上述代码创建了一个Flowable(被观察者)和一个Subscriber(观察者)。不同的是 onSubscribe(Subscription s)中传给我们的不再是Disposable了, 而是Subscription。然而Subscription也可以用于切断观察者与被观察者之间的联系，调用Subscription.cancel()方法便可。不同的地方在于Subscription增加了一个void request(long n)方法, 这个方法有什么用呢, 在上面的代码中也有这么一句代码:

s.request(Long.MAX_VALUE);
这个方法就是用来向生产者申请可以消费的事件数量。这样我们便可以根据本身的消费能力进行消费事件。
当调用了request()方法后，生产者便发送对应数量的事件供消费者消费。
这是因为Flowable在设计的时候采用了一种新的思路也就是响应式拉取的方式,你要求多少，我便传给你多少。

注意：如果不显示调用request就表示消费能力为0。

虽然并不限制向request()方法中传入任意数字，但是如果消费者并没有这么多的消费能力，依旧会造成资源浪费，最后产生OOM。形象点就是不能打肿脸充胖子。而ERROR策略就避免了这种情况的出现。

在异步调用时，RxJava中有个缓存池，用来缓存消费者处理不了暂时缓存下来的数据，缓存池的默认大小为128，即只能缓存128个事件。无论request()中传入的数字比128大或小，缓存池中在刚开始都会存入128个事件。当然如果本身并没有这么多事件需要发送，则不会存128个事件。
在ERROR策略下，如果缓存池溢出，就会立刻抛出MissingBackpressureException异常。

Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                for (int i = 0; i < 129; i++) {
                    Log.d(TAG, "emit " + i);
                    emitter.onNext(i);
                }
            }
        }, BackpressureStrategy.ERROR).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        mSubscription = s;
                    }

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

                    @Override
                    public void onComplete() {
                    }
});

让Flowable发送129个事件，而Subscriber一个也不处理(如果不显示调用request就表示消费能力为0)，就产生了异常。
因此，ERROR即保证在异步操作中，事件累积不能超过128，超过即出现异常。消费者不能再接收事件了，但生产者并不会停止。

BUFFER

所谓BUFFER就是把RxJava中默认的只能存128个事件的缓存池换成一个大的缓存池，支持存很多很多的数据。
这样，消费者通过request()即使传入一个很大的数字，生产者也会生产事件，并将处理不了的事件缓存。
但是这种方式任然比较消耗内存，除非是我们比较了解消费者的消费能力，能够把握具体情况，不会产生OOM。
总之BUFFER要慎用。

DROP

看名字就可以了解其作用：当消费者处理不了事件，就丢弃。
消费者通过request()传入其需求n，然后生产者把n个事件传递给消费者供其消费。其他消费不掉的事件就丢掉。

mFlowable = Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
                for (int i = 0; ; i++) {

                    emitter.onNext(i);
                }
            }
        }, BackpressureStrategy.DROP);
        mSubscriber = new Subscriber<Integer>() {
            @Override
            public void onSubscribe(Subscription s) {
                mSubscription = s;
                s.request(50);
            }

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

            @Override
            public void onError(Throwable t) {
                Log.w(TAG, "onError: ", t);
            }

            @Override
            public void onComplete() {

            }
        };
    }
    
    public void start(View view){
        mFlowable.subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(mSubscriber);

    }
    
    public void consume(View view){
        mSubscription.request(50);

    }

生产者一次性传入128个事件进入缓存池。点击“开始”按钮，消费了50个。然后第一次点击“消费”按钮，又消费了50个，第二次点击“消费”按钮，再次消费50个。然而此时原来的128个缓存只剩下28个了，所以先消费掉28个，然后剩下22个是后来传入的（其实后来的是在消费了96个后传入，并一次性在缓存池中又传入了96个，具体可以看源码，这里不解释了）。

LATEST

LATEST与DROP功能基本一致。
消费者通过request()传入其需求n，然后生产者把n个事件传递给消费者供其消费。其他消费不掉的事件就丢掉。

唯一的区别就是LATEST总能使消费者能够接收到生产者产生的最后一个事件。
还是以上述例子展示，唯一的区别就是Flowable不再无限发事件，只发送1000000个。

上述例子Flowable对象的获取都是通过create()获取的，自然可以通过BackpressureStrategy.LATEST之类的方式指定处理背压的策略。如果Flowable对象不是自己创建的，可以采用onBackpressureBuffer()、onBackpressureDrop()、onBackpressureLatest()的方式指定。

Flowable.just(1).onBackpressureBuffer()
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeOn(Schedulers.io())
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {

                    }
});

背压和流量控制

Flow Control有哪些思路呢？

背压（Backpressure）
节流（Throttling）
打包处理
调用栈阻塞（Callstack blocking）

Flow Control的几种思路

背压（Backpressure）

Backpressure，也称为Reactive Pull，就是下游需要多少（具体是通过下游的request请求指定需要多少），上游就发送多少。这有点类似于TCP里的流量控制，接收方根据自己的接收窗口的情况来控制接收速率，并通过反向的ACK包来控制发送方的发送速率。

这种方案只对于所谓的cold Observable有效。cold Observable指的是那些允许降低速率的发送源，比如两台机器传一个文件，速率可大可小，即使降低到每秒几个字节，只要时间足够长，还是能够完成的。相反的例子是音视频直播，数据速率低于某个值整个功能就没法用了（这种就属于hot Observable了）。

节流（Throttling）

节流（Throttling），说白了就是丢弃。消费不过来，就处理其中一部分，剩下的丢弃。还是举音视频直播的例子，在下游处理不过来的时候，就需要丢弃数据包。

而至于处理哪些和丢弃哪些数据，就有不同的策略。主要有三种策略：

sample (也叫throttleLast)
throttleFirst
debounce (也叫throttleWithTimeout)

从细的方面分别解释一下。

sample，采样。类比一下音频采样，8kHz的音频就是每125微秒采一个值。sample可以配置成，比如每100毫秒采样一个值，但100毫秒内上游可能过来很多值，选哪个值呢，就是选最后那个值。所以它也叫throttleLast。

打包处理

打包就是把上游来的小包裹打成大包裹，分发到下游。这样下游需要处理的包裹的个数就减少了。RxJava中提供了两类这样的机制：buffer和window。

调用栈阻塞（Callstack blocking）

这是一种特殊情况，阻塞住整个调用栈（Callstack blocking）。之所以说这是一种特殊情况，是因为这种方式只适用于整个调用链都在一个线程上同步执行的情况，这要求中间的各个operator都不能启动新的线程。在平常使用中这种应该是比较少见的，因为我们经常使用subscribeOn或observeOn来切换执行线程，而且有些复杂的operator本身也会在内部启动新的线程来处理。另外，如果真的出现了完全同步的调用链，前面的另外三种Flow Control思路仍然可能是适用的，只不过这种阻塞的方式更简单，不需要额外的支持。

这里举个例子把调用栈阻塞和前面的Backpressure比较一下。“调用栈阻塞”相当于很多车行驶在盘山公路上，而公路只有一条车道。那么排在最前面的第一辆车就挡住了整条路，后面的车也只能排在后面。而“Backpressure”相当于银行办业务时的窗口叫号，窗口主动叫某个号过去（相当于请求），那个人才过去办理。

如何让Observable支持Backpressure？

在RxJava 1.x中，有些Observable是支持Backpressure的，而有些不支持。但不支持Backpressure的Observable可以通过一些operator来转化成支持Backpressure的Observable。这些operator包括：

onBackpressureBuffer
onBackpressureDrop
onBackpressureLatest
onBackpressureBlock（已过期）

它们转化成的Observable分别具有不同的Backpressure策略。

而在RxJava 2.x中，Observable不再支持Backpressure，而是改用Flowable来专门支持Backpressure。上面提到的四种operator的前三种分别对应Flowable的三种Backpressure策略：

BackpressureStrategy.BUFFER
BackpressureStrategy.DROP
BackpressureStrategy.LATEST