RxJava学习指南(一)——基础部分

RxJava学习指南

版本：2
作者：陈小默
声明：禁止商用，禁止转载

一、RxJava介绍

RxJava是一种响应式编程解决方案，其核心采用的是基于扩展的观察者模式的设计方法。

1.1 响应式编程

响应式编程是一种基于异步数据流概念的编程模式。数据流就像一条河：它可以被观测，被过滤，被操作，或者与另外一条流合并为一条新的流。

响应式编程的一个关键概念是事件。事件可以被等待，可以触发过程，也可以触发其它事件。

响应式编程最通用的一个场景是UI：我们的移动App必须做出对网络调用、用户触摸输入和系统弹框的响应。在这个世界上，软件之所以是事件驱动并响应的是因为现实生活也是如此。

1.2 观察者模式

在RxJava的世界里，我们有四种角色：

• Observable
• Observer
• Subscriber
• Subjects

Observables和Subjects是两个“生产”实体，Observers和Subscribers是两个“消费”实体。Observable对应于观察者模式中的被观察者，而Observer和Subscriber对应于观察者模式中的观察者。

在观察者模式中，观察者无需主动询问被观察者状态是否改变，而是当被观察者状态变化时会主动推送消息给观察者。

观察者模式很适合下面这些场景中的任何一个：

• 当你的架构有两个实体类，一个依赖另一个，你想让它们互不影响或者是独立复用它们时。
• 当一个变化的对象通知那些与它自身变化相关联的未知数量的对象时。
• 当一个变化的对象通知那些无需推断具体是谁的对象时。

二、Observable

在ReactiveX中，一个观察者(Observer)订阅一个可观察对象(Observable)。观察者对Observable发射的数据或数据序列作出响应。这种模式可以极大地简化并发操作，因为它创建了一个处于待命状态的观察者哨兵，在未来某个时刻响应Observable的通知，不需要阻塞等待Observable发射数据。

2.1 背景知识

在很多软件编程任务中，或多或少你都会期望你写的代码能按照编写的顺序，一次一个的顺序执行和完成。但是在ReactiveX中，很多指令可能是并行执行的，之后他们的执行结果才会被观察者捕获，顺序是不确定的。为达到这个目的，你定义一种获取和变换数据的机制，而不是调用一个方法。在这种机制下，存在一个可观察对象(Observable)，观察者(Observer)订阅(Subscribe)它，当数据就绪时，之前定义的机制就会分发数据给一直处于等待状态的观察者哨兵。

这种方法的优点是，如果你有大量的任务要处理，它们互相之间没有依赖关系。你可以同时开始执行它们，不用等待一个完成再开始下一个（用这种方式，你的整个任务队列能耗费的最长时间，不会超过任务里最耗时的那个）。

有很多术语可用于描述这种异步编程和设计模式，在在本文里我们使用这些术语：一个观察者订阅一个可观察对象 (An observer subscribes to an Observable)。通过调用观察者的方法，Observable发射数据或通知给它的观察者。

在其它的文档和场景里，有时我们也将Observer叫做Subscriber、Watcher、Reactor。这个模型通常被称作Reactor模式。

2.2 创建观察者

本文使用Kotlin代码举例。

普通的方法调用流程通常是这样的：

    val key = "暴走"
    try {
        val list = query(key)
        list.forEach { println(it) }
    } catch (e: Exception) {
        e.printStackTrace()
    }

很明显，使用这种方式进行查询时是阻塞的。在查询期间你的程序不能再做任何其他的事情。

在异步模型中流程更像这样的：

1. 定义一个方法，它完成某些任务，然后从异步调用中返回一个值，这个方法是观察者的一部分
2. 将这个异步调用本身定义为一个Observable
3. 观察者通过订阅(Subscribe)操作关联到那个Observable
4. 继续你的业务逻辑，等方法返回时，Observable会发射结果，观察者的方法会开始处理结果或结果集

2.3 回调方法

Subscribe方法用于将观察者连接到Observable，你的观察者需要实现以下方法的一个子集：

• onNext(T item)
  Observable调用这个方法发射数据，方法的参数就是Observable发射的数据，这个方法可能会被调用多次，取决于你的实现。

• onError(Exception ex)
  当Observable遇到错误或者无法返回期望的数据时会调用这个方法，这个调用会终止Observable，后续不会再调用onNext和onCompleted，onError方法的参数是抛出的异常。

• onComplete
  正常终止，如果没有遇到错误，Observable在最后一次调用onNext之后调用此方法。

我们请求数据的函数就可以写成这个样子：

    val onNext: (item: News) -> Unit = { println(it) }
    val onError: (e: Exception) -> Unit = { it.printStackTrace() }
    val onComplete: (Unit) -> Unit = {}
    observable("暴走")
    subscribe(onNext, onError, onComplete)

2.4 取消订阅 (Unsubscribing)

在一些ReactiveX实现中，有一个特殊的观察者接口Subscriber，它有一个unsubscribe方法。调用这个方法表示你不关心当前订阅的Observable了，因此Observable可以选择停止发射新的数据项（如果没有其它观察者订阅）。

取消订阅的结果会传递给这个Observable的操作符链，而且会导致这个链条上的每个环节都停止发射数据项。这些并不保证会立即发生，然而，对一个Observable来说，即使没有观察者了，它也可以在一个while循环中继续生成并尝试发射数据项。

2.5 Observables的"热"和"冷"

一个"热"的Observable可能一创建完就开始发射数据，因此所有后续订阅它的观察者可能从序列中间的某个位置开始接受数据（有一些数据错过了）。一个"冷"的Observable会一直等待，直到有观察者订阅它才开始发射数据，因此这个观察者可以确保会收到整个数据序列。

2.6 Single

RxJava中有一个名为Single的Observable变种。

Single类似于Observable，不同的是，它总是只发射一个值，或者一个错误通知，而不是发射一系列的值。

因此，不同于Observable需要三个方法onNext, onError, onCompleted，订阅Single只需要两个方法：

• onSuccess - Single发射单个的值到这个方法
• onError - 如果无法发射需要的值，Single发射一个Throwable对象到这个方法

Single只会调用这两个方法中的一个，而且只会调用一次，调用了任何一个方法之后，订阅关系终止。

Single也可以组合使用多种操作，一些操作符让你可以混合使用Observable和Single：

[1]RxJava Essentials CN
[2]ReactiveX文档