上游发送事件给下游的过程中，可以做一些事件类型的转换，这就是变换操作符的作用。
变换操作符：上游 --> 变换操作符（往右变换流向的时候，进行变换）-->下游
map
|
|---|
| 将被观察者发送的事件转换为任意的类型事件， 如 把上游 int 变换成 String，观察者观察 String 类型。 |
public void rx_map() {
Observable.just(1, 2)
// 上游和下游之间 变换 x 10
.map(new Function<Integer, String>() {
@Override
public String apply(Integer integer) throws Exception
log("第一次转换 apply: " + integer);
return String.format("First %d", integer * 10);
// 如果返回 null， 下游无法接收
}
})
// 可多次变换
.map(new Function<String, String>() {
@Override
public String apply(String s) throws Exception {
log("第二次转换 apply: " + s);
return String.format("Second: [%s]", s);
}
})
// 订阅
.subscribe(createObserverString());
}
// 输出
第一次转换 apply: 1
第二次转换 apply: First 10
下游处理事件 onNext Second: [First 10]
第一次转换 apply: 2
第二次转换 apply: First 20
下游处理事件 onNext Second: [First 20]
下游处理事件 onComplete
上面的例子中，上游 int --> [ int to String] --> [ String to String] --> 下游 String，共经过了两次变换。
map 的入参是 Function 接口的对象，里面只有一个方法 apply ，变换的逻辑就这在这里处理。
flatMap
|
|---|
flatMap：将发送的事件进行转换，以一个新的事件发送出去。如 把上层 int 变换成 ObservableSource<String>（还可以再发射多次事件）观察者 String 类型，不排序 |
public void rx_flatMap() {
Observable.just("A", "B", "C")
// 变换操作符
.flatMap(new Function<String, ObservableSource<String>>() {// ? 通配符 默认 Object
@Override
public ObservableSource<String> apply(final String s) throws Exception {
List<String> list = new ArrayList<>();
for (int i = 0; i < 3; i++) {
list.add(s + i);
}
return Observable.fromIterable(list).delay(2, TimeUnit.SECONDS); // 创建型操作符，创建被观察者
}
})
.subscribe(createObserverString());
}
// 输出
下游处理事件 onNext B0
下游处理事件 onNext B1
下游处理事件 onNext B2
下游处理事件 onNext C0
下游处理事件 onNext A0
下游处理事件 onNext A1
下游处理事件 onNext A2
下游处理事件 onNext C1
下游处理事件 onNext C2
下游处理事件 onComplete
例子中就是把发射的每一个字母都转换成 字母加数字 的列表发送出去，比如 A 就转换成 List<String>{ "A1", "A2", "A3"} 发出去。可以从输出发现 flatMap 输出结果是 无序 的。RxJava 的 API 中也提供了有序的操作 concatMap。
concatMap
使用方法和 flatMap 是一样的，但 concatMap 的 输出是有序输出。还使用上面的逻辑，不论执行到少次都是会按照 A0 A1 A2 B0 B1 B2 C0 C1 C2 顺序输出。
groupBy
|
|---|
| groupBy 会将事件按照某一种变化方式编组，把分组后的内容按照以组为单位，将整组事件发射出去 |
public void rx_groupBy() {
Observable.just(1, 2, 6, 3, 4, 5)
.groupBy(new Function<Integer, String>() {
@Override
public String apply(Integer integer) throws Exception {
return integer > 3 ? "3 以上" : "3 以下";
}
})
//使用 groupBy 下游是有标准的
.subscribe(new Consumer<GroupedObservable<String, Integer>>() {
@Override
public void accept(final GroupedObservable<String, Integer> stringIntegerGroupedObservable) throws Exception {
log("accept: " + stringIntegerGroupedObservable.getKey());
// 以上还不能把信息打印全面，只是拿到了分组的 key
// 输出细节，还需要在包裹一层
// 细节 GroupedObservable 被观察者
stringIntegerGroupedObservable.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
log("accept: 类别： " + stringIntegerGroupedObservable.getKey() + " 数值： " + integer);
}
});
}
})
;
}
// 输出
accept: 3 以下
accpet: 类别： 3 以下 数值： 1
accpet: 类别： 3 以下 数值： 2
accept: 3 以上
accpet: 类别： 3 以上 数值： 6
accpet: 类别： 3 以下 数值： 3
accpet: 类别： 3 以上 数值： 4
accpet: 类别： 3 以上 数值： 5
groupBy 的入参也是 Function 在 apply 中，处理分组原则，返回的时期是分组的 key （这里一定要注意）
其下游也是有要求的，订阅的 Observer 类，要包装一个 GroupedObservable<key类型, value类型>，这里观察者会根据分组结果收到对应 GroupedObservable 的被观察者对象，这个对象只含有分组的信息，要为这个 GroupedObservable 再订阅一个对应的下游，才能真正接收到分组后的时事件。
buffer
从被观察者发送的事件中获取一定量事件放到缓存区并发送
|
|
|---|---|
| 每三个事件分为一批发射出去 | 可以为 butter 添加步长，每三个事件中前两个时间为一批发出去 |
如果有很多的数据，不想全部一起发射出去，分批次，可以先缓存到 buffer
public void rx_buffer() {
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
for (int i = 0; i < 20; i++) {
emitter.onNext(i);
}
emitter.onComplete();
}
})
// 变换 buffer
.buffer(10)
.subscribe(new Consumer<List<Integer>>() {
@Override
public void accept(List<Integer> integers) throws Exception {
log("accept: " + integers);
}
});
}
//输出
accept: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
accept: [10, 11, 12, 13, 14, 15, 16, 17, 18, 19]
这里需要注意，最后观察者会受到 List<T> 的数据类型，比如例子中 20个Integer事件， butter(10)，观察者会收到两个 List<integer> 结合。
.buffer(2,6)
// 输出
accept: [0, 1]
accept: [6, 7]
accept: [12, 13]
accept: [18, 19]
网友评论