掌握RxJava的葵花宝典

Posted 2022-12-12 刘某人程序员

本文授权发布公众号【刘桂林】，星球【Hi android】

各位少侠，老夫在黑木崖恭候大驾，欲练此功，必先哈哈。

今天我们分享的是RxJava的知识点，让你快速掌握，所以我们会从0开始讲带RxJava的Api使用和设计理念等，本文较长，所以需要耐心看下去。

目录

• 一.RxJava相关资料
• 二.RxJava的基本使用
• 三.RxJava的调度器
• 四.RxJava的操作符
• 五.RxJava的订阅与反订阅
• 六.RxJava的背压支持
• 七.RxJava结合OkHttp
• 八.RxJava结合Retrofit

我们基于截止到本篇最新的RxJava版本 [ 2.2.8 ] 来讲解，事实上RxJava2.0 对比 RxJava1.0来说，更新了很多的东西，这些下文都将说到，所以这里提示你一下，RxJava 2.0 和 1.0 Api是不兼容的，同时也存在很多坑，如果你的老代码使用了RxJava1.0,你现在想要升级为RxJava2.0的话，是需要大面积的去修改的，好了，说了这么多，我们就不再介绍RxJava到底是什么了，因为这根框架出来了挺久了，相信大家都耳熟能详了，所以我尽可能的对设计理念，逻辑，对象，方法，接口等方面下手，让大家都有一个清晰的认识，做到知其然也知所以然，好了，我们直接进入正题吧。

一.RxJava相关资料
资料是肯定要的，学习也好，温习也罢，都是不可或缺的。

Github：https://github.com/ReactiveX/RxJava
RxJava 中文文档：https://mcxiaoke.gitbooks.io/rxdocs/content/

二.RxJava的基本使用
app/build.gradle中引入

implementation "io.reactivex.rxjava2:rxjava:2.2.8"

如果想充分了解RxJava的话，我建议你通读几遍RxJava中文文档，或许就已经得心应手了，我先来教大家RxJava的一些基础概念和用法

首先，我们要明白观察者模式，这个模式我就不长篇大论了，给大家讲一个故事：小明在复习功课，小明的妹妹则是在看一个电视节目叫做我是歌手，小明告诉妹妹要是出现了周杰伦就叫他，这就有点类似观察者模式了，其中小明是观察者，妹妹是被观察者，通过叮嘱建立关系，叫做订阅，我们不需要小明隔一会儿就问一下妹妹周杰伦上场了没，而是订阅（叮嘱）后，妹妹看到周杰伦主动告诉小明，这个例子与抛物线的有区别，他举例的小偷并不会主动告诉警察自己在犯罪，在RxJava中，我们有三个阶段，分别如下：

我们围绕这段代码来讲解，首先我要告诉你，代码看起来有些长，其实RxJava是很简洁的，有很多简洁的写法，这是一段初学者的代码，我先解释一下流程：小明（Observer）叮嘱（subscribe）妹妹（Observable）当出现周杰伦的时候叫他（onNext）

Observer就是小明，他是一个接口，里面有四个回调方法，首先是第一个onSubscribe，这个方法是当你订阅成功之后回调的，里面传了一个Disposable，这个对象可以追述到RxJava1.0的时候的东西，可以不讲，这个对象可以用来判断订阅关系和解除订阅关系，也就是我在上面代码块中onComplete里的操作，继续来看，onNext方法就是与Observable，也就是被观察者通信用的，当被观察者有什么数据就是通过他来传输的，onError顾名思义就是发生错误的回调，还有onComplete，完成的回调， onNext，onError，onComplete，这三个都是通过被观察者Observable回调中的ObservableEmitter调用的，称为事件发射器。

Observable就是妹妹了，也是被观察者，Observable和Observer容易被混淆，这点一定要分清，至少我以前经常把他们搞混，我们来看他里面的代码，里面我写死的数据判断周杰伦，这个时候，妹妹判断到周杰伦就通过onNext()叫小明过来，然后再调用onComplete() 结束自己的任务，到此，观察者和被观察者的任务就完成了

接着，我们再来说一个特性，叫做链式调用，用在我们的创建对象上也是可以做到简化的效果的，看如下代码：

这里只是为了告诉你链式调用这个概念，因为后面会经常用到，所以在这里先抛砖引玉，同时你可能有些疑问，为什么订阅的代码是Observable订阅Observer，即被观察者订阅观察者，形象一点就是 报纸订阅人，这明显是有些奇怪的，难道不应该是人订阅报纸，观察者订阅被观察者吗？实际上这是跟他的设计理念有些关系，这里给大家解惑一下，就是佛系，不要太在意。

接着，我们就要来继续简化这段操作了，在RxJava1中的onAction,onFunc等方法已经被修改了，所以你现在应该记住如下的快捷方法：

• Consumer 即观察者，用于接收单个值，
• BiConsumer 则是接收两个值，
• Function 用于变换对象，
• Predicate 用于判断。

记住这四个方法，我们一一来讲解一下：

Consumer和BiConsumer，其实就是一个简化的Observer，区分于Consumer接收单个值，而BiConsumer 则接收两个值，我们来看如下用法：

这里的代码使用Consumer完全替代了Observer，也能接收到onNext传输的数据，这就做到了简化的代码了，从最开始的一大段代码到现在的这些代码，不过这个还可以继续优化，但是目前你还是不知道为好，我们继续往下看，由于BiConsumer 是双参传入，我们到后面再来补充使用方式，现在讲需要用到其他函数，不利于阅读性，至少到目前为止，你已经知道了观察者和被观察者建立订阅传输数据，其实这已经足够了，我写着篇文章就是慢慢引导，最终掌握技能。

我们接着来看Function ，Function 是一个变换对象的函数，如果你有一个Drawable需要转换成Bitmap或者你有一个String需要转Int都可以使用到他，不过我们这个例子用到了just和map,这两个你暂时可以不用理解，你只要明白这段代码的含义就好了：

我们看到代码中首先Observable.just里传入了一个字符串，但是我们的接口可能需要int，那么就需要对他进行转换了，通过map的Function把字符串1传入回调函数进行处理后再return回来给到Observer中的Consumer 处理，在Consumer 中进行了相加的运算得到的结果为2，说明一切正常

接着来看下Predicate ，这个方法是用来做一些条件判断并且拦截事件的，我们来看下如下这一段代码：

从这段代码中我们看到，首先我们通过create创建被观察者，他发送出来的数据是3，然后通过filter方法实现了一个Predicate接口，filter方法你可以暂时不要理会，我们先来明白Predicate接口，可以看到他的返回值是一个布尔类型，这里我判断是否大于5，而我们传入的数据是3，小于5，那么这里返回的应该是false,接着我们实现Consumer，用来打印，这里返回false,那么事件就会被拦截，Consumer就不会打印，如果Predicate返回true,那么事件正常执行。

好了，读到这里，我相信你已经随手能写出一段基础RxJava代码了，我们接着要来理解一个新的概念，那就是调度器。

三.RxJava的调度器
调度器 Scheduler的作用很多，我们这里来说一下几个特性，其中关于Android部分的我们后面补充，先单指RxJava中的线程调度器：

Schedulers.computation()
用于计算任务，用于事件循环或和回调处理，不要用于IO操作，默认线程数等同于处理器的数量

Schedulers.from(executor)
使用指定的Executor作为调度器

Schedulers.immediate( )
在当前线程立即开始执行任务，这也是默认的调度器

Schedulers.io( )
用于IO操作的调度器，这个调度器的线程池会根据需要增长，很像一个CachedThreadScheduler线程池

Schedulers.newThread( )
每个任务都开启一个新的线程

Schedulers.trampoline( )
当其它排队的任务完成后，在当前线程排队开始执行

这里同时需要明白两个概念，那就是subscribeOn和observeOn，前者是指定工作线程，后一个是指定回调线程，我们来看如下代码：

这里我用subscribeOn指定了Schedulers.newThread()这个调度器，意思就是Observable在工作的时候每次执行任务都创建一个新的线程，比如我在里面处理了一个耗时的任务，这个时候是处于子线程中，我们想要Observer中直接更新UI，那么应该怎么做？这里这一句就是关键：observeOn(AndroidSchedulers.mainThread())，指定线程为主线程，AndroidSchedulers这个调度器是RxAndroid中的，所以一般我们是配合使用的，一法通万法，只要你学会了RxJava，其实RxAndroid很多东西也能迎刃而解的，所以我们总结一下：subscribeOn是用来指定工作线程，而observeOn是用来指定回调线程。

RxAndroid http://github.com/ReactiveX/RxAndroid

app/build.gradle:

implementation 'io.reactivex.rxjava2:rxandroid:2.1.1'

代码中多次提到了一些操作符如create，map等，每一个都有特殊的含义与作用，同时Rx真正的强大也在于这些操作符，所以我们接下来花大篇幅讲解一下这些操作符的作用。

四.RxJava的操作符
RxJava的操作符还是很有魅力的，在上面几个章节中，我们提到了create，just，map，filter 这些都是它的操作符，操作符一共分为十种，分别如下，下文摘抄自中文文档：

创建操作符 / 用于创建Observable的操作符
Create — 通过调用观察者的方法从头创建一个Observable
Defer — 在观察者订阅之前不创建这个Observable，为每一个观察者创建一个新的Observable
Empty/Never/Throw — 创建行为受限的特殊Observable
From — 将其它的对象或数据结构转换为Observable
Interval — 创建一个定时发射整数序列的Observable
Just — 将对象或者对象集合转换为一个会发射这些对象的Observable
Range — 创建发射指定范围的整数序列的Observable
Repeat — 创建重复发射特定的数据或数据序列的Observable
Start — 创建发射一个函数的返回值的Observable
Timer — 创建在一个指定的延迟之后发射单个数据的Observable

变换操作符 / 用于对Observable发射的数据进行变换
Buffer — 缓存，可以简单的理解为缓存，它定期从Observable收集数据到一个集合，然后把这些数据集合打包发射，而不是一次发射一个
FlatMap — 扁平映射，将Observable发射的数据变换为Observables集合，然后将这些Observable发射的数据平坦化的放进一个单独的Observable，可以认为是一个将嵌套的数据结构展开的过程。
GroupBy — 分组，将原来的Observable分拆为Observable集合，将原始Observable发射的数据按Key分组，每一个Observable发射一组不同的数据
Map — 映射，通过对序列的每一项都应用一个函数变换Observable发射的数据，实质是对序列中的每一项执行一个函数，函数的参数就是这个数据项
Scan — 扫描，对Observable发射的每一项数据应用一个函数，然后按顺序依次发射这些值
Window — 窗口，定期将来自Observable的数据分拆成一些Observable窗口，然后发射这些窗口，而不是每次发射一项。类似于Buffer，但Buffer发射的是数据，Window发射的是Observable，每一个Observable发射原始Observable的数据的一个子集

过滤操作符 / 用于从Observable发射的数据中进行选择
Debounce — 只有在空闲了一段时间后才发射数据，通俗的说，就是如果一段时间没有操作，就执行一次操作
Distinct — 去重，过滤掉重复数据项
ElementAt — 取值，取特定位置的数据项
Filter — 过滤，过滤掉没有通过谓词测试的数据项，只发射通过测试的
First — 首项，只发射满足条件的第一条数据
IgnoreElements — 忽略所有的数据，只保留终止通知(onError或onCompleted)
Last — 末项，只发射最后一条数据
Sample — 取样，定期发射最新的数据，等于是数据抽样，有的实现里叫ThrottleFirst
Skip — 跳过前面的若干项数据
SkipLast — 跳过后面的若干项数据
Take — 只保留前面的若干项数据
TakeLast — 只保留后面的若干项数据

组合操作符 / 用于将多个Observable组合成一个单一的Observable
And/Then/When — 通过模式(And条件)和计划(Then次序)组合两个或多个Observable发射的数据集
CombineLatest — 当两个Observables中的任何一个发射了一个数据时，通过一个指定的函数组合每个Observable发射的最新数据（一共两个数据），然后发射这个函数的结果
Join — 无论何时，如果一个Observable发射了一个数据项，只要在另一个Observable发射的数据项定义的时间窗口内，就将两个Observable发射的数据合并发射
Merge — 将两个Observable发射的数据组合并成一个
Switch — 将一个发射Observable序列的Observable转换为这样一个Observable：它逐个发射那些Observable最近发射的数据
Zip — 打包，使用一个指定的函数将多个Observable发射的数据组合在一起，然后将这个函数的结果作为单项数据发射
错误处理操作符 / 用于从错误通知中恢复
Catch — 捕获，继续序列操作，将错误替换为正常的数据，从onError通知中恢复
Retry — 重试，如果Observable发射了一个错误通知，重新订阅它，期待它正常终止

辅助操作符 / 用于处理Observable的操作符
Delay — 延迟一段时间发射结果数据
Do — 注册一个动作占用一些Observable的生命周期事件，相当于Mock某个操作
Materialize/Dematerialize — 将发射的数据和通知都当做数据发射，或者反过来
ObserveOn — 指定观察者观察Observable的调度程序（工作线程）
Serialize — 强制Observable按次序发射数据并且功能是有效的
Subscribe — 收到Observable发射的数据和通知后执行的操作
SubscribeOn — 指定Observable应该在哪个调度程序上执行
TimeInterval — 将一个Observable转换为发射两个数据之间所耗费时间的Observable
Timeout — 添加超时机制，如果过了指定的一段时间没有发射数据，就发射一个错误通知
Timestamp — 给Observable发射的每个数据项添加一个时间戳
Using — 创建一个只在Observable的生命周期内存在的一次性资源

条件和布尔操作符 / 用于单个或多个数据项，也可用于Observable
All — 判断Observable发射的所有的数据项是否都满足某个条件
Amb — 给定多个Observable，只让第一个发射数据的Observable发射全部数据
Contains — 判断Observable是否会发射一个指定的数据项
DefaultIfEmpty — 发射来自原始Observable的数据，如果原始Observable没有发射数据，就发射一个默认数据
SequenceEqual — 判断两个Observable是否按相同的数据序列
SkipUntil — 丢弃原始Observable发射的数据，直到第二个Observable发射了一个数据，然后发射原始Observable的剩余数据
SkipWhile — 丢弃原始Observable发射的数据，直到一个特定的条件为假，然后发射原始Observable剩余的数据
TakeUntil — 发射来自原始Observable的数据，直到第二个Observable发射了一个数据或一个通知
TakeWhile — 发射原始Observable的数据，直到一个特定的条件为真，然后跳过剩余的数据

算数和聚合操作符 / 用于整个数据序列
Average — 计算Observable发射的数据序列的平均值，然后发射这个结果
Concat — 不交错的连接多个Observable的数据
Count — 计算Observable发射的数据个数，然后发射这个结果
Max — 计算并发射数据序列的最大值
Min — 计算并发射数据序列的最小值
Reduce — 按顺序对数据序列的每一个应用某个函数，然后返回这个值
Sum — 计算并发射数据序列的和

连接操作符 / 有精确可控的订阅行为的特殊Observable
Connect — 指示一个可连接的Observable开始发射数据给订阅者
Publish — 将一个普通的Observable转换为可连接的
RefCount — 使一个可连接的Observable表现得像一个普通的Observable
Replay — 确保所有的观察者收到同样的数据序列，即使他们在Observable开始发射数据之后才订阅

转换操作符 / 用于转换成其他对象
To — 将Observable转换为其它的对象或数据结构
Blocking 阻塞Observable的操作符
可以看到其中的操作符非常的多，这也意味着使用起来更加灵活，我们篇幅有限，也不可能全部讲一遍，不过我们可以挑一些重点的来讲，这里我也尽量保证多赘述一些，让大家尽可能的理解，不过要是太啰嗦了，也就没得办法了，只能说，要想学会RxJava，这些都是必经之路的。

创建操作符
1.Create
这应该是我们第一个学习的操作符了，create操作符的作用是从头开始创建一个Observable,并且回调一个事件发射器ObservableEmitter用来调用onNext，onError,onCompleted,一个正确的Observable应该正好调用一次onCompleted或者onError，我们来看这段示例代码：

实际上这段代码我们看了很多遍了，Observable创建后订阅一个Observer

2.Defer
defe和create的区别在于只有等关系订阅之后Observable才会去建立，相当于懒加载，我们来看下示例代码：

也就是说当订阅关系建立的时候，才会return一个Observable出去，这就做到了只有等关系订阅之后Observable才会去建立的特性了，只要能创建一个Observable，不管是just还是create都是可以的，如下：

3.Empty/Never/Throw
这三个放在一起讲师应该这三个比较特殊

Empty：创建一个不发射任何数据但是正常终止的Observable
Never：创建一个不发射数据也不终止的Observable
Throw：创建一个不发射数据以一个错误终止的Observable

可以说这三个一般是用于测试或者传空参的时候才会用到。

4.From / FromX
这个操作符使用比较常见，他的含义是将其它种类的对象和数据类型转换为Observable，不过在RxJava2.0中，已经看不到from的身影了，取而代之的是如下操作符：

我们拿fromArray来举例子，这里将一个数组转换成observable，然后由observer来一个个分析，相当于将这些数据遍历后发射给观察者，看下面的示例代码：

打印的数据也是将数组中的元素依次打印。

5.Interval
这个操作的意识是定时任务，可以用来做心跳包之类的需求，我们来看下如下的代码：

这里有必要说明一下，我们interval里面传了三个参数，第一个为延时时长，第二个为循环时长，第三个是时间单位，我们这里延时时长传了5，循环时长传了2，也就是说，建立任务5s后开启循环，然后每2s发射一个无限递增的整数序列。

当然，你也可以不传第一个参数，也就是这个5，直接启动相隔2s的心跳任务。

6.Just
just的出场率也是很高的，这也是源于他的特性，我们看如下的代码段：

这里输出的是这个数组，just和from类似，不过from会解析传入的参数，而just只是单纯的发射而已，如果上面的代码用from来发射，那将打印 1 3 5 7 9 。

7.Range
range是用来发射一个有序的整数列，他需要传入两个参数，一个起点，一个数目，看示例代码：

我传入了两个参数 5 和 10，代表着从5开始有序的发射10个，打印的结果为：5 6 7 8 9 10 11 12 13 14 。

8.Repeat
repeat并不是创建Observable的，而是重复发射原始Observable的，我们来看例子：

这个例子很简单，通过range创建Observable来发射一个从5开始，数量为3的数据，并且重复2此，也就是发射： 5 6 7 5 6 7

9.Start / StartXXX
在RxJava2.0中，已经没有start操作符了，取而代之的是以start开头的操作符，他和repeat一样并不是一个创建Observable的操作符，说白点，他是为你所发射的Observable前面加点数据，看示例代码：

这里打印出来的数据为： 3 1

10.Timer
timer顾名思义和时间有关，实际上他和interval有点类似，time传两个参数，一个是延时时长，一个是单位，看我的示例代码图片:

延时两秒后调用onNext，不过要做到interval也是可以的，只要加上一个repeat就可以实现每2s发射一次的效果了，所以这两个操作符有点重复。

变换操作符
1.Buffer
这个操作符有些复杂，顾名思义就是缓冲区的意思，你如果明白他的意思你就不难理解了，所以下面这段示例代码你要仔细看下：

这里我使用了just发送五个字符串，并且设置了buffer为3和1，第一个参数为缓存长度，这二个为向后位移距离，要明白是什么意思，你得先看下打印：

分析：正常打印应该是打印：One,Two,Three,Four,Five,但是由于我设置了buffer，所以打印就变成了了：
（One,Two,Three）,Four,Five
One,（Two,Three,Four）,Five
One,Two,（Three,Four,Five）
One,Two,Three,（Four,Five）
One,Two,Three,Four,（Five）
也就是说每次取3位，并且每一次后移一位，就得到了如此的结果了，如果我们将buffer改成2,2，那将如何打印呢？
(One,Two),Three,Four,Five
One,Two,(Three,Four),Five
One,Two,Three,Four,(Five)
现在就好理解了，每次取2位，并且后移两位，就得到了如上的打印。

注意：如果原来的Observable发射了一个onError通知，Buffer会立即传递这个通知，而不是首先发射缓存的数据，即使在这之前缓存中包含了原始Observable发射的数据。

2.FlatMap
FlatMap将一个发射数据的Observable变换为多个Observables，然后将它们发射的数据合并后放进一个单独的Observable,这是我们介绍map家族的第一个操作符，来看下他的使用：

这段代码其实不长，只是我加了一个实体类而已，我们可以看到，我在创建Observable后通过事件发射器发射了三个学生，然后再flatMap中判断这个学生的分数是否大于90分，如果大于，，则isGood设置为true，否则反之，然后通过just重新创建一个Observable返回回去，最后在Consumer中处理打印，起到了一层转换过滤的作用，这也就明白了上面的含义，将一个Observable转换成多个自己想要的Observable。

FlatMap操作符需要多用，理解。

3.GroupBy
groupby这个操作符会进行分拆，然后发射一个原始Observable的序列，看如下例子：

在这个例子中，我使用range创建了0 - 10的数据发射出去后通过groupby来分类，然后groupby就会发送一个带有标签的原始序列到observer了，我们只要取出来就可以用了，打印如下：

4.Map
map我们很早就用到了，map是对每一项数据都进行变换操作,在之前的案例其实就有提过：

从Int转为String了，没什么好讲的

5.Scan
连续地对数据序列的每一项应用一个函数，然后连续发射结果,我们来看下示例：

看这段代码，我发射了 0 -10的整数，探后在BiFuntion中有两个参数，分别是 integer和integer2，事实上在这个回调方法中时用来处理发射的数据的，integer2就是依次发射过来的数据了，也就是说每次 integer + integer2 return回去就是赋值在 integer上，也就起到了累加的效果，所以打印出来是 45。

6.Window
window的用法和buffer有一些类似，定期将来自原始Observable的数据分解为一个Observable窗口，发射这些窗口，而不是每次发射一项数据，来看示例代码：

在这段代码中我们可以看到window我传了一个4，那是什么意思呢？其实只要看到打印，你就知道意思了：

每四个元素为一个窗口发射出去，为什么说他跟buffer很像呢，因为他也是可以做到向后位移距离的，这个就自行探究吧。

过滤操作符

1.Debounce
Debounce操作符会过滤掉发射速率过快的数据项。也可以理解为防抖，在约定时间内不响应，比如你设置了1s,按钮重复点击也可以用此方法来处理，.debounce(1,TimeUnit.SECONDS)，这里就不多讲了。

2.Distinct
这个操作符代表着只发射没有发射过的数据，相当于去重复

打印的结果就是： 1 2 3

3.ElementAt
只发射第N项数据

这些操作符都很简单，这里是只发射第二个数据，下标从0开始，也就是3

4.Filter
这个在讲Predicate的时候其实就已经用到了，只有通过的数据才能发射：

5.First
只发射第一项（或者满足某个条件的第一项）数据

6.IgnoreElements
不发射任何数据，只发射Observable的终止通知

7.Last
只发射最后一项（或者满足某个条件的最后一项）数据

8.Sample
定期发射Observable最近发射的数据项

9.Skip
抑制Observable发射的前N项数据,也可以理解为跳过，看如下示例代码：

这里skip设置为4，也就是跳过4个参数，输入： 2 和 3

10.SkipLast
抑制Observable发射的后N项数据，这个就和skip反过来了，从后面开始跳过

11.Take
只发射前面的N项数据，如果设置为1，那效果就和操作符first一致了

12.TakeLast
只发射后面的N项数据，如果设置为1，那效果就和操作符last一致了

结合操作符
1.And/Then/When
使用Pattern和Plan作为中介，将两个或多个Observable发射的数据集合并到一起,组合的行为类似于zip，但是它们使用一个中间数据结构。接受两个或多个Observable，一次一个将它们的发射物合并到Pattern对象，然后操作那个Pattern对象，变换为一个Plan。随后将这些Plan变换为Observable的发射物。

不过他们是rxjava-joins里的模块

2.CombineLatest
当两个Observables中的任何一个发射了数据时，使用一个函数结合每个Observable发射的最近数据项，并且基于这个函数的结果发射数据。

理解起来可能有点拗口，我们来看示例代码：

从这里不难看出创建了三个Observable，当这三个Observable都发射了数据之后CombineLatest才会发射一个数据，也就起到组合效果。

3.Join
任何时候，只要在另一个Observable发射的数据定义的时间窗口内，这个Observable发射了一条数据，就结合两个Observable发射的数据。这样说起来有些笼统，可以举例，A Observable 正在顺序发送 1 ， 2 ， 3 ， 4 ，这个时间B Observable 通过join 发送了一个 Hello 字符串，则组合为 1 Hello , 2 Hello , 3 Hello , 4 Hello .

4.Merge
合并多个Observables的发射物

这段代码打印出来的结果是： 1 2 3 4 5 6 ，已经结合了

5.Switch
将一个发射多个Observables的Observable转换成另一个单独的Observable，后者发射那些Observables最近发射的数据项
6.Zip
通过一个函数将多个Observables的发射物结合到一起，基于这个函数的结果为每个结合体发射单个数据项。

其实就是可操作的两个对象，可以用来求相加值，最大值等

这里我是将两个相加，得到的结果是： 5 ， 7 ， 9

错误处理操作符
1.Catch
从onError通知中恢复发射数据

2.Retry
如果原始Observable遇到错误，重新订阅它期望它能正常终止

辅助操作符

1.Delay
延迟一段指定的时间再发射来自Observable的发射物,就是一个延时发送的操作符，可以使用.delay(3,TimeUnit.SECONDS)来处理

2.Do / DoXXX
注册一个动作作为原始Observable生命周期事件的一种占位符,他提供了很多监听回调

我们在回调中就能知道对应事件，类似回调。

3.Materialize/Dematerialize
Materialize将数据项和事件通知都当做数据项发射，Dematerialize刚好相反。

一个合法的有限的Obversable将调用它的观察者的onNext方法零次或多次，然后调用观察者的onCompleted或onError正好一次。Materialize操作符将这一系列调用，包括原来的onNext通知和终止通知onCompleted或onError都转换为一个Observable发射的数据序列。

4.ObserveOn
指定一个观察者在哪个调度器上观察这个Observable

5.Serialize
强制一个Observable连续调用并保证行为正确，怎么说呢，之前也说过，一个正确的逻辑应该是只有一个onCompleted或者onError,如果你违反这个规定可能会导致一些其他错误，这时你可以用Serialize

6.SubscribeOn
指定Observable自身在哪个调度器上执行

这些很多大家可以自己去看吧，写得太多了，手快断了

五.RxJava的订阅与反订阅
我们的使用三部曲就使用到了订阅，也就是我们本文一开头小明叮嘱妹妹，叮嘱的这个动作，通过Observable的subscribe就可以订阅一个Observer了，但是有时候，我们有必要的时候必须要取消这个订阅，取消订阅通过Disposable对象，这里分两个情况，如果直接使用完，那么可以直接在observer结尾调用dispose即可，如果需要持续使用，那么这个时候就有必要注意内存泄漏这个问题了，我们可以在Activity的onDestroy中去取消订阅。

六.RxJava的背压支持
背压是什么呢？如果Observer和Observer处在不同的线程中，那么此处存在数据同步的问题，如果Observable发射的数据过快，Observer都处理不过来了，此时的数据会被存放在一个特定的内存当中，叫做异步缓存块，可想而知，当程序运行一段时间后，这个异步缓存块越来越大，最终突破了App的最大内存，导致内存溢出，在RxJava2.0中，出现了一个Flowable的对象，他就是用来处理此问题的，我们先来模拟一段背压代码：

从UI上看因为这里一直Sleep所以只会有卡主的现象，但是如果你看下你的内存情况，我想你会崩溃的

那我们如何解决此问题，来看如下代码：

这块代码已经被改动了，我们使用Flowable的create创建，里面需要传入一个BackpressureStrategy.BUFFER，事实上他的策略模式有MISSING，ERROR，BUFFER，DROP，LATEST五种，每一种都有特殊的含义，并且在onSubscribe中默认设置数据量为1，通过onNext去进行拉取形成互相呼应。

• MISSING：Observable不处理，由Observer处理背压模式
• ERROR：超过缓存块最大限制 128 就会抛出异常
• BUFFER：默认的模式，超出最大限制会有转而一个新的队列中，不过再多就会OOM
• DROP：超过缓存块最大限制就会抛弃发射过来的数据
• LATEST：和DROP一样，不过会把最新的数据放置缓存块最后

我们再次运行：

内存就不会出现溢出的现象了。

七.RxJava结合OkHttp
这里结合OkHttp来讲解一下如何使用RxJava，详细代码请阅读Sampe中的RxJavaOkHttpActivity.java

首先是在app/build.gradle中引入OkHttp

implementation "com.squareup.okhttp3:okhttp:3.14.1"

然后开始编写OkHttp请求Url的代码了：

这段代码就是OkHttp基础的使用方法了，我们创建一个Request后，通过OkHttpClient去new一个Call，通过Call去同步请求获得一个Response，而我们的请求结果就在这个Response中，接下来就看我们RxJava的运用了：

在这段代码中，我通过发射器将请求得到的结果发射给Observer，就可以显示出结果了，异常的简单，你可能会问，我直接调用disposeUrlResult不就可以获得数据了？那你就错了，直接调用的代码如下：

mTextView.setText(disposeUrlResult(Url));

如果直接调用disposeUrlResult将会报如下的错误：

你可以注意到是一个NetworkOnMainThreadException的错误，也就是说不能在主线程访问网络，好吧，那你又说了，那还不简单，我加个Thread不就好了，代码如下：

你再次运行一遍，你得意洋洋，然后发现，呀，还是报错了

可以看到错误是：Only the original thread that created a view hierarchy can touch its views.只能在主线程更新UI，好了，你又要说了，那我用Handler更新UI总可以了吧，那好，你的代码如下：

这样确实可以，但是你发现了吗，你已经弯弯绕绕很长了，而且这还只是一个简单的请求，而在RxJava中只需要subscribeOn和observeOn就好了，省去了你new Thread 和 new Handler这个过程，这下孰轻孰重，自行判断了，而且链式调用，逻辑分明，这才是我们所需要的，在大型的应用场景中，它的优势会越来越明显。最后看一下请求结果，Json你们就自行解析了。

八.RxJava结合Retrofit
首先我们引入Retrofit的依赖吧，这里要注意的是RxJava2 和 Retrofit2 是有冲突的，所以我们引入的依赖是经过了修复的：

implementation 'com.squareup.retrofit2:adapter-rxjava2:2.5.0'

好的，这次我们以请求天气为例子，我们先来看下传统的写法和RxJava的写法有什么区别：

至于Retrofit的基本用法，我相信大家应该是会用的，如果不会用也没有关系，我之后的文章也会详细介绍的，好的，我们来看这个接口类，传统的请求和RxJava的请求写法，其实就返回值不一样，传统的我返回了一个Call对象回去作为请求体，而Rx方式我是直接返回一个Observable，紧接着初始化Retrofit，这个可以一笔带过了：

我们来对比一下两个请求的区别：

首先来看标准的请求，很显然，拿到call之后进行一步的请求然后得到数据，再来看RxJava的请求方式：直接在Observer中处理数据，至于线程的调度，这个还是一贯的优势，到这里你也应该明白如何使用RxJava了，详细代码请阅读Sampe中的RxJavaRetrofitActivity.java

好了，到这里就讲完了，实际上这只是教会了你如何使用，但是还并没有深入的去理解RxJava，后面的文章会把这些落下的知识点补回来，比如自定义操作符，几大核心接口的用法，包括源码的思路和解析，好的，再会！

有兴趣可以加入我的星球：Hi Android , 里面可都是我手撸的新鲜文章，高质量你值得拥有！

进入星球你可以做什么？
1.我的所有视频可以观看
2.发布提问贴可以得到满意的答案
3.可指定我写你感兴趣的技术文章
4.初学者可配套视频辅导
5.有机会线下交流聚会