原创ReactiveX之基础思路被观察者和观察者 （RxJava为例子）

Posted 2021-04-30 优才网

　　首先现在我们来认识下ReactiveX是什么。ReactiveX是一个使用观察流的异步编程接口API,它的官方网址为：http://reactivex.io/

【官方大标题图】

　　上图中官方给出的说明是一个种基于观察者流的异步编程接口API，不光如此还可以看到可以选择多个平台其中包括RxJava、Rxandroid、RxSwift、RxJS、RxScala、RxCpp、RxClojure、Rxphp等等，这就说明这个异步框架还是值得一学的，也从另外一个角度发现，这个异步框架想要描述一个颠覆传统的编程思维。像Android Framework里面其实有AsyncTask和Handler这样的库也是异步处理机制。我们慢慢看看ReactiveX到底用什么样的思维来解决异步编程的问题。

　　ReactiveX结合了观察者模式、迭代器模式和函数式编程，既然是提供给我们使用的异步框架那么自然在设计上一定是容易上手，并且简单易用，如果要研究这个框架的运作机制，那么观察者模式、迭代器模式和函数式编程需要有一个透彻的掌握。后续的文章也将会带着大家来研究研究，今天的文章以RxJava为例子带领大家一起感受下ReactiveX的玩法。除此之外ReactiveX也是可以使用在不同的地方（官方广告：ReactiveX is everywhere, and it's meant for everything.【ReactiveX无处不在，它意味着一切。】哈哈这吹的我给101分。

【官方吹水图】

【官方吹水图】

【官方吹水图】

　　总之一句话，异步操作利器、跨语法平台的框架。

 RxJava的使用

 首先要到https://github.com/ReactiveX/RxJava去获取并且编译，因为github上面使用的是Gradle构建的工程，如果不会的话可以直接百度去下载rxjava-1.1.1.jar包也可以。

 本例使用的是Eclipse和rxjava-1.1.1.jar包

【项目结构图】

 一个hello world例子。我们先看看代码以及完整的四个基本步骤。

1. 创建订阅事件对象【OnSubscribe对象】

2. 创建被观察者对象【Observable对象它决定什么时候触发事件以及触发怎样的事件】（这里如果不知道观察者对象和被观察者对象可以看看观察者模式。例如：【这里使用LOL来举例】我是莫甘娜辅助我朋友是ADC，我和我朋友说我放Q技能之后，你就给我开大，这个时候我就是被观察者，我朋友就是观察者，理解起来很简单，我放完了Q我会通知我朋友，我朋友接到通知后就会放大招。被观察者是订阅事件的通知者，观察者是接到被观察者通知之后有怎么样的反应。）

3. 创建观察者对象【Observer 即观察者，它决定事件触发的时候将有怎样的行为。这里的Subscriber是Observer的子类，Subscriber只是在Observer基础上添加了一些功能，并且还实现了Subscription接口】

4. 执行被观察者对象的订阅行为，创建了 Observable【被观察者对象】 和 Observer【观察者】 之后，再用 被观察者的subscribe() 方法将被观察者和观察者依赖在一起，这样整个框架的基本骨架就有了。写起来也比较简单

 总结上述四个流程，当然注意依赖关系先有谁后有谁的关系比较重要，因此上述代码没有写出匿名类部类的写法。并且这里的单词也是很扯蛋的（Observable被观察者、Observer、Subscriber观察者、OnSubscribe订阅事件、被观察者对象的subscribe()方法是开始订阅方法）。从最后的开始订阅来看代码为：mObservable.subscribe(mSubscriber)，你会发现【被观察者（报纸）订阅观察者（人）】这是和实际情况正好是反的，应该是mSubscriber.subscribe(mObservable);这样才对。之所以ReactiveX这样设计是为了API的一个设计的考虑。

 第一步创建订阅事件【也就是被观察者需要通知观察者的数据流】使用的是Observable.OnSubscribe接口，然后重写call方法，然后在call方法里面使用局部对象Subscriber<? super String>【这里对象名称定义为sub】来操作数据。

 sub.onNext()方法传送数据流。类似于我们以前写的事件监听器onClick，这里我们可以使用LinkedList集合来控制事件是FIFO还是FILO

 sub.onCompleted()方法说明订阅事件结束。RxJava 不仅把每个事件单独处理，还会把它们看做一个队列。RxJava 规定，当不会再有新的onNext()发出时，需要触发onCompleted()方法作为标志。

 sub.onError(Throwable e)onError(): 事件队列异常。在事件处理过程中出异常时，onError()会被触发，同时队列自动终止，不允许再有事件发出。

 在一个正确运行的事件序列中,onCompleted()和onError()有且只有一个，并且是事件序列中的最后一个。需要注意的是，onCompleted()和onError()二者也是互斥的，即在队列中调用了其中一个，就不应该再调用另一个。

 第二步创建一个被观察者对象【这里的考虑就是被观察者肯定要依赖一个订阅事件，因此实际上这里是第二步,当然这里也要注意如果先创建观察者对象也是可以当作第一步的，但是理解起来就不太优雅了（纯属个人观点）】

 Observable对象是被观察者对象，创建一个被观察者对象官方强烈建议我们尽量通过Observable.create(OnSubscribe f)来创建Obserable对象，而不是使用继承。这里的代码比较简单

 第三步观察者对象Observer

 这里使用了Observer对象的子类Subscriber来代替观察者对象。因为Subscriber观察者对象不光增加了一些功能而且还是内部实现了Subscription接口

 实现观察者对象的时候重点是需要重写观察者对象的几个方法，用来处理得到被观察者的通知之后做出的种种反应

 onStart()获取通知之前做点什么

 onNext(T t)获取通知流做点什么

 onCompleted()获取通知结束标记做点什么

 onError(Throwable e)获取通知错误标记做点什么

 第四步被观察者使用subscribe方法开始订阅观察者（反向思维确实有点意思）

 第四步要注意的是被观察者对象的订阅方法，而不是观察者的订阅方法，所以这里是反向思维有点别扭，但是你要是看我上面举的LOL例子就觉得不别扭了，如果是报纸和人的例子就别扭了。

 UML类图

　　上面介绍的ReactiveX框架仅仅只是开始，并且不是异步行为。在ReactiveX的默认规则中，被观察者事件的通知和消费者的反应都是在同一个线程，这只能说是一个同步观察者模式，然而ReactiveX框架的异步观察者模式才是重头戏，如果要实现异步【后台处理，前台回调】那么就需要知道另外一个类Scheduler。本系列下一篇将会讨论关于ReactiveX框架的异步规则。希望大家多多关注。

推荐阅读↓