一、啥是操作符
理解为可以控制流程的方法。非常强大而且复杂。
看中文翻译文档:
https://www.gitbook.com/book/mcxiaoke/rxdocs/details
虽然文档是Rx1.x的，但是操作符部分和Rxjava2内容相差还是不大的，可以进行观看。
二、操作符分类
操作符分类有十三种:
| 名称 | 解析 |
|---|---|
| 创建操作 | 用于创建Observable的操作符 |
| 变换操作 | 这些操作符可用于对Observable发射的数据进行变换 |
| 过滤操作 | 这些操作符用于从Observable发射的数据中进行选择 |
| 组合操作 | 组合操作符用于将多个Observable组合成一个单一的Observable |
| 错误处理 | 这些操作符用于从错误通知中恢复 |
| 辅助操作 | 一组用于处理Observable的操作符 |
| 条件和布尔操作 | 用于单个或多个数据项，也可用于Observable |
| 算术和聚合操作 | 用于整个数据序列 |
| 异步操作 | 单独的 rxjava-async 模块，它们用于将同步对象转换为Observable |
| 连接操作 | 一些有精确可控的订阅行为的特殊Observable |
| 转换操作 | 只有一个To，将Observable转换为其它的对象或数据结构 |
| 阻塞操作 | 阻塞Observable的操作符 |
| 字符串操作 | 一些用于处理字符串序列和流的特殊操作符 |
几种主要的需求：
- 直接创建一个Observable（创建操作）
- 组合多个Observable（组合操作）
- 对Observable发射的数据执行变换操作（变换操作）
- 从Observable发射的数据中取特定的值（过滤操作）
- 转发Observable的部分值（条件/布尔/过滤操作）
- 对Observable发射的数据序列求值（算术/聚合操作）
因为操作符太多了，所以每种操作符记录一点。创建操作符在基本使用的时候有说到了，本次是变换操作符
三、变换操作符
顾名思义，变换操作符就是用来变换的，变换什么呢？ 没错就是类型。
3.1 变换操作符
| 操作符 | 解析 |
|---|---|
| buffer() | 缓存，可以简单的理解为缓存，它定期从Observable收集数据到一个集合，然后把这些数据集合打包发射，而不是一次发射一个 |
| map() | 对序列的每一项都应用一个函数来变换Observable发射的数据序列 |
| flatMap() , concatMap() , flatMapIterable() | 将Observable发射的数据集合变换为Observables集合，然后将这些Observable发射的数据平坦化的放进一个单独的Observable |
| switchMap() | 将Observable发射的数据集合变换为Observables集合，然后只发射这些Observables最近发射的数据 |
| scan() | 对Observable发射的每一项数据应用一个函数，然后按顺序依次发射每一个值 |
| groupBy() | 将Observable分拆为Observable集合，将原始Observable发射的数据按Key分组，每一个Observable发射一组不同的数据 |
| buffer() | 它定期从Observable收集数据到一个集合，然后把这些数据集合打包发射，而不是一次发射一个 |
| window() | 定期将来自Observable的数据分拆成一些Observable窗口，然后发射这些窗口，而不是每次发射一项 |
| cast() | 在发射之前强制将Observable发射的所有数据转换为指定类型 |
3.2 map
这个操作符的意思官方翻译过来是这样子说的:
对Observable发射的每一项数据应用一个函数，执行变换操作
是不是一脸懵逼，好吧。其实我也是。其实大白话的意思就是:
把Observable的数据经过转换，返回转换后的数据。
这样子估计你也不懂，看个栗子��，map后面的Function有两个泛型参数，第一个是输入的类型，第二个是转换后输出返回的类型，可以看到apply方法要返回的是第二个参数的类型。
Observable.just(userParm)
.map(new Function<UserParm, Login>() {
@Override
public Login apply(UserParm userParm) throws Exception {
//进行网络请求登录
Login login = api.loginCall(userParm).execute().body();
return login;
}
})
.subscribeOn(Schedulers.io()) //网络请求是io操作，需要在线程中执行
.observeOn(AndroidSchedulers.mainThread()) //结果需要更新ui就要在主线程中操作
.subscribe(new Consumer<Login>() {
@Override
public void accept(Login login) throws Exception {
//onNext
}
});api为:
//登录Call
@POST("mobile/user/login")
Call<Login> loginCall(@Body UserParm userParm);3.3 flatMap
理解为: Observable发射了啥出去，变换之后 还是返回 Observable，知识发射的啥变成了转换之后的啥。
看个栗子��：
/**
* 登录
* @param name 用户名
* @param pass 密码
* @return 返回登录成功的token
*/
private void getUserInfo(String name,String pass){
final UserParm userParm = new UserParm(name,pass);
//Function的第二个泛型一定是ObservableSource类型，要转换的类型在ObservableSource里面
Observable.just(userParm).flatMap(new Function<UserParm, ObservableSource<Login>>() {
@Override
public ObservableSource<Login> apply(UserParm userParm) throws Exception {
//返回的是ObservableSource
//登录
return api.login(userParm);
}
}).flatMap(new Function<Login, ObservableSource<UserProfile>>() {
@Override
public ObservableSource<UserProfile> apply(Login login) throws Exception {
//获取用户信息
return api.getUserProfile(login.getData().getToken());
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<UserProfile>() {
@Override
public void accept(UserProfile userProfile) throws Exception {
//onNext
tvResult.setText(userProfile.getData().getUser().getName());
}
}, new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
//onError
}
}, new Action() {
@Override
public void run() throws Exception {
//onComplete
}
});
}3.4 switchMap
使用和flatMap是一样的，有一点不同，就是有新的数据发来的时候，之前的就会取消掉。
可以应用在 多次操作的时候，避免之前的操作慢，后来的操作快，导致前面的数据覆盖了后面的新数据。
3.5 buffer
缓存发射的数据为一个list，到观察者的时候参数就是一个list。
/**
* 测试缓存
*/
private void testBuffer() {
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
for(int i=0; i<10 ;i++){
e.onNext(i+"");
}
e.onComplete();
}
})
.buffer(500,TimeUnit.MILLISECONDS) //缓存500毫秒内发射的数据
.subscribe(new Consumer<List<String>>() {
@Override
public void accept(List<String> strings) throws Exception {
//这里参数就为list，缓存了 500毫秒内发射的String
Log.e(TAG, "accept: "+strings.size());
}
});
}还有其他的操作符自行测试啦。