2023年7月25日发(作者：)

RxJava⾯试⼆，拿去，不谢!话题通过上⼀篇我们分析了RxJava的订阅，以及RxJava常见的⾯试问题，还没有看上⼀篇点这⾥RxJava⾯经⼀，拿去，不谢!。在上⼀篇我们分析过多次调⽤subscribeOn只有第⼀次调⽤的时候有效，原因是因为最开始调⽤的subscribeOn返回的observable会把后⾯执⾏的subscribeOn返回的observable给覆盖了，因此我们感官的是只有第⼀次的subscribeOn能⽣效。实际中间每⼀个Observable⽣成的时候还是会有指定的线程的，只是在最上游的observable只接收第⼀次的subscribeOn指定的线程，那么我们可以通过doOnSubscribe监测中间的observable确实有⾃⼰的线程，这是我们这节探讨的话题。前⼀节分析了通过observeOn指定了紧跟其后的observer的线程,如果我们多次调⽤observeOn，其实是最后⼀次observeOn才有效指定observer的线程，那我们可以通过doOnNext来监听每⼀次的observeOn线程的切换，这是我们这节探讨的话题。如果我们没指定observer的线程，只指定了observable的线程，则observer的线程则会跟observable的线程⼀起⾛的，也就是我们只设置了subscribeOn，⽽没有设置observeOn的情况。这是我们这节探讨的话题。上⼀节我们简单的提过背压，那么背压是什么呢，以及Flowable怎么能控制背压也是我们这节讨论的话题。这节探讨的话题doOnSubscribe是怎么做到监听中间的observable的线程？doOnNext是怎么做到监听每⼀次observeOn线程的切换，以及map的apply⽅法的线程有谁控制？如果不指定observer的线程，也就是指设置subscribeOn，⽽不设置observeOn，那observer的线程是什么样的？背压是什么，以及Flowable怎么能控制背压？doOnSubscribe的监听在上⼀节我们介绍过subscribeOn是控制上游的observable在哪个线程执⾏，关于怎么控制上游的observable可以看我上篇⽂章RxJava⾯经⼀，拿去，不谢!，那如果多次执⾏subscribeOn的时候，Observable接收的是第⼀次的subscribeOn指定的线程，因为每次设置都会被上⼀层subscribeOn设置的线程所覆盖了，这⾥的覆盖是对于最上游的Observable⽽⾔的，中间⽣成的Observable其实是有线程切换的，我们可以通过doOnSubscribe来监听每⼀次subscribeOn线程的切换，我们还是拿例⼦来说：image在上⼀篇⽂章我们已经说过，订阅是从最下游的observer到上⾯⼀层⼀层的observable，所以我们最下游的observable开始发⽣订阅，也就是①处通过subscribeOn⽣成的ObservableSubscribeOn观察者开始订阅，它会在订阅⽅法中，给它的上游的observable添加订阅，也就是②号处通过doOnSubscribe⽣成的ObservableDoOnLifecycle观察者开始订阅，然后在它的订阅⾥⾯给③号订阅，③号给④号添加订阅，最后到最上游的observable发⽣订阅，也就是最上游的ObservableOnSubscribe的subscribe⽅法被调⽤。这就是从下到上依次订阅的顺序，下⾯以⼀张图说明订阅顺序：image那什么时候doOnSubscribe的内部类Consumer的accept⽅法什么调⽤呢？我们直接看上⾯的图，它是在上⼀个Observable，也就是doOnSubscribe⽣成的ObservableDoOnLifecycle⾥⾯的装饰observer(DisposableLambdaObserver)监听到订阅的时候调⽤的。⽽在该例⼦中②号、④号通过doOnSubscribe⽣成的observable的上游observable是subscribeOn⽣成的，⽽subscribeOn最终是⽣成了ObservableSubscribeOn的observable，在它的订阅⾥⾯是直接给下游的observer添加订阅监听了：image所以由上⾯可知②号处的doOnSubscribe打印是在③号上游的subscribeOn发⽣订阅的时候，所以它最先打印出结果，再⼀次是④号打印出结果，最后是最上游的observable的订阅打印。那每⼀处的doOnSubscribe中accept接收到的线程是怎么回事呢，这个我先说结论，是跟它下⾯的subscribeOn指定的线程保持⼀致。所以②号处打印是①号处指定的线程，④号是③号处指定的线程打印，后⾯我们分析doOnSubscribe时候说。上⼀节我们知道subscribeOn是指定它上游的observable订阅发⽣的线程，⽽doOnSubscribe操作符最终也是⽣成了⼀个ObservableDoOnLifecycle的observable，所以可以这么说ObservableDoOnLifecycle的订阅发⽣的线程是由紧跟它后⾯的subscribeOn指定的线程所决定的。⽽在ObservableDoOnLifecycle的订阅⽅法中，它是直接订阅了上游的observable，在上⾯⽰例中也就是第⼆个observable，ObservableDoOnLifecycle的subscribeActual⽅法如下：[图⽚上传失败...(image-5c5f67-16)]上⼀篇介绍了在每⼀个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：image上⾯⼀上来就是给传进来的Consumer执⾏了accept的回调，紧接着给下游的observer添加订阅的监听，⽅便下游的observer能收到订阅的回调啊，是不是这么回事呢？那此处装饰的observer(DisposableLambdaObserver)订阅监听是由谁发起的呢，肯定是上游的observable开始订阅的时候发起的下游observer订阅监听啊，⽽上⾯我们分析了此处的上游observable订阅线程是由紧挨doOnSubscribe的subscribeOn决定的，所以此处不难看出最终doOnSubscribe中的consumer监听的是subscribeOn指定上游的observable订阅过程中发⽣的线程，⼤家可以多理解这句话下⾯画张图补补脑：imagedoOnSubscribe⼩节上⾯分析在没有结合源码的情况下，不好分析，整体就是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的使⽤：image关于doOnNext其实很好理解，发射数据因为是从上游的observable到下游的observable，⽽observeOn是指定下游的observer发射数据的线程，这个我在上⼀篇讲过，⽽doOnNext实际⽣成的是⼀个ObservableDoOnEach的observable，在该订阅⽅法中，会⽣成装饰的observer，也就是DoOnEachObserver，所以observeOn实际是控制了DoOnEachObserver发射数据的线程，⽽在它发射onNext数据的时候，会调⽤onNext传进来的Consumer的accept⽅法：imageobservable⾥⾯还有onComplete、onError的监听，他们最终都是⽣成了ObservableDoOnEach的observableimagedoOnNext⼩节doOnNext中是通过传进去的Consumer作为上游发射数据过来的监听，在上游observable发射数据的时候，会执⾏doOnNext的Consumer的accept⽅法，所以在上⾯多次通过observeOn指定线程的时候，可以通过doOnNext拿到切换线程的。所以这就是多次调⽤observeOn可以通过onNext来做线程切换的监听。关于map的apply⽅法的线程由谁来控制，我们这块直接看map的observable，它是⼀个ObservableMap：image不难看出，map操作符⽣成的ObservableMap，在它的订阅⽅法中，⽣成装饰的MapObserver，接着给上游的observable添加订阅，在MapObserver接收到上游的observable发射onNext数据的时候会调⽤map传进来的function的apply⽅法，因此apply的⽅法是跟上游的observable发射数据的线程有关，我们来看下⾯例⼦：image我们知道subscribeOn是指定上游的observable的订阅线程，我们在上篇⽂章讲过多个subscribeOn指定线程，只有第⼀次有效，这是针对最上游的observable⽽⾔的，所以最上游的observable发射数据端的线程紧跟它后⾯指定的io线程保持⼀致，所以会有如下打印：image⽽在每⼀个subscribeOn发射数据的时候不会改变线程，所以map的线程会保持最上游的observable的线程，也就是io线程，所以打印会有如下：image既然subscribeOn不会改变发射数据的线程，导致多次subscribeOn不会改变map的线程，所以只会跟最上游的observable发射数据的线程保持⼀致，那我们如果中间插⼊observeOn呢，下⾯来看下这个例⼦：image由于observeOn会改变给下游发送数据的时候线程,也就是改变下游observer接收数据的线程,也即onNext、onComplete、onError⽅法，所以observeOn指定的线程会⼀直传到了下游MapObserver的onNext⽅法中，所以最终map中的function的apply⽅法是main线程，打印结果如下：image其他情况⼤家可以尝试，⽐如多次指定observeOn线程，看map最终的线程如何mmaap⼩节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是⽀持背压的，只不过它是以异常的形式展⽰给⽤户，下⾯我们拿上游不断地发送数据的例⼦来模拟下：Create(new cribe() { @Override public void call(Subscriber subscriber) { int i = 0; while (true) { (i); i++; } }}).subscribe(new Observer() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { Log.d(TAG, "onError:" + sage()); tackTrace(); } @Override public void onNext(Integer integer) { Log.d(TAG, "onNext:" + integer); }});在我们⽇志中，没看到下游处理数据处理不过来的问题，也没出现异常信息啊，这不挺好的，没出现背压的情况，看官，别急啊，背压是发⽣在多线程中的问题，因为在单线程中，发送数据和接收数据都是在⼀个线程中，所以每次发送数据前得等到下游处理完数据才发送数据。好吧，说完这么多，我们还是加上多线程吧，我们再来试试：Create(new cribe() { @Override public void call(Subscriber subscriber) { for (int i = 0; ; i++) { (i); } }}).subscribeOn(()).observeOn(read()).subscribe(new Observer() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { Log.d(TAG, "onError:" + sage()); tackTrace(); } @Override public void onNext(Integer integer) { Log.d(TAG, "onNext:" + integer); }});我们在上游发送数据的时候，做了个死循环，然后指定了上游是io线程，下游是main线程，结果还没接收到数据直接报异常了：image报的是MissingBackpressureException异常信息，其实在RxJava1.0中这就是⽀持背压的策略，直接通过异常的信息反馈给⽤户。在RxJava1.0中⽀持最⼤的发射数据是16个，也就是说发射⼤于或等于17个的时候就会出现异常，下⾯通过代码验证下：image数据能正常接收，如果我们把数据调整到17呢，是不是会发⽣异常呢：image看吧，不⽤我说啥吧，看来底层是限制了上游发送数据的个数，其实这个是RxJava1.0背压策略的⼀种机制，通过数量来控制，我们可以在这⾥找到定义的数量⼤⼩：imageRxJava背压对android平台做了发送数据的限制，如果⼤于16个则直接抛MissingBackpressureException异常，底层通过上游Observable发送的数据放到队列中，⽽这⾥的16则是定义队列的容量，每次在往队列中放数据的时候会先获取下⼀个要放数据的索引，如果发现索引位置的数据不为空，则认为队列已经满了，那么满了就直接返回onError的信息。⽐如我们在发送第17个数据的时候，在获取索引的时候是通过与对接容量16 进⾏相与得到索引，相与之后得到⼀个⼩于16的索引，发现相与之后得到的索引上还有数据，则发送第17个数据放进队列的时候失败，所以直接抛出onError的信息。核⼼源码在这⾥：imageimageimage这个过程还是蛮清晰的，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正常发送数据的问题：(new ObservableOnSubscribe() { @Override public void subscribe(@NonNull ObservableEmitter emitter) throws Exception { for (int i = 0; ; i++) { (i); } }}).subscribeOn(()).observeOn(read()).subscribe(new Observer() { @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线程每次发送的时候停留⼀会：(new ObservableOnSubscribe() { @Override public void subscribe(@NonNull ObservableEmitter emitter) throws Exception { for (int i = 0; ; i++) { (i); (1000); } }}).subscribeOn(()).observeOn(read()).subscribe(new Observer() { @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() { }});[图⽚上传失败...(image-c600a8-16)]图⽚来⾃于关于 RxJava 背压针对第⼆种的话，我们可以让下游的observer少接收点数据：(new ObservableOnSubscribe() { @Override public void subscribe(ObservableEmitter emitter) throws Exception { for (int i = 0; ; i++) { //⽆限循环发事件 (i); } }}).filter(new Predicate() { @Override public boolean test(Integer integer) throws Exception { return integer % 100 == 0; }}) .subscribeOn(()) .observeOn(read()) .subscribe(new Consumer() { @Override public void accept(Integer integer) throws Exception { Log.d(TAG, "" + integer); } });[图⽚上传失败...(image-8f1812-16)]图⽚来⾃于关于 RxJava 背压知道了背压产⽣的原因后，我们再回头看RxJava本⾝⽤Flowable来⽀持背压策略，⽽且它的策略⽐较丰富，下⾯来⼀⼀介绍，我们先从⼊⼿：imageFlowableCreate⽀持两个参数，第⼀个不⽤说了吧，是上游发射数据的FlowableOnSubscribe，第⼆个参数是背压的⼏种策略：public enum BackpressureStrategy {

MISSING,//如果流的速度⽆法保持同步，可能会抛出MissingBackpressureException或IllegalStateException。

ERROR,//会在下游跟不上速度时抛出MissingBackpressureException。

BUFFER,//上游不断的发出onNext请求，直到下游处理完，也就是和Observable⼀样了，缓存池⽆限⼤，最后直到程序崩溃

DROP,//会在下游跟不上速度时把onNext的值丢弃。

LATEST//会⼀直保留最新的onNext的值，直到被下游消费掉。}在知道了Flowable这多的策略时候，我们先来看看Flowable在单线程下是什么样的：image在策略为error情况下，并且没切换线程的时候，直接报gBackpressureException: create: could not emit value due to lack of

requests错误，该错误是告诉你没对下游的observer设置request的⽅法，这个是由于在单线程情况下，没默认给observer设置处理数据的能⼒，也即是个数，所以上游不知道下游的处理能⼒，直接抛error错误。下⾯怎么设置下游处理能⼒呢：image直接下游接收到订阅⽅法中添加t(_VALUE)，官⽅建议我们使⽤_VALUE，表⽰告诉上游，下游的处理能⼒最⼤，你尽管发送数据给我吧。那如果上游发送的数据个数⼤于下游设置的个数呢：image可以看到在发送第4个数据的时候，直接抛异常了，因为下游设置的处理能⼒是3个，每次在发送完⼀次的时候，会“削弱”，下游的处理数据的能⼒，等到发送第四个数据的时候，发现下游已经不能再处理了，直接抛异常。上⾯都是在error策略，单线程下的结果，那如果在多线程中结果会是咋样呢，还得从⼏种策略情况来看：MISSINGimage在多线程下必须设置下游的处理能⼒，因为在observeOn给下游发送数据的时候需要知道下游能处理数据的个数。上⾯我们演⽰的是上游发送128个数据，结果没有像MISSING策略所说的抛出抛出MissingBackpressureException或IllegalStateException异常信息，这是因为Flowable默认认为128个数据是上游发送最多的数据，我们可以通过这⾥找到定义的数量：image底层其实是跟RxJava1.0点的做法是⼀样的，也是把这个容量作为队列的⼤⼩，只不过RxJava1.0的容量是16个，所以再发送第129个数据的时候，会出现队列放满的问题，⼀旦放满，再往⾥⾯放数据就会出现RxJava中定义的各种策略情况，下⾯我们把发送数据改为129个，看看MISSING会出现什么情况：image看到了吧，直接抛gBackpressureException: Queue is full?!异常信息。MISSING内部的发射器⾥⾯其实啥都没做，它发送异常是在eOnSubscriber内部类的onNext时候，发现128个⼤⼩的队列满了后，给下游的observer发送onError的信息。ERROR测试代码我就不贴了，直接把上⾯上⾯的G改为mage其实和MISSING抛出的异常是⼀样的，只不过异常的message不⼀样⽽已。内部通过发送数据的时候定义⼀个AtomicLong的计数器，每次在给下游发送完⼀个数据后，会将该计数器减⼀，等到减到0的时候，直接在上游给下游发送onError的信息。BUFFER这个其实跟RxJava2.0的Observable使⽤没什么区别，输出的容量没有⼤⼩限制，也不会像RxJava1.0⼀样抛异常，请谨慎使⽤。DROPdrop是在第⼀次拿到128个数据后，第⼆次从队列中拿数据的时候，中间跟不上速度的数据抛弃了，等到下游处理完先前的128个数据的时候，才能接收后⾯96个数据，⾄于这⾥为什么是96个数字，是因为后⾯的容量减为 = prefetch - (prefetch >> 2);这个⼤⼩了，prefetch是128，⼤家⾃⼰算吧：image这⾥从96个数...从5118算起第96个数DROP是当给下游发送数据的时候，⾃⼰有个限流的策略，通过AtomicLong装载的128⼤⼩的计数器，每次发送完⼀个数据后，会将该计数器减⼀，那如果发送到了128个数据的时候，由于计数器减到0了，等到下游处理完这128个数据的时候，才会把计数器给调整到96，所以中间会出现丢数据的情况，等到下游处理完先前128个数据的时候，上游再次发数据的时候已经不会从129个数开始了，⽽且发的这96个数是随机的，因为下游处理前⾯128个数的时间是不确定的。LATEST我们先来看latest是什么样的效果，为了要区分和drop的效果，我们将发射数据改为2000的数据量：(new FlowableOnSubscribe() { @Override public void subscribe(FlowableEmitter emitter) throws Exception { for (int i = 0; i < 2000; i++) { (i); } }}, ).subscribeOn(()).observeOn(read()) .subscribe(new Subscriber() { @Override public void onSubscribe(Subscription s) { Log.d(TAG, "onSubscribe"); t(_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"); } });这⾥得到的结果会是这样的：image它是这么接收数据的：96个数...最后⼀个数，latest是先发送0到127的数据，然后中间发送96个数字，这中间会有丢失数据的，⽽最后会把最后⼀个数据发送给下游。latest没有像drop那样通过计数器的形式限制发送数据的速度，⽽是在发送数据的时候定义了AtomicReference原⼦类，把数据放在⾥⾯，所以它是每次只能存储⼀个数据，等处理完前⾯的128个数据的时候，会将AtomicLong定义的128个数量减到0，所以在下游接收完前⾯的128个数的时候，上游才能给下游发送后⾯96个数，等到最后的时候会由于缓存的是最后⼀个数，所以只能发送给下游的只能是最后⼀个数。好了，关于背压的⼏种策略就那么⼏种，其实我们总结下来：MISSION：上游没有限流，在下游⾥⾯发现队列满了，给下游发送onError的信息，该信息是gBackpressureException: 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背压的使⽤，以及他们的区别。关于Rxjava第⼆篇⽂章就介绍完了，如果还有什么不懂的地⽅可以直接留⾔问我。thanks：背压源码分析关于 RxJava 背压