RxJava2.0 操作符（2）—— Transform 转换符
Transforming 转换符
Buffer — 它定期从 Observable
收集数据到一个集合，然后把这些数据集合打包发射，而不是一次发射一个。
Cast — 在发射之前强制将 Observable 发射的所有数据转换为指定类型。
FlatMap, concatMap(), and flatMapIterable() 将 Observable 发射的数据集合变换为 Observables 集合，然后将这些 Observable 发射的数据平坦化的放进一个单独的 Observable。
GroupBy — 将 Observable 分拆为 Observable 集合，将原始 Observable 发射的数据按 Key 分组，每一个 Observable 发射一组不同的数据。
Map — 对序列的每一项都应用一个函数来变换 Observable 发射的数据序列。
Scan — 对 Observable 发射的每一项数据应用一个函数，然后按顺序依次发射每一个值。
Window — 定期将来自 Observable 的数据分拆成一些 Observable 窗口，然后发射这些窗口，而不是每次发射一项。
2.1 Buffer
它定期从 Observable 收集数据到一个集合，然后把这些数据集合打包发射，而不是一次发射一个。
buffer
示例代码：
Observable.just("hello", "wolder", "rxjava", "transforming")
.buffer(3)
.subscribe(new Observer<List<String>>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.e(TAG, " onSubscribe : " + d.isDisposed());
}
@Override
public void onNext(@NonNull List<String> stringList) {
Log.e(TAG, " onNext : size :" + stringList.size());
Log.e(TAG, " onNext : content :" + stringList.toString());
}
@Override
public void onError(@NonNull Throwable e) {
Log.e(TAG, " onError" + e.getMessage());
}
@Override
public void onComplete() {
Log.e(TAG, " onComplete");
}
});
输出结果：
onSubscribe : false
onNext : size :3
onNext : content :[hello, wolder, rxjava]
onNext : size :1
onNext : content :[transforming]
onComplete
buffer(count, skip) 从原始 Observable 的第一项数据开始创建新的缓存，此后每当收到 skip 项数据，用 count 项数据填充缓存：开头的一项和后续的 count-1 项，它以列表 (List) 的形式发射缓存，取决于 count 和 skip 的值，这些缓存可能会有重叠部分（比如 skip < count 时），也可能会有间隙（比如 skip > count 时）。
示例代码 2：
Observable.just("hello", "wolder", "rxjava", "transforming"，"buffer")
.buffer(3, 2)
.subscribe(new Observer<List<String>>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.e(TAG, " onSubscribe : " + d.isDisposed());
}
@Override
public void onNext(@NonNull List<String> stringList) {
Log.e(TAG, " onNext : size :" + stringList.size());
Log.e(TAG, " onNext : content :" + stringList.toString());
}
@Override
public void onError(@NonNull Throwable e) {
Log.e(TAG, " onError" + e.getMessage());
}
@Override
public void onComplete() {
Log.e(TAG, " onComplete");
}
});
输出结果：
onSubscribe : false
onNext : size :3
onNext : content :[hello, wolder, rxjava]
onNext : size :3
onNext : content :[rxjava, transforming, buffer]
onNext : size :1
onNext : content :[buffer]
onComplete
2.1.1 Cast
在发射之前强制将 Observable 发射的所有数据转换为指定类型。
cast
2.2 FlatMap
flatmap
示例代码：
Observable.just("hello", "world")
.flatMap(new Function<String, ObservableSource<String>>() {
@Override
public ObservableSource<String> apply(@NonNull String s) throws Exception {
return Observable.just("flatMap : " + s);
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
}
@Override
public void onNext(@NonNull String s) {
Log.e(TAG, " onNext : " + s);
}
@Override
public void onError(@NonNull Throwable e) {
Log.e(TAG, " onError" + e.getMessage());
}
@Override
public void onComplete() {
Log.e(TAG, " onComplete");
}
});
输出结果：
onNext : flatMap : hello
onNext : flatMap : world
onComplete
2.2.1 ConcatMap
它类似于最简单版本的 flatMap，但是它按次序连接而不是合并那些生成的 Observables，然后产生自己的数据序列。
2.2.2 SwitchMap
它和 flatMap 很像，除了一点：当原始 Observable 发射一个新的数据（Observable）时，它将取消订阅并停止监视产生执之前那个数据的 Observable，只监视当前这一个。
2.2.3 Split
在特殊的 StringObservable 类（默认没有包含在 RxJava 中）中还有一个 split 操作符。它将一个发射字符串的 Observable 转换为另一个发射字符串的 Observable，只不过，后者将原始的数据序列当做一个数据流，使用一个正则表达式边界分割它们，然后合并发射分割的结果。
2.3 GroupBy()
将 Observable 分拆为 Observable 集合，将原始 Observable 发射的数据按 Key 分组，每一个 Observable 发射一组不同的数据。
groupBy
示例代码：
Observable.interval(1, TimeUnit.SECONDS).take(10).groupBy(new Function<Long, Long>() {
@Override
public Long apply(@NonNull Long aLong) throws Exception {
return aLong % 3;
}
}).subscribe(new Consumer<GroupedObservable<Long, Long>>() {
@Override
public void accept(@NonNull final GroupedObservable<Long, Long> go) throws Exception {
go.subscribe(new Consumer<Long>() {
@Override
public void accept(@NonNull Long aLong) throws Exception {
Log.e(TAG, "key:" + go.getKey() + ", value:" + aLong);
}
});
}
输出结果：
key:0, value:0
key:1, value:1
key:2, value:2
key:0, value:3
key:1, value:4
key:2, value:5
key:0, value:6
key:1, value:7
key:2, value:8
key:0, value:9
2.4 Map
对序列的每一项都应用一个函数来变换 Observable 发射的数据序列。
map
示例代码：
Observable.just("hello", "world", "rxjava").map(new Function<String, String>() {
@Override
public String apply(@NonNull String s) throws Exception {
return "map + " + s;
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(@NonNull String s) throws Exception {
Log.e(TAG, s);
}
});
输出结果：
map + hello
map + world
map + rxjava
2.5 Scan
对 Observable 发射的每一项数据应用一个函数，然后按顺序依次发射每一个值。
scan
示例代码：
Observable.just(8, 5, 2, 3).scan(new BiFunction<Integer, Integer, Integer>() {
@Override
public Integer apply(@NonNull Integer i1, @NonNull Integer i2) throws Exception {
return i2 + i1;
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(@NonNull Integer integer) throws Exception {
Log.e(TAG, "accept:" + integer);
}
});
输出结果：
accept:8
accept:13
accept:15
accept:18
2.6 Window
定期将来自 Observable 的数据分拆成一些 Observable 窗口，然后发射这些窗口，而不是每次发射一项。
window
示例代码：
List<String> list = new ArrayList<>();
for (int i = 0; i < 10; i++) {
list.add("hello i:" + i);
}
Observable.fromIterable(list).window(4).subscribe(new Consumer<Observable<String>>() {
@Override
public void accept(@NonNull Observable<String> stringObservable) throws Exception {
Log.e(TAG, "\n group:");
stringObservable.subscribe(new Consumer<String>() {
@Override
public void accept(@NonNull String s) throws Exception {
Log.e(TAG, "Content:" + s);
}
});
}
});
输出结果：
group:
Content:hello i:0
Content:hello i:1
Content:hello i:2
Content:hello i:3
group:
Content:hello i:4
Content:hello i:5
Content:hello i:6
Content:hello i:7
group:
Content:hello i:8
Content:hello i:9
网友评论