反应式编程 RxJava 设计原理解析

Posted 2021-05-21 涂程

好文推荐：
作者：Yunjie Ma

一、ReactiveX 与 RxJava

ReactiveX 的全称为Reactive Extension，一般缩写为 Rx，即我们平常所说的反应式编程。其设计原理主要使用了观察者模式，区分数据的生产者和消费者，通过事件流的方式进行数据的异步处理。

RxJava 是 ReactiveX Java语言的实现，其编程体验与Java 8中的函数式编程和流（Stream）有很大的相似之处，在掌握了Java8的相关知识后，你可以很轻松的就上手使用 RxJava。

本篇文章主要聚焦对RxJava中几种主要的设计模式的理解，通过梳理Observable的相关类图以及讲解这些类之间的关系，让大家能够更清晰的理解RxJava中事件驱动的工作原理。

二、RxJava中的概念

首先我们写一个简单的RxJava的程序，把数组中的元素作为事件发送，最终由消费者打印在控制台：

我们以这段简单的代码为基础，讲解下贯穿整个ReactiveX设计的四个概念：观察者，被观察者，事件，订阅。

• **观察者：**对事件进行响应的对象，也可以称作消费者，在上述的代码中，subscirbe方法的参数是一个Consumer对象，该对象后续会被包装成一个LambdaObserver对象，即为这段代码中的观察者（消费者）。

• **被观察者：**产生事件的对象，也可以称作生产者，在上述代码中，Observable.fromArray(…)返回的是一个Observable对象，即为这段程序的被观察者（生产者）。

• 事件：RxJava中存在四种事件流：onSubscribe（订阅事件），onNext（正常事件），onError（异常事件），onComplete（完成事件）。在上述代码中，是将数组中的元素作为onNext事件中的数据进行发送。

• **订阅：**创建观察者与被观察者之间观察关系，对应着上述代码中的subscribe()方法。RxJava的事件驱动模型是一种“拉模型”，在观察者没有进行事件订阅之前是不会有事件产生的，只有观察者进行订阅后，才会触发被观察者生产事件。

对上述代码进行时序分析，可以清晰的看出这一段代码的运行过程，最终由FromArrayDisposable生产了onNext和onComplete事件，并通知Observer进行消费。

与此同时，我们也看到，简单的一行代码，竟然涉及这么多类的交互，如果增加一些其他的操作符，我们对整个程序把控起来就没那么容易了，下面我们将通过分析RxJava中的一些主要的设计模式，剖析类与类的关联关系，来更清晰地理解RxJava的工作原理。

三、 集大成者Observable

在整个数据处理的过程中，Observable可以说是最重要的一个对象。从上面的时序图可以看出，客户端（消息的生产者或者消费者）只和Observable进行交互，观察者和被观察者之间关系的创建也是由Observable去实现，而不用我们显示的编码实现，这大大降低了我们使用观察者模式的成本。

那么Observable主要有哪些作用呢，我们首先来看下和Observable相关的类图：

从图中我们可以看出：

• Observable实现了ObservableSource接口，从字面意思就可以理解，这是一个提供观察能力的接口，所以Observable的一大能力是供观察者进行事件订阅，而进行事件订阅的方法实现就是调用Observable的subscribe()方法

• Observable是一个抽象类，它提供了subscribeActual模板方法供子类实现，从源码中可以看出，Observable的subscribe()方法最终会委托子类的subscribeActual()方法实现，这个方法会建立生产者与消费者之间的关联关系。

• 除此之外，Observable还是一个工厂类，它提供了静态方法fromArray()、create()等用来创建具体的可观察对象，同时还提供了flatMap()、concatMap()等操作方法对可观察对象进行包装。

Observable的存在让生产者和消费者完全的解耦了，生产者只需关注自己生成何种Observable对象，而消费者也只需关注自己观察的是哪种Observable。

在实际的应用中，Rxjava已经提供了各种各样的操作符供我们使用，生产者只需要调用Observable中相应的方法即可以生成所需的可观察对象，供消费者进行事件订阅。消费者只需调用可观察对象的subscribe()方法即可与生产者建立观察关系，极其方便。

四、 真实的观察

观察者模式是RxJava设计的核心思想，在观察者模式中总是存在观察的对象和被观察的对象，从上文的解析中也可以看出Observable更多的是一个控制器的作用，而并非真正的事件的来源。那么在RxJava中，什么才是真正的生产者，什么才是真正的消费者呢。

我们来分析下以下三种常见的Observable：

先简单介绍下这几个Observable的作用，fromArray的作用是将数组中的元素作为onNext事件发送，create的作用是发送自定义事件，just的作用是发送单个事件。

上一小节有讲到实际的订阅行为是由各个Observable类中subscribeActual()方法实现的，我们来看下这三个类的subscribeActual()方法。

除去细枝末节，这三个方法都可以分成以下三步

1. 创建被观察者对象，并传入观察者observer，建立两者的关联关系；

2. 触发onSubscribe事件，观察者响应该事件；

3. 进行事件的拉取，我们可以进入到d.run()，source.subscribe(parent)，sd.run()这些方法的内部看一些，可以看到这些方法就是在发送onNext(),onError(),onComplete()等事件。

下图是整个流程中的相关类图。实际事件的发送者是FromArrayDisposable等对象，而实际的观察者，则是一个实现了Observer接口的实体类。如果我们在subscribe时传入的是一个lambda表达式，之后会被包装成一个默认的LambdaObserver对象，进行事件消费。

五、 包装的必要

RxJava中提供了丰富的操作符，比如flatMap，concatMap等可以对事件转换，subscribeOn，observableOn等可以对生产和消费的线程进行控制。这些操作符实际上调用了Observable中的包装方法对原有的可观察对象进行包装，返回了一个增强了的可观察对象。

操作符种类繁多，在这就不一一举例，我们以flatMap为例，分析一下这些操作符是如何工作的。

首先，flatMap操作会返回一个ObservableFlatMap对象，在创建这个对象时，会将原始的Observable对象作为构造函数的参数传入。

查看其核心方法subscribeActual，

可以看到这一类对象的subscribeActual方法和上一节中的方法不太一样，这里面并没有去实际的创建观察关系，而是做了两件事：

1. 对观察者进行增强，将其包装成为MergeObserver对象，由其对产生的时间进行响应。

2. 再调用source的subscribe方法，这里source就是前面构造函数中传入的Observable对象，由其再进行观察关系的建立。
   下图是RxJava中装饰器模式的相关类图：所有的包装类都继承了AbstractObservableWithUpstream类，该抽象类有一个类型为ObservableSource的成员函数，用来持有被装饰的对象。

Observable是支持链式操作的，就和Java 8中的Stream一样，我们来考虑这样一行代码。

我们在分析上面这串代码时，一定会凌乱非常，在看源码时也会看到前面忘掉后面，但是如果我们对RxJava的包装流程足够了解的话，就可以很轻松的对上述代码进行分析。

六、 小结

RxJava的封装足够强大，可以让我们很方便的进行使用和扩展，但这也给我们理解其真实的工作原理带来了难度，如果我们对整个事件的处理过程处于一知半解的状态，那我们就无法从容的对服务进行异步编排，在实际开发过程中也难以发现问题的根源。

本文主要分析了RxJava中主要的设计模式，其中有模板模式、工厂模式、观察者模式、装饰器模式，理解了这些设计模式，理解了RxJava中类与类的关系，我们就能够对整个事件的处理流程了然于胸，分析代码时也能够事半功倍。

最后

大家如果还想了解更多android 相关的更多知识点，可以点进我的GitHub项目中：https://github.com/733gh/GH-Android-Review-master自行查看，里面记录了许多的Android 知识点。最后还请大家点点赞支持下！！！