Android基础（39）RxJava

创建操作：

1. create 通过调用观察者的方法从头调用一个Observable
2. just 将对象或对象集合转换为一个会发射这些对象的Observable
3. interval 创建一个定时发射的整数序列的Observable
4. from 将其他的对象或数据结构转化为Observable
5. defer 在观察者订阅之前不创建这个Observable，为每个观察者创建一个新的Observable
6. range 创建发射指定范围的整数序列的Observable
7. repeat 创建重复发射特定的数据或数据序列的Observable
8. start 创建发射一个函数的返回值的Observable
9. timer 创建在一个指定的延迟之后发射单个数据的Observable

变换操作：

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

过滤操作：

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

组合操作：

用于将多个Observable组合成一个Observable

1. And/Then/When 通过模式（And条件）和计划（Then次序）组合两个或多个Observable发射的数据集
2. CombineLatest 当两个Observable中的任何一个发射了一个数据时，通过一个指定的函数组合每个Observable发射的最新数据（一共两个数据），然后发射这个函数的结果。
3. Join 无论何时，如果一个Observable发射了一个数据项，只要在另一个Observable发射的数据项定义的时间窗口内，就将两个Observable发射的数据合并发射。
4. Merge 将两个Observable发射的数据组合并成一个。
5. StartWith 在发射原来的Observable的数据序列之前，先发射一个指定的数据序列或数据项。
6. Switch 将一个发射Observable序列的Observable转换为这样一个Observable：它逐个发射哪些Observable最近发射的数据。
7. Zip 打包，使用一个指定的函数将多个Observable发射的数据组合在一起，然后将这个函数的结果作死你单项数据发射

错误操作：

用于从错误通知中恢复

1. Catch 捕获，继续序列操作，将错误替换为正常的数据，从onError通知中恢复。
2. Retry 重试，如果Observable发射了一个错误通知，重新订阅它，期待它正常终止

辅助操作：

用于处理Observable的操作符

1. Delay 延迟一段时间发射结果数据
2. Do 注册一个动作占用一些Observable的生命周期事件，相当于Mock某个动作
3. Materialize/Dematerialize 将发射的数据和通知都当作数据发射，或者反过来
4. ObserveOn 指定观察者观察Observable的调度程序（工作线程）
5. Serialize 强制Observable按次序发射数据并且功能是有效的。
6. Subscribe 受到Observable发射的数据和通知后执行的操作
7. SubscribeOn 指定Observable应该在哪个调度程序上执行
8. TimeInterval 将一个Observable转换为发射两个数据之间所耗费时间的

Observable

9. Timeout 添加超时机制，如果过了指定的一段时间没有发射数据，就发射一个错误通知
10. Timestamp 给Observable发射的每个数据项添加一个事件戳
11. Using 创建一个只在Observable的生命周期内存在的一次性资源

条件和布尔操作：

用于单个或多个数据项，也可用于Observable

1. All 判断Observable发射的所有的数据项是否都满足某个条件
2. Amb 给定多个Observable，只让第一个发射数据的Observable发射全部数据
3. Contains 判断Observable是否会发射一个特定的数据项
4. DefaultIfEmpty 发射来自原始Observable的数据，如果原始Observable没有发射数据，就发射一个默认数据
5. SequenceEqual 判断两个Observable是否按相同的数据序列
6. SkipUntil 丢弃原始Observable发射的数据，直到第二个Observable发射了一个数据，然后发射原始Observable的剩余数据
7. SkipWhile 丢弃原始Observable发射的数据，直到第一个特定的条件为假，然后发射原始Observable剩余的数据
8. TakeUntil 发射来自原始Observable的数据，直到第二个Observable发射了一个数据或一个通知
9. TakeWhile 发射原始Observable的数据，直到一个特定的条件为真，然后跳过剩余的数据

算术和聚合操作

用于整个数据序列

1. Average 计算Observable发射的数据序列的平均值，然后发射这个结果
2. Concat 不交错的连接多个Observable的数据
3. Count 计算observable发射的数据个数，然后发射这个结果。
4. Max 计算并发射数据序列的最大值
5. Min 计算并发射数据序列的最小值
6. Reduce 按顺序对数据序列的每一个应用某个函数，然后返回这个值
7. Sum 计算并发射数据序列的和

连接操作

有精确可控的订阅行为的特殊Observable

1.Connect 指示一个可连接的Observable开始发射数据给订阅者

2. Publish 将一个普通的Observable转换为可连接的
3. RefCount 使一个可连接的Observable表现得像一个普通的Observable
4. Replay 确保所有的观察者收到同样的数据序列，即使他们在Observable开始发射数据之后才开始订阅

转换操作

1. To 将Observable转换为其他的对象或数据结构
2. Blocking 阻塞Observable的操作符

Observable的辅助操作符

1. matrialize() 将Observable转换成一个通知列表convert an Observable into a list of Notifications
2. dematerialize() 将上面的结果逆转回一个Observable
3. timestamp() 给Observable发射的每个数据项添加一个时间戳
4. serialize() 强制Observable按次序发射数据并且要求功能是完好的。
5. cache() 记住Observable发射的数据序列并发射相同的数据序列给后续的订阅者
6. observeOn() 指定观察者观察Observable的调度器
7. subscribeOn() 指定Observable执行任务的调度器
8. doOnEach() 注册一个动作，对Observable发射的每个数据项使用
9. doOnComplete 注册一个动作，对正常完成的Observable使用
10. doOnError 注册一个动作，对发生错误的Observable使用
11. doOnTerminate 注册一个动作，对完成的Observable使用，无论是否发生错误
12. doOnSubscribe 注册一个动作，在观察者订阅时使用
13. doOnUnsubscribe 注册一个动作，在观察者取消订阅时使用
14. finallyDo 注册一个动作，在Observable完成时使用
15. delay 延时发射Observable的结果
16. delaySubscription 延时处理订阅请求
17. timeInterval 定期发射数据
18. using 创建一个只在Observable生命周期存在的资源
19. single 强制返回单个数据，否则抛出异常
20. singleOrDefault 如果Observable完成时返回了单个数据，就返回它，否则返回默认数据
21. toFuture、toIterable、toList 将Observable转换为其它对象或数据结构

操作符决策树

几种主要的需求
1）直接创建一个Observable（创建操作）
2）组合多个Observable（组合操作）
3）对Observable发射的数据执行变换操作（变换操作）
4）从Observable发射的数据中取特定的值（过滤操作）
5）转发Observable的部分值（条件/布尔/过滤操作）
6）对Observable发射的数据序列求值（算术/聚合操作）

Rxjava基础分析

数据流：按照事先规定的顺序被读取一次的数据的一个序列
变化传播：类似观察者模式，变化了通知别人，把变化传播出去。
编程范式：计算机编程的基本风格或典范模式

类比流水线，数据流是要加工的物品，变化传播是前一环节的加工结果传到下一环节，编程范式每种物品的加工方式不同，流水线就不同。

异步的、基于回调的、基于事件的 事件分发的库，消息传递的库。

RxJava1 基本元素：
Observable、Observer、Subscribtion、OnSubscribe、Subscriber

RxJava2 基本元素：
observable和Flowable、Observer和Subsciber、Disposable和Subscription、相应的onSubscribe、Emitter

Flowable新增了背压策略，Observable完全不具备背压能力

RxJava1 基本元素

Observable
是一个被观察者
通过Observable创建一个可观察的序列（create方法）
通过subscribe去注册一个观察者

Observer
用于接收数据
作为Observable的subscribe方法的参数

Subscription
订阅，用于描述被观察者与观察者之间的关系
用于取消订阅和获取当前的订阅状态

OnSubscribe
1. 当订阅时会出发此接口调用
2. 在Observable内部，实际作用是向订阅者发射数据

Subscriber
1. 实现了Observer和Subscription
2. 只有自己才能阻止自己

分析：
Observable相当于打电话的人 它subscribe了一个Subscriber，通过onSubscribe通信链路向接电话的人发射数据

接电话的人随时可以去挂掉电话

背压概念

1. 异步环境下产生的问题
2. 发送和处理速度不统一
3. 是一种流速控制解决策略

生产者生产了东西，丢入缓存队列，消费者从缓存队列中去消费。
消费者的处理速度大于生产速度，空置。
生产者生产速度大于消费者消费速度，会导致队列溢出。

RxJava2的基本元素

1. Observable
2. Observer 多了一个onSubscribe方法，参数是Disposable
3. Disposable
4. OnSubscribe -> ObservableOnSubscribe
5. Emitter （新增的）

RxJava2（背压版）基本元素

1. Flowable

易流动的 被观察者，支持背压
通过Flowable创建一个可观察的序列（create方法）
通过subscribe去注册一个观察者

2. Subscriber

单独的接口，和Observer的方法类似

Flowable的subscribe方法的参数

3. Subscription

订阅。支持背压，有用于背压的request方法

4. OnSubscribe

FlowableOnSubscribe实例
当订阅时回触发此接口调用。在Flowable内部，实际作用是向观察者发射数据

5. Emitter

一个发射数据的接口，和Observer的方法类似

本质是对Observer和Subscriber的封装

进行仿写：
1. Observable -> Caller
2. Observer -> Callee
3. Subscription -> Calling
4. OnSubscribe -> OnCall
5. Subscriber -> Receiver

RxJava2 无背压仿写

1. Observable -> Caller
2. Observer -> Callee
3. Disposable -> Release
4. OnSubscribe ->OnCall
5. Emitter ->Emitter

有背压版本
1. Flowable -> Telephoner
2. Subscriber -> Receiver
3. Subscription -> Drop
4. OnSubscribe -> OnCall
5. Emitter -> Emitter

操作符操作

像现实中的寄信
map就是邮局。
lift操作符

1. 接收原OnSubscribe和当前的Operator
2. 创建一个新的OnSubscribe并返回新的Observable
3. 用新的Subscriber包裹旧的Subscriber
4. 在新的Subscriber里做完变换再传给旧的 Subscriber

类似于代理机制。用新的对象包裹，对旧包裹后再回传回去。像是寄快递

AbstractObservableWithUpstream

1. 实现操作符需要继承
2. 利用其subscribeActual方法
3. 用原 Observable 去 subscribe 变换后的 Observer

线程变换简介
1. 让代码可以在不同的线程执行
2. subscribeOn 订阅时的线程
3. observeOn 接收时的线程
4. scheduler 实际做线程变换

Scheduler调度者源码分析

1. 传入不同Scheduler来使用不同的线程
2. 用Scheduler创建Worker来使用真正的线程池
3. 传入具体操作 Action0
4. 通过schedule方法来实现调度

Schedulers 抽象类

Worker真正做线程调度的类

Runnable 在线程中执行的操作

schedule 做线程调度的方法，入参为Runnable

Android中的Scheduler
1. 通过Handler和looper来说实现执行在主线程

subscribeOn原理

1. 继承AbstractFlowableWithUpstream
2. 实现subscribeActual方法
3. 利用Scheduler将发出动作放到线程中执行

整体变换
RxJava与Retrofit结合有两个点：
返回本来应该是Call 变成了Observable
通过CallAdapter返回的。

MVP模式：

public interface BaseView<T extends BasePresenter>{

}

public interface BasePresenter<T extends BaseView>{

protected final T mView;

public BasePresenter(T view){
          mView = view;
}

Contract

interface View extends BaseView<Presenter>{

void update(Sring text);

}

interface Presenter extends BasePresenter<View>{
public Presenter(View view){
          super(view);
}
abstract void loadData();

}