RxJava操作符
官网地址：http://reactivex.io/documentation/operators.html
一个很详细的博客：https://juejin.cn/post/6844903885245513741
按照操作符的类型可分为：创建操作符，变换操作符，过滤操作符，组合操作符，错误操作符，辅助操作符，条件和布尔操作符，算术和聚合操作符 及 连接操作符 等。
RxJava3 中
Action<T>被弃用了，改为Consumer<T>
创建操作符
-
create传入自定义
Onsubscribe。
Observable.create(new Observable.OnSubscribe<T>(){
@Overrode
public void call(Subscriber<? super T>) sub){
sub.onNext("0");
sub.onNext("1");
sub.onCompleted();
}
})
- from 传入数组：
String list = {"0","1","2"}
Observable.from(list);
- just
Observable.just("0","1","2")
- interval 创建一个 按固定时间间隔发射整数<Long>序列 的Observable。
Observable<Long> observable = Observable.interval(3, TimeUnit.SECONDS, Schedulers.trampoline());
//, Schedulers.trampoline() 指定一个线程，默认新线程（ＮＥＷ＿ＴＨＲＥＡＤ），这种情况下主线程结束，程序就结束了。
observable.subscribe(new Consumer<Long>() { //泛型只能是Long类型，下同
public void accept(Long aLong) throws Throwable {
//TODO
}
});
<img src="C:\Users\Administrator\AppData\Roaming\Typora\typora-user-images\image-20201207122725363.png" alt="image-20201207122725363" style="zoom:50%;" />
- range 创建 发射指定范围的整数<Integer>序列 的Observable。
Observable<Integer> observable = Observable.range(0,5);//左闭右开
- repeat 重复多少次
Observable<Integer> observable = Observable.range(0,5)
.repeat(2)//重复两次
变换操作符
- map 将一个对象转变为另一个对象； ----偷梁换柱
Observable<Integer> observable = Observable.range(0,5);
Observable<String> map = observable.map(new Function<Integer, String>() {
public String apply(Integer integer) throws Throwable {
return integer.toString();
}
});
-
flatMap和cast
flatMap将一个对象转为Observable集合；cast转为集合后电脑不知道你的Observable的泛型类型，需要用cast转换。 -----偷天换日
final Observable<Integer> observable = Observable.range(0,5);
observable.flatMap(new Function<Integer, ObservableSource<?>>() {
public ObservableSource<?> apply(Integer integer) throws Throwable {
return Observable.just("0","1","haha");
}
}).cast(String.class);
- concatMap
和FlatMap类似。区别在于：flatMap不保证顺序，当上一个操作未执行时，下一个操作可能已经发射。
concatMap保证了顺序，必须等到上一个操作发射完，才会发射下一个。
concatMap和flatMap的功能是一样的， 将一个发射数据的Observable变换为多个Observables，然后将它们发射的数据放进一个单独的Observable。只不过最后合并ObservablesflatMap采用的merge，而concatMap采用的是连接(concat)。总之一句一话,他们的区别在于：concatMap是有序的，flatMap是无序的，concatMap最终输出的顺序与原序列保持一致，而flatMap则不一定，有可能出现交错。 ---出自这里
- **flatMapIteravle ** 把每一个元素转换成
Iterable
Observable
.interval(1, TimeUnit.SECONDS, Schedulers.trampoline())
.flatMapIterable( aLong -> Arrays.asList("a", "integer") )
.subscribe(s -> System.out.println(s));
- buffer 将Observable转为一个新的Observable，这个新Observable每次发射一组列表，而不是一个个发射。
Observable
.interval(1, TimeUnit.SECONDS, Schedulers.trampoline())
.buffer(3)//集满3个再一并发射
- groupBy 将Observable转为一个新的Observable,然后可以分组。
Person p1 = new Person("hh",13);
Person p2 = new Person("ff",13);
Person p3 = new Person("gg",23);
Person p4 = new Person("aa",13);
Person p5 = new Person("bb",33);
@NonNull Observable<GroupedObservable<Integer, Person>> group = Observable.just(p1, p2, p3, p4, p5)
.groupBy(p -> p.age);//按年龄进行排序
Observable.concat(group).subscribe(new Consumer<Person>() {
@Override
public void accept(Person p) throws Throwable {
System.out.println(">>>>>>"+p.name+","+p.age);
}
});
//Result ----------------------按年龄进行排序结果----
>>>>>>hh,13
>>>>>>ff,13
>>>>>>aa,13
>>>>>>gg,23
>>>>>>bb,33
过滤操作符
- filter 过滤
Observable.just(1, 2, 3, 4,7, 5)
.filter(i -> i>2)//只保留大于2的元素
-
elementAt 返回指定位置的数据
-
distinct 去重
-
skip 和 take skip过滤掉前n项，take只取前n项目。
-
ignoreElements 忽略所有元素，只保留onComplete和onError通知
-
throttleFirst 定期发射这个时间段里源Observable发送的第一条数据。默认调度器
computation。
Observable.interval(1000,TimeUnit.MILLISECONDS, Schedulers.trampoline())
.throttleFirst(2000,TimeUnit.MILLISECONDS)
.subscribe(l -> System.out.println(l));
//-------------------结果-----------
// 0，3，6，9， .........
//2000改为1900则是 2，4，6，8 ...自己体会
image-20201208100601354.png
- throttleWithTimeOut
通过时间来限流。源Observable每次发射出来一个数据后都会进行计时，如果再设定好的时间结束前Observable有新的数据发射出来，这个数据就会被丢弃，同时开始重新计时。
组合操作
- startWith 在源数据之前插入一系列数据；
Observable.just(3,4,5).startWith(1,2)
-
merge 合并
Observable，可能会数据交错。
Observable.merge(obs1,obs2)
-
concat 合并
Observable，严格按照顺序发射。
Observable.concat(obs1,obs2)
- zip 合并数据。
@NonNull Observable<Long> obs1 = Observable.just(1L,2L,3L);
@NonNull Observable<String> obs2 = Observable.just("1L","2L","3L");
Observable.zip(obs1,obs2,(l1,l2) -> l1+","+l2).subscribe(l -> System.out.println(l));
----------------------
1,1L
2,2L
3,3L
- **combineLasetst ** 用前者最后一条发射的数据进行合并。
@NonNull Observable<Long> obs1 = Observable.just(1L,2L,3L);
@NonNull Observable<String> obs2 = Observable.just("1L","2L","3L");
Observable.combineLatest(obs1,obs2,(l1,l2) -> l1+","+l2).subscribe(l -> System.out.println(l));
------------------------
3,1L
3,2L
3,3L
辅助操作符
-
delay 延迟操作
-
Do
doXXX,do后面加上某个生命周期，即对这个生命周期加了个回调。 -
subscribeOn、observeOn 指定线程；在哪个线程
发布，在哪个线程观察； -
timeOut 指定一段时间，超时则
onError;
错误处理操作符
- catch onErrorReturn、onErrorResume和onExceptionResumeNext
RxJavaPlugins.setErrorHandler(e -> { });
Observable.create((ObservableOnSubscribe<Integer>) emitter -> {
for (int i = 0; i < 20; i++) {
if (i>2{
emitter.onError(new Throwable("ERROR"));
}
emitter.onNext(i);
}
emitter.onComplete();
})onErrorReturn(t -> 100)//错误的时候直接返回数据100
0
1
2
100
complete
- retry 出错后重试的次数。
布尔操作符
-
all 对源
Observable发射的所有数据进行判断，全满足条件则返回true，否则返回false。
Observable.just(1,2,3,4,5)
.all(integer -> {
System.out.println(integer);
return integer<3;
})
.subscribe(aBoolean -> System.out.println(aBoolean));
//-----------------结果------
1
2
3
false
- contains 是否包含这个数据；isEmpty 是否为空；
条件操作符
- amb 多个Observable中，只发射第一个发射数据的的Observable。
- defaultEmpty 没发射数据则返回一个默认数据。
转换操作符
将Observable转换成另一个对象或数据结构。比map更简介。
- toList
Observable.just(1,2,3,4,5)
.toList()
.subscribe(list -> {System.out.println(list.size())});//{1,2,3.4,5}
- toSortedList 和toList一样，不过多了个排序功能，所以发射对象要Comparable。
Observable.just(1,2,3,4,5)
.toMap(new Function<Integer, String>() {
@Override
public String apply(Integer integer) throws Throwable {
return integer.toString();
}
})
.subscribe(map -> System.out.println(map.toString()));
//-------------------------result---------------
//.subscribe(map -> System.out.println(map.toString()));
//返回的是 {("1",1),("2",2) ...("5",5）}这个map
未完待续。。。
ps: 其实还有很多没写，太多了，学不动了😵
网友评论