浅析RxJava 1.x&2.x版本使用区别及原理：ObservableFlowable等基本元素源码解析

Posted 鸽一门

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了浅析RxJava 1.x&2.x版本使用区别及原理：ObservableFlowable等基本元素源码解析相关的知识，希望对你有一定的参考价值。

RxJava开源框架的风靡程度在Github上无需多言，它带来的响应式编程模式和线程随意切换、嵌套请求、背压等功能给了开发者耳目一新的体验，更是成为了大多数APP中常用的RxJava+Okhttp/Retrofit+MVP/MVVM/Clean黄金组合中的一员。我犹记得知乎团队在去年线下还开展过一次线下RxJava交流会，邀请了扔物线讲解相关知识及体验，可见各大厂对此的重视度。如非要列举一个RxJava缺点，那就是学习曲线高，但是后期收益高。我记得去年初个人开发一款教育APP时使用的还是RxJava 1.x版本，而今年3月31号官方已经宣布停止对1.x 版本的维护，在用过2.x版本后，可见对之前版本的改进。

特此，此系列文章来浅析RxJava 1.x&2.x版本使用区别及原理，分析源码的过程是艰难而又通过的，深入理解之后的豁然开朗更是无法比拟的。需要强调的是这几篇文章重点部分在于分析源码，并不适合作为框架的初学篇，熟悉其用法后再观此博客讨论学习。另外，此文章将着重于RxJava两个版本用法的对比，由此为出发点分析两者的源码及运作原理。

此系列文章将分成以下三个部分讲解：

  • RxJava 1.x&2.x版本的基本元素（Observable、Subscriber、Flowable、Observer等）流程源码分析
  • RxJava 1.x&2.x版本的操作符（map、lift）源码分析
  • RxJava 1.x&2.x版本的线程变换源码分析

一. RxJava简介

说到RxJava开源框架，首先第一反应是响应式编程，它是一种面向数据流和变化传播的编程范式。

  • 数据流：只能以事先规定好的顺序被读取一次的一个数据序列。在计算机中是数据，在现实中可以是任意对象组成的有序队列。
  • 变化传播：类似于设计模式中的观察者模式，一旦监视的事件状态或其他发现变化，立即通知观察者。
  • 编程范式：是计算机编程的基本风格或典范模式，例如面向对象编程、面向过程编程。

可以把响应式编程比喻为流水线上的生产工作，数据流则是流水线上加工的一系列物品，变化传播类似于加工过程中多个环节的转化，而编程范式就相当于不同的加工环节，其工作方式不同。如此，方便理解。

以下是RxJava的官方介绍语：https://github.com/ReactiveX/RxJava

A library for composing asynchronous and event-based programs by using observable sequences

  • asynchronous异步，说明RxJava是一个异步的库，基于回调的。
  • event-based基于事件，说明RxJava是一个事件分发库，消息传递的库。

It extends the observer pattern to support sequences of data/events and adds operators that allow you to compose sequences together declaratively while abstracting away concerns about things like low-level threading, synchronization, thread-safety and concurrent data structures.

更进一步解释，它扩展了观察者模式来支持数据/事件序列，并添加了运算符，使开发者可以声明性地组合序列，同时抽象出对低级线程、同步、线程安全性和并发数据结构等事物的观察。

二. RxJava 1.x基本元素源码分析

1. 基本元素用法介绍

下面通过一个简单的RxJava 1.x的代码实例来了解 1.x版本中的重要元素：

import rx.Observable;
import rx.Observer;
import rx.Subscriber;
import rx.Subscription;

      Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() 
            @Override
            public void call(Subscriber<? super String> subscriber) 
                if (!subscriber.isUnsubscribed()) 
                    subscriber.onNext("test");
                    subscriber.onCompleted();
                
            
        );

        Subscription subscription = observable.subscribe(new Observer<String>() 
            @Override
            public void onCompleted() 
                System.out.println("onCompleted");
            
            @Override
            public void onError(Throwable e) 
            
            @Override
            public void onNext(String s) 
                System.out.println("onNext:" + s);
            
        );
  • 首先通过调用Observable的create() 方法创建了一个Observable对象，传入的参数是OnSubscribe接口，故而在参数内实现其接口的call 方法。
  • 再调用Observable对象的订阅subscribe 事件方法，其参数传入Observer接口，故而在参数内实现其接口的onCompleted()、 onError(Throwable e)、onNext(T t)方法。
  • 在创建Observable中实现OnSubscribe接口的call 方法中，注意其参数为Subscriber订阅者，调用订阅者的onNext方法传入数据，再调用onCompleted 方法。

以上是RxJava 1.x版本的简单使用，一个简单的例子已展示出其库的基本元素：

  • Observable
  • Observer
  • Subscriber
  • OnSubscriber
  • Subscription

2. 基本元素源码分析

在了解以上基本元素后，接下来进行源码分析，首先从最简单的Subscription入手：

（1）Subscription 接口

Subscription是Observable调用subscribe订阅事件方法后返回的一个接口，其内容也很简单，两个方法，一个解除订阅的unsubscribe()方法，一个是判断是否解除的isUnsubscribed() 方法，源码如下：

（2）Observer 接口

Observer是Observable调用subscribe订阅事件方法中传入的参数，也是一个接口，三个待实现的方法，分别是回调完成方法onCompleted()、 错误处理方法onError(Throwable e)、数据接收方法onNext(T t)。

（3）Subscriber 抽象类

在创建Observable时，需要给create传入参数Observable.OnSubscribe接口，并实现其接口的call方法，此方法的参数类型就是Subscriber，而开发者也是在此方法中调用onNext、onComplete，因此可以推测Subscriber必然实现了Observer 接口。

仔细查看源码，确实如此，它还实现了Subscription 接口中的unsubscribe()、isUnsubscribed()两个方法，简单做了封装，但并未详细实现Observer 接口中的三个方法，因此只是一个抽象类。

（4）OnSubscriber 接口

OnSubscriber是在创建Observable时传入create 方法中的参数类型，也是一个接口。此接口又是继承于Action1 接口，Action1 接口中有一个未实现的call方法，而Action1 接口又继承于Action接口，Action接口是一个空实现，最后它又继承于Fcuntion接口，Fcuntion接口也是一个空实现。

OnSubscriber -> Action1 -> Action -> Fcuntion

（5）Observable

  • create方法

首先查看Observable的静态创建create方法，可见其只是多做了一层封装，new这个对象时，将参数onSubscribe传入构造方法中，而RxJavaHooks.onCreate(f)也只是多做了一个判断，最终返回的还是onSubscribe。

  • subscribe方法

再看subscribe 方法：

先忽略掉一开始的判断，直接查看最后一行代码，这里重载了另外一个subscribe 方法，参数为Subscriber类型，因此new ObserverSubscriber<T>(observer)，这里使用Subscriber将我们传入的observer接口做了一层简单的封装，来查看ObserverSubscriber的具体实现：

可见，这里使用Subscriber将我们传入的observer接口做了一层简单的封装。还是回到它重载的另一个subscribe 方法，它才是研究的重点：

由以上代码可知，从调用的参数为observer接口的subscribe 方法内做了一层封装，调用了参数为subscriber抽象类的subscribe 方法，最终调用的是参数为subscriber、observable的静态subscribe 方法。

3个subscribe 方法方法的层次调用，为的只是参数的封装获取，而最终在静态subscribe 方法中的重点处理如下：

第一行代码调用OnSubscriber接口的call方法，这意味着我们创建Observable时实现的call方法回被调用，那么call方法中对数据传递、标志完成的操作会执行，即实现的Observer接口中的onNext方法中接收到数据，onComplete()方法也被调用。最后返回Subscription。

3. 基本元素总结

这整个过程可以用一个简单的打电话事例来理解，Observable即打电话的人，拨通号码的过程相当于subscribe确定了接电话的人Subscriber，OnSubscribe则类似电话连接的线路，开启信息的传递。两者开始通话时，状态建立，Observer则监视整个状态，完成通话数据传递、完成、错误过程。而当Subscriber取消通话时，两者的关系解除，专门用于描述两者关系的Subscription返回。

（1）Observable

  • 被观察者
  • 可通过Observable创建一个可观察到序列
  • 通过subscribe去注册一个观察者

（2）Observer

  • 用于接收数据的观察者
  • 作为Observable的subscribe方法的参数

（3）Subscription

  • 订阅，用于描述被观察者和观察者之间的关系
  • 用于取消订阅和获取当前的订阅状态

（4）OnSubscribe

  • 当订阅时会出发此接口的调用
  • 在Observable内部，实际作用时向订阅者发布数据

（5）Subscriber

  • 实现了Observer和Subscription接口
  • 只有自己才能阻止自己

三. RxJava 2.x基本元素源码分析

此处涉及到一个背压问题，在RxJava 1.x版本中有的Observable可以处理，有的不行，而在2.x版本中直接将此分离出来，Observable不处理背压相关问题，而是由Flowable这样一个新的类来处理，后续详解。

1. 基本元素介绍

以下是RxJava 2.x的简单使用：


import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;

                Observable.create(new ObservableOnSubscribe<String>() 
                    @Override
                    public void subscribe(ObservableEmitter<String> e) throws Exception 
                        if (!e.isDisposed()) 
                            e.onNext("test");
                            e.onComplete();
                        
                    
                ).subscribe(new Observer<String>() 
                    @Override
                    public void onSubscribe(Disposable d) 
                        System.out.println("onSubscribe");
                    
                    @Override
                    public void onNext(String value) 
                        System.out.println("onNext:" + value);
                    
                    @Override
                    public void onError(Throwable e) 

                    
                    @Override
                    public void onComplete() 
                        System.out.println("onCompleted");
                    
                );

两个版本的使用方式都类似，不再赘述，直接分析其区别：

  • 在通过调用Observable的create() 方法创建对象时，传入的参数是ObservableOnSubscribe接口，实现的是其接口的subscribe 方法，方法内的参数是ObservableEmitter，不再是1.x版本的OnSubscribe接口（call 方法）。
  • 在调用Observable对象的订阅subscribe 事件方法，其参数传入的Observer接口，多了一个需要实现onSubscribe(Disposable d)方法，方法内的参数是Disposable。

RxJava 2.x中的基本元素如下：

  • Observable
  • Observer
  • Disposable（Subscription）
  • ObservableOnSubscriber（OnSubscriber）
  • Emitter

在源码分析之前先提出一个问题，Rxjava 2.x的使用相比较1.x 版本有一个明显之处是调用subscribe方法时需要实现的方法多了一个onSubscribe，从上面打印日志可知，其方法的执行是要优先于创建Observable时实现的subscribe方法，为何？通过以下源码分析答案便可水落石出。

2. 源码分析

RxJava 2.x版本相较于 1.x版本的使用类似，但又有稍许变化，此处只着重讲解不同的部分：

（1）Observer接口

多了一个void onSubscribe(Disposable d);方法，用于观察者取消事件订阅，来查看Disposable接口组成：（注意：2.x版本新增的Disposable可以做到切断订阅事件的操作，让观察者Observer不再接收上游事件，避免内存泄漏）

接口中两个方法，一个dispose方法，另一个事检测是否dispose方法，其结构与Subscription类似。

（2）ObservableOnSubscribe接口

是创建Observable时传入的接口参数，在2.x版本中单独独立出来了。为观察者提供了取消订阅连接的功能，该接口中的subscribe方法用于接收ObservableEmitter的实例，该实例允许用安全的方式取消事件。

（3）ObservableEmitter接口

前两个接口我们一直都在强调2.x版本新增的Disposable切断订阅事件，使得观察者不再接收上游事件的功能，可预先此接口也是为它所用，作用是设置Emitter的disposable和cancellable

继续查看Emitter接口的组成，会发现其中包含的三个方法竟然与Observer接口完全相同，其中缘由后续讲解。

（4）Observable

  • create方法

在简单了解以上结构后，直袭老窝，查看Observable的create方法：

此方法中可以得出两个信息，第一个是调用了RxJavaPlugins的静态onAssembly方法，第二个是传入此方法的参数，将ObservableOnSubscribe接口通过ObservableCreate做了一次封装。首先来了解onAssembly方法：

此方法中的一个关键成员变量onObservableAssembly，它最初被赋值为null，为外界提供了set方法，因此当我们刚开始调用时f 被判断为null，直接将source返回。再来查看new ObservableCreate<T>(source) 具体构成：

可见ObservableCreate类被final 修饰，继承于Observable，值得注意的是它实现了一个从父类继承而来的subscribeActual 方法，此处暂时不讲解，它主要作用于Observable的subscribe方法，结合此方法再讲。

  • subscribe方法

再次回到Observable，查看subscribe方法：

首先查看到observer = RxJavaPlugins.onSubscribe(this, observer);，在Observable的create中也出现了RxJavaPlugins相关用法，而此处它的作用也是类似，就是将传入的参数observer返回，重点在于后面的subscribeActual(observer);，也就是刚介绍ObservableCreate实现的subscribeActual 方法。

回到上上张图，subscribeActual 方法中首先借助CreateEmitter将传入的参数observe接口做了一次封装，接着调用observer的onSubscribe(parent)方法，在我们实现该方法时的参数是disposable，而这里传入的是CreateEmitter参数，因此可以推断CreateEmitter必然实现了Disposable接口。接下来的source.subscribe(parent);，即调用ObservableOnSubscribe接口的subscribe方法，传入实现了Disposable接口的CreateEmitter，正式触发事件的订阅流程！

到了这里，必须强调了一下，再次回顾RxJava 1.x版本中Observable的subscribe处理，通过调用创建Observable传入的OnSubscribe接口的call方法正式触发订阅事件，后续Observe接口中onNext、onComplete方法才回被调用。而RxJava 2.x版本中的处理亦如是，只不过OnSubscribe接口换成了ObservableOnSubscribe接口，call方法换成了subscribe方法，参数由subscriber更换成了ObservableEmitter，这些变换也是RxJava 2.x的改进，新增的Disposable切断订阅事件，使得观察者不再接收上游事件的功能，来避免内存泄漏。由此可见以上处理过程，RxJava 2.x 与 1.x的区别不大。

以上订阅流程基本已经结束，还有几个关键点再来补充补充。之前我们推测CreateEmitter必然实现了Disposable接口，不仅如此，对比上图再查看两版本的最后一步源码，RxJava 1.x版本调用OnSubscribe接口传入的参数是实现了observer接口的subscribe， 2.x版本传入的则是实现了Disposable接口的CreateEmitter，Disposable是 2.x新增的概念，暂且不予考虑。按理说这两者性质应当相同，1.x版本中的subscribe实现了observer接口，有onNext、onComplete、onError基本方法，继续猜测CreateEmitter是否也应该继承了2.x版本中的observer接口，需要注意的是2.x版本中的observer接口中多了一个subscribe方法，对于一些运算符所需的基本接口而言，这个方法并不符合需求，只需要onNext、onComplete、onError基本方法。

  • 为何observer接口中onSubscribe方法最先执行，ObservableOnSubscribe接口中的subscribe方法在后？

以上的源码正好解释了代码执行的一个顺序，首先执行的是Subscribe接口中的onSubscribe方法，再是ObservableOnSubscribe接口中的subscribe方法，由于在此方法中做了数据传递、标志完成等操作，因此Subscribe接口中的onNext、onComplete方法会被调用。

  • CreateEmitter

结合之前介绍的创建Observable传入ObservableOnSubscribe接口后，实现的subscribe方法中的参数是ObservableEmitter接口，1.x版本中是Subscribe抽象类，而ObservableEmitter继承于的Emitter接口中正好有onNext、onComplete、onError基本方法，因此CreateEmitter必然于Emitter有所关联！至此，分析出了CreateEmitter实现了Disposable、Emitter接口，查看详细源码：

可见我们分析并无误，此类中也对onNext、onComplete、onError方法做了一些基本封装，我们在创建Observable传入ObservableOnSubscribe接口时实现的subscribe方法中，虽然使用的是其参数ObservableEmitter调用onNext传入数据，但追溯到此根源，最终还是交由CreateEmitter来处理，即我们实现的observer接口来调用onNext等方法。

  • DisposableHelper

分析完Emitter接口后，CreateEmitter的核心还没有结束，它确实是一座宝库。接下来是它对Disposable的处理，它实现的方法中出现了一个关键部分，即借助了DisposableHelper来处理Disposable相关事务：

查看DisposableHelper，它是一个枚举类，实现了Disposable接口，内有DISPOSED，作为单例模式存在。因此它实现的判断isDisposed方法，直接将参数与DISPOSED比较即可，源码如下：

3. 基本元素总结

在以上基本元素的对比上，两个版本其实没有很大的区别，只是部分写法做了改变，2.x版本中将精华ObservableCreate独立出来，但其核心内容还是与1.x 相同，调用传进来的OnSubscribe接口中的subscribe（call）方法，并传入observe接口到其中。而2.x中还特地使用了CreateEmitter对传入的observe接口做了包装，因此我们手动调用onNext方法时，实际上就是通过observe接口调用这些方法。

（1）Observable

  • 被观察者，不支持背压
  • 可通过Observable创建一个可观察到序列
  • 通过subscribe去注册一个观察者

（2）Observer

  • 用于接收数据的观察者
  • 作为Observable的subscribe方法的参数

（3）Disposable

  • 和RxJava1的Subscription作用相当
  • 用于取消订阅和获取当前的订阅状态

（4）ObservableOnSubscriber

  • 当订阅时会出发此接口的调用
  • 在Observable内部，实际作用时向订阅者发布数据

（5）Emitter

  • 一个发射数据的接口，和Observer 的方法类似
  • 本质是对Observer和Subscribe的包装

4. 背压

概念

  • 异步环境下产生的问题：同步环境下会等待一件事处理完后再进行下一步，而异步环境下是处理完一件事，未等它得出结果接着处理下一步，在获得结果之后进行回调，再处理结果。
  • 发送和处理速度不统一：例如生产者生产出的产品放置到缓存队列中，供消费者消费。若生产者生产的速度大于消费者消耗的速度，则会出现缓存队列溢出的问题。
  • 是一种流速控制及解决策略：例如背压中的丢弃策略，一旦发现缓存队列已满，为了整个过程顺利进行，则会丢弃最新产生的产品，避免溢出，因此背压也是一种流速控制的解决策略

背压Flowable方法的简单使用如下：（注意Flowable专门用于背压使用，在onSubscribe方法中需要手动做一个响应式拉取，即s.request(Long.MAX_VALUE);，否则不会调用onNext 方法）

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;

                Flowable.create(new FlowableOnSubscribe<String>() 
                    @Override
                    public void subscribe(FlowableEmitter<String> e) throws Exception 
                        if (!e.isCancelled()) 
                            e.onNext("Flowable test");
                            e.onComplete();
                        
                    
                , BackpressureStrategy.DROP).subscribe(new Subscriber<String>() 
                    @Override
                    public void onSubscribe(Subscription s) 
                        s.request(Long.MAX_VALUE);
                        System.out.println("onSubscribe");
                    
                    @Override
                    public void onNext(String s) 
                        System.out.println("onNext:" + s);
                    
                    @Override
                    public void onError(Throwable t) 
                    
                    @Override
                    public void onComplete() 
                        System.out.println("onCompleted");
                    
                );

  • 首先创建通过调用Flowable的create方法创建实例，传入两个参数，第一个是OnSubscribe接口实例，并实现其subscribe方法，第二参数则是背压特有的背压策略；
  • 调用Flowable的subscribe方法，你会发现不同于之前，传入的并非是observer接口，而是Subscriber接口，实现onSubscribe、onNext、onComplete、onError方法。

以上是RxJava 2.x版本中背压的简单使用，基本元素如下：

  • Flowable
  • Subscriber
  • Subscription
  • OnSubscriber
  • Emitter

注意：以上使用Flowable时，包括在实现subscribe方法中的数据传递等操作时，如果没有在Subscriber接口中的onSubscribe方法中设置响应式拉取，即s.request(Long.MAX_VALUE);，onNext方法中打印的log不会显示！为何？阅读以下源码将获得解释。

5. 背压源码分析

（1）Subscriber接口

接口中的方法与2.x版本中的observer接口中方法相同，即onSubscribe、onNext、onComplete、onError，源码如下：

（2）Subscription

在查看Subscriber接口中的onSubscribe方法，发现其参数类型为Subscription，注意此接口与1.x版本不同！其中request方法是通过响应式拉取解决背压问题，cancel()方法则是停止发送数据，清空所有资源。 源码如下：

（3）FlowableOnSubscribe

再看到调用Flowable的create方法时传入的参数FlowableOnSubscribe，类似于Observable的ObservableOnSubscribe，接口中只有一个subscribe方法，参数为FlowableEmitter，源码为：

此处的FlowableEmitter同样类似于Obsa vableEmitter，都继承于Emitter接口，内含熟悉的三个onNext、onComplete、onError方法，上述意见介绍过，此处不再粘贴源码。

（4）Flowable

首先Flowable抽象类实现了Publisher接口，其中只有一个subscribe方法，这与无背压版的Observable类似。依旧先从create方法开始分析：

  • create方法

源码如上，太熟悉的格式了，这里依旧调用的是RxJavaPlugins.onAssembly方法，内部还是回将参数直接返回，所以重点在于参数new FlowableCreate<T>(source, mode))，与Observable类似，都有一个XXXCreate类，再预测其内部必有一个subscribeActual方法，来查看FlowableCreate源码组成：

果不其然，格式完全相同，只不过原先实现的observer接口变成了背压独有的Subscriber接口，因此方法中的参数变成了Subscriber。此时暂不介绍，等讲解至Flowable的subscribe方法时综合讲解。

  • subscribe方法

熟悉的逻辑操作，简直就是将Observable那一套搬过来了，RxJavaPlugins.onSubscribe最后还是将参数Subscriber返回，重点还是在于调用Flowable的抽象方法subscribeActual，即调用了继承Flowable的FlowableCreate的subscribeActual方法，查看其具体实现：

由此可见背压的策略有MISSING、ERROR、DROP、LATEST这几种，而不同的策略对应不同的BaseEmitter，与Observable做法相同，再包装传入的Subscribe参数，接着调用Subscribe接口的onSubscribe方法，即t.onSubscribe(emitter);，传入指定策略对应的Emitter，此时我们使用Flowable实例去subscribe时实现的Subscriber接口中的方法onSubscribe被回调。接着源码中source.subscribe(emitter);，即调用FlowableOnSubscribe接口中唯一的subscribe方法，并传入emitter，意味着订阅流程开启，代码中创建Flowable时实现的subscribe方法被调用，里面调用的onNext等方法依次执行。

  • 方法执行顺序解释

此处的源码也解释了代码执行的一个顺序，首先执行的是Subscribe接口中的onSubscribe方法，再是FlowableOnSubscribe接口中的subscribe方法，由于在此方法中做了数据传递、标志完成等操作，因此Subscribe接口中的onNext、onComplete方法会被调用。正因如此，我们才回在最先执行的onSubscribe方法中进行响应式拉取的设置。

  • 背压策略BaseEmitter

以上基本流程的源码分析基本结束，接下来查看它是如何结局背压问题，首先来查看所有背压策略的父类BaseEmitter，源码如下：

其实它的作用同Observable中的CreateEmitter相同，都是实现了onNext、onComplete、onError相关接口，在其中做了简单处理，使得最后我们在代码中通过FlowableEmitter调用onNext相关方法，实际上是通过我们传入并实现的Subscribe接口本身自己调用。但是不同之处在于，BaseEmitter还实现了背压特有的request(long n)方法，查看源码：

首先判断传入的参数是否有效（不得小于0），接着调用BackpressureHelper的add方法， 传入BaseEmitter实例和指定的数值，查看其方法实现：

如上所示，首先获取默认值r，为0，接着调用addCap方法，其实就是返回n，最后将AtomicLong类型的requested设置为n，将其返回。你可能感到疑惑，这样做有什么意义，以DROP丢弃策略为例，来查看DropAsyncEmitter具体实现的onNext方法：

答案呼之欲出，首先调用get方法取出对应的值，之前说过默认值时为0，这意味着如果我们不在最先执行的onSubscribe方法中设置这个值，那么此处获取的值为0，它并不会执行actual变量（实质为Subscribe接口实例）的onNext方法！

以上也解决了一开始抛出的疑问，为何如果没有在Subscriber接口中的onSubscribe方法中设置响应式拉取，即s.request(Long.MAX_VALUE);，onNext方法中打印的log则不会显示。

6. 背压总结

在了解了RxJava 2.x版本的基本元素使用后，Flowable背压学习也不算太难，皆是换汤不换药，通过调用onSubscribe接口中的方法，并传入observer接口，来实现整个订阅流程的开启。

（1）Flowable

  • 被观察者，支持背压
  • 可通过Observable创建一个可观察到序列
  • 通过subscribe去注册一个观察者

（2）Subscriber

  • 一个单独接口（并非是1.x版本中的抽象类），和Observer的方法类似
  • 作为Flowable的subscribe方法的参数

（3）Subscription

  • 订阅，和1.x版本有所不同
  • 支持背压，有用于背压的request方法

（4）FlowableOnSubscribe

  • 当订阅时会出发此接口的调用
  • 在Flowable内部，实际作用时向订阅者发布数据

（5）Emitter

  • 一个发射数据的接口，和Observer 的方法类似
  • 本质是对Observer和Subscribe的包装

四. 小结

以上将Rxjava 1.x和2.x版本的基本元素源码流程发那些完了，类、接口直接的方法互相调用可能稍有繁琐，最后借助总结部分的UML图来理解，整体分析下来，你会发现，RxJava 2.x相较于1.x 的改动没有很大，最值得令人赞颂的便是将背压单独独立出来，避免与Observable混淆在一起，这也导致两者涉及的接口也随之增加，在理解2.x 的Observable订阅套路后，Flowable的套路简直就是照搬照套，只是多了对背压策略BaseEmitter的处理，整体结构不变。

源码分析完后，可以解惑在分析2.x 源码之前存在的两个疑问：

  • 为何observer接口中onSubscribe方法最先执行，ObservableOnSubscribe接口中的subscribe方法在后？
  • 为何没有在Subscriber接口中的onSubscribe方法中设置响应式拉取，即s.request(Long.MAX_VALUE);，onNext方法不会被调用？

再将两个版本基本元素原理的易混淆点总结：

  • Observable对象create方法：
    • RxJava 1.x 中调用Observable的create方法创建实例时，是直接new 一个Observable对象，直接传入OnSubscribe接口。
    • 2.x 版本是new 一个ObservableCreate对象（继承于Observable），类似于多做了一层封装，与之对应的OnSubscribe接口作为其成员变量传入。
  • Observable对象subscribe方法：
    • RxJava 1.x 内部是通过重载subscribe方法，最终核心处就是调用OnSubscribe接口的call方法，传入Subscriber类型参数正式触发订阅事件。
    • 2.x 版本最终调用的是一开始创建Observable对象的subscribeActual方法，方法内的最终核心处也是调用OnSubscribe接口的subscribe方法，传入ObservableEmitter类型参数正式触发订阅事件。

以上基本的元素源码分析后，对两个版本的了解也稍有深入。

若有错误，虚心指教～

以上是关于浅析RxJava 1.x&2.x版本使用区别及原理：ObservableFlowable等基本元素源码解析的主要内容，如果未能解决你的问题，请参考以下文章

浅析RxJava 1.x&2.x版本区别及原理：maplift操作符源码解析

浅析RxJava 1.x&2.x版本使用区别及原理：ObservableFlowable等基本元素源码解析

浅析RxJava 1.x&2.x版本区别及原理：1.x 版本 Scheduler调度线程变换及subscribeOnobserveOn方法源码解析

浅析RxJava 1.x&2.x版本区别及原理：1.x 版本 Scheduler调度线程变换及subscribeOnobserveOn方法源码解析

浅析RxJava 1.x&2.x版本区别及原理：1.x 版本 Scheduler调度线程变换及subscribeOnobserveOn方法源码解析

RxJava 2.0 简单介绍