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

这篇文章给大家介绍如何解析反应式编程 RxJava 设计原理，内容非常详细，感兴趣的小伙伴们可以参考借鉴，希望对大家能有所帮助。

一、ReactiveX 与 RxJava

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

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

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

二、RxJava中的概念

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

从图中我们可以看出：

• 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 设计原理就分享到这里了，希望以上内容可以对大家有一定的帮助，可以学到更多知识。如果觉得文章不错，可以把它分享出去让更多的人看到。