rxjava理论(二)

2023-05-16

doOnSubscribe的监听

在上一节我们介绍过subscribeOn是控制上游的observable在哪个线程执行，关于怎么控制上游的observable可以看我上篇文章RxJava面经一，拿去，不谢!，那如果多次执行subscribeOn的时候，Observable接收的是第一次的subscribeOn指定的线程，因为每次设置都会被上一层subscribeOn设置的线程所覆盖了，这里的覆盖是对于最上游的Observable而言的，中间生成的Observable其实是有线程切换的，我们可以通过doOnSubscribe来监听每一次subscribeOn线程的切换，我们还是拿例子来说：

在上一篇文章我们已经说过，订阅是从最下游的observer到上面一层一层的observable，所以我们最下游的observable开始发生订阅，也就是①处通过subscribeOn生成的ObservableSubscribeOn观察者开始订阅，它会在订阅方法中，给它的上游的observable添加订阅，也就是②号处通过doOnSubscribe生成的ObservableDoOnLifecycle观察者开始订阅，然后在它的订阅里面给③号订阅，③号给④号添加订阅，最后到最上游的observable发生订阅，也就是最上游的ObservableOnSubscribe的subscribe方法被调用。这就是从下到上依次订阅的顺序，下面以一张图说明订阅顺序：

那什么时候doOnSubscribe的内部类Consumer的accept方法什么调用呢？

我们直接看上面的图，它是在上一个Observable，也就是doOnSubscribe生成的ObservableDoOnLifecycle里面的装饰observer(DisposableLambdaObserver)监听到订阅的时候调用的。而在该例子中②号、④号通过doOnSubscribe生成的observable的上游observable是subscribeOn生成的，而subscribeOn最终是生成了ObservableSubscribeOn的observable，在它的订阅里面是直接给下游的observer添加订阅监听了：

所以由上面可知②号处的doOnSubscribe打印是在③号上游的subscribeOn发生订阅的时候，所以它最先打印出结果，再一次是④号打印出结果，最后是最上游的observable的订阅打印。那每一处的doOnSubscribe中accept接收到的线程是怎么回事呢，这个我先说结论，是跟它下面的subscribeOn指定的线程保持一致。所以②号处打印是①号处指定的线程，④号是③号处指定的线程打印，后面我们分析doOnSubscribe时候说。

上一节我们知道subscribeOn是指定它上游的observable订阅发生的线程，而doOnSubscribe操作符最终也是生成了一个ObservableDoOnLifecycle的observable，所以可以这么说ObservableDoOnLifecycle的订阅发生的线程是由紧跟它后面的subscribeOn指定的线程所决定的。

而在ObservableDoOnLifecycle的订阅方法中，它是直接订阅了上游的observable，在上面示例中也就是第二个observable，ObservableDoOnLifecycle的subscribeActual方法如下：

上一篇介绍了在每一个Observable的订阅方法中，会先创建装饰的observer，并且把下游的observer传到创建的装饰的observer中，接着会给下游的observer添加订阅的回调，接着会给上游的observable添加订阅，而在此处的ObservableDoOnLifecycle订阅方法中先是创建了DisposableLambdaObserver的装饰observer，接着给上游的observable添加订阅。那给下游的obserer添加订阅的监听呢，这就放在了DisposableLambdaObserver的装饰observer的onSubscribe中了。

由于上面我们通过doOnSubscribe生成最下游的observable(ObservableDoOnLifecycle)的订阅线程是io线程，所以它的上游observable也是io线程，我们还没分析完doOnSubscribe传进去的Consumer的accept方法发生的线程，这个需要我们看下上面分析的ObservableDoOnLifecycle订阅中创建的装饰DisposableLambdaObserver：

上面一上来就是给传进来的Consumer执行了accept的回调，紧接着给下游的observer添加订阅的监听，方便下游的observer能收到订阅的回调啊，是不是这么回事呢？

那此处装饰的observer(DisposableLambdaObserver)订阅监听是由谁发起的呢，肯定是上游的observable开始订阅的时候发起的下游observer订阅监听啊，而上面我们分析了此处的上游observable订阅线程是由紧挨doOnSubscribe的subscribeOn决定的，所以此处不难看出最终doOnSubscribe中的consumer监听的是subscribeOn指定上游的observable订阅过程中发生的线程，大家可以多理解这句话!!!

下面画张图补补脑：

doOnSubscribe小节

上面分析在没有结合源码的情况下，不好分析，整体就是subscribeOn会指定它上游的observable线程，而它上游又正好是doOnSubscribe生成的observable，该observable是ObservableDoOnLifecycle，在它的订阅里面又直接去订阅了它上游的observable，所以此时doOnSubscribe的上游observable线程也是doOnSubscribe它下面的subscribeOn指定的，而doOnSubscribe的上游observable是subscribeOn生成的，它是ObservableSubscribeOn，在它的订阅里面是直接监听了下游的observer订阅回调，也就是doOnSubscribe生成的ObservableDoOnLifecycle订阅中生成的装饰DisposableLambdaObserver，它的订阅监听会调用doOnSubscribe传进来的Consumer的accept方法。

所以这就是多次调用subscribeOn可以通过doOnSubscribe来做线程切换的监听。

doOnNext监听observeOn线程的切换，map的apply方法的线程由谁控制？

首先我们还是通过例子来回答上面你的问题，先来看doOnNext的使用：

关于doOnNext其实很好理解，发射数据因为是从上游的observable到下游的observable，而observeOn是指定下游的observer发射数据的线程，这个我在上一篇讲过，而doOnNext实际生成的是一个ObservableDoOnEach的observable，在该订阅方法中，会生成装饰的observer，也就是DoOnEachObserver，所以observeOn实际是控制了DoOnEachObserver发射数据的线程，而在它发射onNext数据的时候，会调用onNext传进来的Consumer的accept方法：

observable里面还有onComplete、onError的监听，他们最终都是生成了ObservableDoOnEach的observable

doOnNext小节

doOnNext中是通过传进去的Consumer作为上游发射数据过来的监听，在上游observable发射数据的时候，会执行doOnNext的Consumer的accept方法，所以在上面多次通过observeOn指定线程的时候，可以通过doOnNext拿到切换线程的。

所以这就是多次调用observeOn可以通过onNext来做线程切换的监听。

关于map的apply方法的线程由谁来控制，我们这块直接看map的observable，它是一个ObservableMap：

不难看出，map操作符生成的ObservableMap，在它的订阅方法中，生成装饰的MapObserver，接着给上游的observable添加订阅，在MapObserver接收到上游的observable发射onNext数据的时候会调用map传进来的function的apply方法，因此apply的方法是跟上游的observable发射数据的线程有关，我们来看下面例子：我们知道subscribeOn是指定上游的observable的订阅线程，我们在上篇文章讲过多个subscribeOn指定线程，只有第一次有效，这是针对最上游的observable而言的，所以最上游的observable发射数据端的线程紧跟它后面指定的io线程保持一致，所以会有如下打印：

而在每一个subscribeOn发射数据的时候不会改变线程，所以map的线程会保持最上游的observable的线程，也就是io线程，所以打印会有如下：

既然subscribeOn不会改变发射数据的线程，导致多次subscribeOn不会改变map的线程，所以只会跟最上游的observable发射数据的线程保持一致，那我们如果中间插入observeOn呢，下面来看下这个例子：

由于observeOn会改变给下游发送数据的时候线程,也就是改变下游observer接收数据的线程,也即onNext、onComplete、onError方法，所以observeOn指定的线程会一直传到了下游MapObserver的onNext方法中，所以最终map中的function的apply方法是main线程，打印结果如下：

其他情况大家可以尝试，比如多次指定observeOn线程，看map最终的线程如何

map小节

map将传进去的function作为上游发射数据过来的监听，在上游observable发射数据的时候，会执行function的apply方法来达到转换数据的目的，所以map中function的apply方法是跟上游的observable发射数据的线程有关。

如果不指定observer的线程，也就是指设置subscribeOn，而不设置observeOn，那observer的线程是什么样的？

我感觉理解了整个订阅的过程，其实理解这个问题一点都不难，既然subscribeOn是指定上游的observable的线程，那么最终的上游observable发射数据时候的线程也会被紧挨着它的subscribeOn指定的线程有关啊，并且不设置observeOn指定下游的observer的线程，那么observer的线程是不是跟最上游observable发射数据的线程保持一致啊。

背压是什么，以及Flowable怎么能控制背压？

它是指由于上游的observable发射数据太快，下游observer接收数据跟不上来导致的一种现象。可以形象理解为水坝在存储水的时候为了保持水的平衡，给下游的水库放水，同时会接收上游的水流，如果上游的水流很大，那么水坝中的水位激增，而水坝给下游放水的能力有限，所以就会导致水坝中的水漫过水坝。

RxJava1.0背压

注：说到RxJava背压还得从RxJava1.0开始说起，这里分析的RxJava1.0版本源码是在1.3.8版本分析

在RxJava的1.0版本中Observable是支持背压的，只不过它是以异常的形式展示给用户，下面我们拿上游不断地发送数据的例子来模拟下：

Observable.unsafeCreate(new Observable.OnSubscribe<Integer>() {
    @Override
    public void call(Subscriber<? super Integer> subscriber) {
        int i = 0;
        while (true) {
            subscriber.onNext(i);
            i++;
        }
    }
}).subscribe(new Observer<Integer>() {
    @Override
    public void onCompleted() {
    }
    @Override
    public void onError(Throwable e) {
        Log.d(TAG, "onError:" + e.getMessage());
        e.printStackTrace();
    }
    @Override
    public void onNext(Integer integer) {
        Log.d(TAG, "onNext:" + integer);
    }
});
复制代码

在我们日志中，没看到下游处理数据处理不过来的问题，也没出现异常信息啊，这不挺好的，没出现背压的情况，看官，别急啊，背压是发生在多线程中的问题，因为在单线程中，发送数据和接收数据都是在一个线程中，所以每次发送数据前得等到下游处理完数据才发送数据。

好吧，说完这么多，我们还是加上多线程吧，我们再来试试：

Observable.unsafeCreate(new Observable.OnSubscribe<Integer>() {
    @Override
    public void call(Subscriber<? super Integer> subscriber) {
        for (int i = 0; ; i++) {
            subscriber.onNext(i);
        }
    }
}).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<Integer>() {
    @Override
    public void onCompleted() {
    }
    @Override
    public void onError(Throwable e) {
        Log.d(TAG, "onError:" + e.getMessage());
        e.printStackTrace();
    }
    @Override
    public void onNext(Integer integer) {
        Log.d(TAG, "onNext:" + integer);
    }
});
复制代码

我们在上游发送数据的时候，做了个死循环，然后指定了上游是io线程，下游是main线程，结果还没接收到数据直接报异常了：

报的是MissingBackpressureException异常信息，其实在RxJava1.0中这就是支持背压的策略，直接通过异常的信息反馈给用户。在RxJava1.0中支持最大的发射数据是16个，也就是说发射大于或等于17个的时候就会出现异常，下面通过代码验证下：

数据能正常接收，如果我们把数据调整到17呢，是不是会发生异常呢：

看吧，不用我说啥吧，看来底层是限制了上游发送数据的个数，其实这个是RxJava1.0背压策略的一种机制，通过数量来控制，我们可以在这里找到定义的数量大小：

RxJava背压对android平台做了发送数据的限制，如果大于16个则直接抛MissingBackpressureException异常，底层通过上游Observable发送的数据放到队列中，而这里的16则是定义队列的容量，每次在往队列中放数据的时候会先获取下一个要放数据的索引，如果发现索引位置的数据不为空，则认为队列已经满了，那么满了就直接返回onError的信息。

比如我们在发送第17个数据的时候，在获取索引的时候是通过与对接容量16 进行相与得到索引，相与之后得到一个小于16的索引，发现相与之后得到的索引上还有数据，则发送第17个数据放进队列的时候失败，所以直接抛出onError的信息。核心源码在这里：

这个过程还是蛮清晰的，OperatorObserveOn是observeOn方法传给OnSubscribeLift的Operator对象，在OperatorObserveOn中会先初始化队列，并且队列的容量是16，接着在onNext接收上游发送过来的数据的时候，会判断队列的offer是否成功，如果不成功，则直接抛onError的错误，那什么时候offer会失败呢，得看当前发送过来的数据是否超过了队列的容量，如果超过则offer失败。

所以这就是RxJava1.0中背压策略，通过设置上游发送过来的数据的接收队列容量来达到背压。

RxJava2.0背压

注：RxJava运行版本是2.2.20

在RxJava2.0中不再在Observable支持背压，而是通过Flowable来代替了，也就是说Observable中不再通过异常的形式告诉用户了，也就是不抛MissingBackpressureException异常了，下面来看看RxJava2.0正常发送数据的问题：

Observable.create(new ObservableOnSubscribe<Integer>() {
    @Override
    public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception {
        for (int i = 0; ; i++) {
            emitter.onNext(i);
        }
    }
}).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<Integer>() {
    @Override
    public void onSubscribe(@NonNull Disposable d) {
    }
    @Override
    public void onNext(@NonNull Integer integer) {
        Log.d(TAG, "onNext：" + integer);
    }
    @Override
    public void onError(@NonNull Throwable e) {
    }
    @Override
    public void onComplete() {
    }
});
复制代码

虽然说不抛MissingBackpressureException异常了，但是内存占用很糟糕啊，所以针对RxJava2.0的问题，我们有没有处理办法呢，大家会说RxJava2.0不是已经支持Flowable了吗，直接使用它啊，如果让我们自己来处理啊该怎么办呢，首先我们分析背压产生的原因是什么：

上游发送的事件太快，下游处理不过来
上游发送的事件太多，下游处理不过来

首先针对第一种我们可以让上游发送速度慢点，怎么慢点呢，让io线程每次发送的时候停留一会：

Observable.create(new ObservableOnSubscribe<Integer>() {
    @Override
    public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception {
        for (int i = 0; ; i++) {
            emitter.onNext(i);
            Thread.sleep(1000);
        }
    }
}).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<Integer>() {
    @Override
    public void onSubscribe(@NonNull Disposable d) {
    }
    @Override
    public void onNext(@NonNull Integer integer) {
        Log.d(TAG, "onNext：" + integer);
    }
    @Override
    public void onError(@NonNull Throwable e) {
    }
    @Override
    public void onComplete() {
    }
});
复制代码

图片来自于关于 RxJava 背压

针对第二种的话，我们可以让下游的observer少接收点数据：

Observable.create(new ObservableOnSubscribe<Integer>() {
    @Override
    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
        for (int i = 0; ; i++) {   //无限循环发事件
            emitter.onNext(i);

        }
    }
}).filter(new Predicate<Integer>() {
    @Override
    public boolean test(Integer integer) throws Exception {
        return integer % 100 == 0;

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

                Log.d(TAG, "" + integer);
            }
        });
复制代码

图片来自于关于 RxJava 背压

知道了背压产生的原因后，我们再回头看RxJava本身用Flowable来支持背压策略，而且它的策略比较丰富，下面来一一介绍，我们先从Flowable.create入手：

FlowableCreate支持两个参数，第一个不用说了吧，是上游发射数据的FlowableOnSubscribe，第二个参数是背压的几种策略：

public enum BackpressureStrategy {
    
    MISSING,//如果流的速度无法保持同步，可能会抛出MissingBackpressureException或IllegalStateException。
   
    ERROR,//会在下游跟不上速度时抛出MissingBackpressureException。
   
    BUFFER,//上游不断的发出onNext请求，直到下游处理完，也就是和Observable一样了，缓存池无限大，最后直到程序崩溃
   
    DROP,//会在下游跟不上速度时把onNext的值丢弃。
   
    LATEST//会一直保留最新的onNext的值，直到被下游消费掉。
}
复制代码

在知道了Flowable这多的策略时候，我们先来看看Flowable在单线程下是什么样的：

在策略为error情况下，并且没切换线程的时候，直接报io.reactivex.exceptions.MissingBackpressureException: create: could not emit value due to lack of requests错误，该错误是告诉你没对下游的observer设置request的方法，这个是由于在单线程情况下，没默认给observer设置处理数据的能力，也即是个数，所以上游不知道下游的处理能力，直接抛error错误。

下面怎么设置下游处理能力呢：

直接下游接收到订阅方法中添加subscription.request(Long.MAX_VALUE)，官方建议我们使用Long.MAX_VALUE，表示告诉上游，下游的处理能力最大，你尽管发送数据给我吧。

那如果上游发送的数据个数大于下游设置的个数呢：

可以看到在发送第4个数据的时候，直接抛异常了，因为下游设置的处理能力是3个，每次在发送完一次的时候，会“削弱”，下游的处理数据的能力，等到发送第四个数据的时候，发现下游已经不能再处理了，直接抛异常。

上面都是在error策略，单线程下的结果，那如果在多线程中结果会是咋样呢，还得从几种策略情况来看：

MISSING

在多线程下必须设置下游的处理能力，因为在observeOn给下游发送数据的时候需要知道下游能处理数据的个数。上面我们演示的是上游发送128个数据，结果没有像MISSING策略所说的抛出抛出MissingBackpressureException或IllegalStateException异常信息，这是因为Flowable默认认为128个数据是上游发送最多的数据，我们可以通过这里找到定义的数量：

底层其实是跟RxJava1.0点的做法是一样的，也是把这个容量作为队列的大小，只不过RxJava1.0的容量是16个，所以再发送第129个数据的时候，会出现队列放满的问题，一旦放满，再往里面放数据就会出现RxJava中定义的各种策略情况，下面我们把发送数据改为129个，看看MISSING会出现什么情况：

看到了吧，直接抛io.reactivex.exceptions.MissingBackpressureException: Queue is full?!异常信息。MISSING内部的发射器里面其实啥都没做，它发送异常是在FlowableObserveOn.ObserveOnSubscriber内部类的onNext时候，发现128个大小的队列满了后，给下游的observer发送onError的信息。

ERROR

测试代码我就不贴了，直接把上面上面的BackpressureStrategy.MISSING改为BackpressureStrategy.ERROR

其实和MISSING抛出的异常是一样的，只不过异常的message不一样而已。内部通过发送数据的时候定义一个AtomicLong的计数器，每次在给下游发送完一个数据后，会将该计数器减一，等到减到0的时候，直接在上游给下游发送onError的信息。

BUFFER

这个其实跟RxJava2.0的Observable使用没什么区别，输出的容量没有大小限制，也不会像RxJava1.0一样抛异常，请谨慎使用。

DROP

drop是在第一次拿到128个数据后，第二次从队列中拿数据的时候，中间跟不上速度的数据抛弃了，等到下游处理完先前的128个数据的时候，才能接收后面96个数据，至于这里为什么是96个数字，是因为后面的容量减为this.limit = prefetch - (prefetch >> 2);这个大小了，prefetch是128，大家自己算吧：

这里从5118....96个数...从5118算起第96个数

DROP是当给下游发送数据的时候，自己有个限流的策略，通过AtomicLong装载的128大小的计数器，每次发送完一个数据后，会将该计数器减一，那如果发送到了128个数据的时候，由于计数器减到0了，等到下游处理完这128个数据的时候，才会把计数器给调整到96，所以中间会出现丢数据的情况，等到下游处理完先前128个数据的时候，上游再次发数据的时候已经不会从129个数开始了，而且发的这96个数是随机的，因为下游处理前面128个数的时间是不确定的。

LATEST

我们先来看latest是什么样的效果，为了要区分和drop的效果，我们将发射数据改为2000的数据量：

Flowable.create(new FlowableOnSubscribe<Integer>() {
    @Override
    public void subscribe(FlowableEmitter<Integer> emitter) throws Exception {
        for (int i = 0; i < 2000; i++) {
            emitter.onNext(i);
        }
    }
}, BackpressureStrategy.LATEST).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Subscriber<Integer>() {
            @Override
            public void onSubscribe(Subscription s) {
                Log.d(TAG, "onSubscribe");
                s.request(Long.MAX_VALUE);
                subscription = 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() {
                Log.d(TAG, "onComplete");
            }
        });
复制代码

这里得到的结果会是这样的：

它是这么接收数据的：0-127....96个数...最后一个数，latest是先发送0到127的数据，然后中间发送96个数字，这中间会有丢失数据的，而最后会把最后一个数据发送给下游。

latest没有像drop那样通过计数器的形式限制发送数据的速度，而是在发送数据的时候定义了AtomicReference原子类，把数据放在里面，所以它是每次只能存储一个数据，等处理完前面的128个数据的时候，会将AtomicLong定义的128个数量减到0，所以在下游接收完前面的128个数的时候，上游才能给下游发送后面96个数，等到最后的时候会由于缓存的是最后一个数，所以只能发送给下游的只能是最后一个数。

好了，关于背压的几种策略就那么几种，其实我们总结下来：

MISSION：上游没有限流，在下游里面发现队列满了，给下游发送onError的信息，该信息是io.reactivex.exceptions.MissingBackpressureException: Queue is full?!。
ERROR:在上游通过限流的形式给下游发送数据，在发现数据量到了128个的时候，会给下游发送onError的信息，该信息是create: could not emit value due to lack of requests。
DROP:也是在上游通过限流的形式给下游发送数据，在发现数据量到了128的时候，会等下游处理完这128个数据，等到处理完了，继续处理梳理，所以在等的过程中会有数据丢失的问题。
LATEST:虽然和DROP都是同样的丢数据，但是它两的做法是不一样的，LATEST通过只能放一个数据的容易来给下游发送数据，最开始丢数据基本是一样的，但是LATEST会保留最后一条数据，是因为最后处理数据的时候，容器里面还有一条数据。
BUFFER:这个跟RxJava2.0普通发送数据是一样的，它不支持背压，上游发送多少数据，下游会接收多少数据，直到发生OOM。

总结

介绍了doOnSubscribe监听每次subscribeOn的线程切换
doOnNext监听每一次observeOn线程的切换，以及map的apply方法的线程是由上游发送数据的observable决定的。
如果不指定observer的线程，也就是指设置subscribeOn，而不设置observeOn，那observer的线程跟上游的observable一起走的。
介绍了RxJava1.0和RxJava2.0背压的使用，以及他们的区别。

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

RXJava