- Observable的创建操作符
- just——直接返回list
Observable.just(new String[]{"This", "is", "RxJava"})
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<String[]>() {
@Override
public void accept(@NonNull String[] strings) throws Exception {
for(String string :strings){
Log.i("just",""+strings);
}
}
}) ;
- fromArray——依次返回list的每个item
Observable.fromArray(new String[]{"This", "is", "RxJava"})
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<String>() {
@Override
public void onSubscribe(@NonNull Disposable disposable) {
}
@Override
public void onNext(@NonNull String s) {
}
@Override
public void onError(@NonNull Throwable throwable) {
}
@Override
public void onComplete() {
}
});
- interval操作符——延时几秒，每隔几秒开始执行
Flowable.interval(1, TimeUnit.SECONDS)
Flowable.interval(0, 1, TimeUnit.SECONDS)////参数一：延时0秒；参数二：每隔1秒开始执行
- timer——延时之后发射数据项为0的Observable<Long>
Observable.timer(1000, TimeUnit.MILLISECONDS)
.subscribe(new Consumer<Long>() {
@Override
public void accept(@NonNull Long aLong) throws Exception {
}
});
- range——创建从n开始，个数为m的连续数字，比如range(3,4)，就是创建3、4、5、6的一组数字
- Observable的聚合操作符
- reduce—— 发射最终的结果
Observable.just(1,2,3,4).reduce(new BiFunction<Integer, Integer, Integer>() {
@Override
public Integer apply(@NonNull Integer integer, @NonNull Integer integer2) throws Exception {
return integer+integer2;
}
})
.subscribe(integer -> Log.d("rxjava",integer.toString()));//10
scan——与reduce很像，对Observable发射的每一项数据应用一个函数，然后按顺序依次发射每一个值
Observable.just(2, 3, 4, 5, 6)
.scan(new BiFunction<Integer, Integer, Integer>() {
@Override
public Integer apply(@NonNull Integer integer, @NonNull Integer integer2) throws Exception {
return integer + integer2;
}
})
.subscribe(integer -> Log.d("rxjava", integer.toString())); //2,5,9,14,20
- collect——收集数据到一个可变的数据结构
Observable.just(1,2,3,4)
.collect(new Callable<List<Integer>>() { //创建数据结构
@Override
public List<Integer> call() {
return new ArrayList<Integer>();
}
}, new BiConsumer<List<Integer>, Integer>() { //收集器
@Override
public void accept(@NonNull List<Integer> integers, @NonNull Integer integer) throws Exception {
integers.add(integer);
}
})
.subscribe(new Consumer<List<Integer>>() {
@Override
public void accept(@NonNull List<Integer> integers) throws Exception {
}
};
- Observable的条件操作符
- contains
Observable.just(1,2,3,4)
.contains(3)
.subscribe(aBoolean -> Log.d("rxjava",aBoolean.toString())); //true
-
takeUntil——当发射的数据满足某个条件（包含该数据）后终止
takeWhile—— 当发射的数据满足某个条件（不含该数据）后终止
Observable.just(2,3,4,5)
.takeUntil(new Predicate<Integer>() {
@Override
public boolean test(@NonNull Integer integer) throws Exception {
return integer==4;
}
}).subscribe(integer -> Log.d("rxjava",integer.toString())); //2,3,4
- sequenceEqual—— 判断两个Observable发射的数据是否相同（数据，发射顺序）
Observable.sequenceEqual(Observable.just(2,3,4,5),Observable.just(2,3,4,5)) .subscribe(new Consumer<Boolean>() {
@Override
public void accept(@NonNull Boolean aBoolean) throws Exception {
}
});
- Observable的过滤操作符
- filter——过滤数据
.filter(new Predicate<String>() {
@Override
public boolean test(String s) throws Exception {
//true为继续执行下去，false拦截
return s.trim().length() > 0;
}
})
- take——只发射开始的N项数据或者一定时间内的数据
Observable.just(1,2,3,4)
.take(3)//发射前三个数据项
.take(100, TimeUnit.MILLISECONDS)//发射100ms内的数据
.takeLast(3) //只发射最后的N项数据或者一定时间内的数据
- first——只发射第一项（或者满足某个条件的第一项）数据
- distinct——过滤重复数据
Observable.just(1,2,3,4,3,5,3).distinct().subscribe(item->Log.d("rxjava",item.toString())); //1,2,3,4,5
distinctUntilChanged——过滤掉连续重复的数据
Observable.just(1,2,3,4,5,3,3).distinctUntilChanged().subscribe(item->Log.d("rxjava",item.toString())); //1,2,3,4,5,3
- ofType——过滤指定类型的数据
Observable.just(1,2,"3").ofType(Integer.class).subscribe(item -> Log.d("rxjava",item.toString()));
- skip——跳过开始的N项数据或者一定时间内的数据
Observable.just(2,3,4,5)
.skip(1)
.subscribe(integer -> Log.d("rxjava",integer.toString()));//3,4,5
- Observable的组合操作符
- concat——按顺序连接多个Observables。Observable.concat(a,b)等价于a.concatWith(b)。
Observable<Integer> observable1=Observable.just(1,2,3,4);
Observable<Integer> observable2=Observable.just(4,5,6);
Observable.concat(observable1,observable2).subscribe(item->Log.d("rxjava",item.toString()));//1,2,3,4,4,5,6
- startWith——在数据序列的开头增加一项数据。startWith的内部也是调用了concat
Observable.just(1,2,3,4,5)
.startWith(6,7,8)
.subscribe(item->Log.d("JG",item.toString()));//6,7,8,1,2,3,4,5
merge——按照时间线来连接
zip——以最少的Observable为标准进行压合
Observable<Integer> observable1=Observable.just(1,2,3,4);
Observable<Integer> observable2=Observable.just(4,5,6);
Observable.zip(observable1, observable2, new BiFunction<Integer, Integer,String>() {
@Override
public String apply(@NonNull Integer integer, @NonNull Integer integer2) throws Exception {
return integer+"and"+integer2; //1and4,2and5,3and6
}
});
-
Observable的转换操作符(
toList，toMap)
- map——类型转换
.map(new Function<String, String>() {
@Override
public String apply(@NonNull String s) throws Exception {
return s.toUpperCase();//转成大写字母
}
})
.toList() //转成List
.map(new Function<List<String>, List<String>>() {
@Override
public List<String> apply(@NonNull List<String> strings) throws Exception {
Collections.reverse(strings);
return strings;
}
})
** flatmap**——把Observable产生的结果转换成多个Observable，然后把这多个Observable“扁平化”成一个Observable，并依次提交产生的结果给订阅者
http://blog.csdn.net/qq_35064774/article/details/53057359switchMap——与flatMap操作符不同的是，switchMap操作符会保存最新的Observable产生的结果而舍弃旧的结果
buffer——定期收集Observable的数据放进一个数据包裹，然后发射这些数据包裹，而不是一次发射一个值
Observable.range(1, 5)
.buffer(2)//一次订阅2个(int count)
//.buffer(5, 1)//每次剔除一个(int count, int skip)
//.buffer(2, TimeUnit.SECONDS)//每隔2秒取出消息(long timespan, TimeUnit unit)
.subscribe(new Observer<List<Integer>>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(List<Integer> integers) {
//当buffer(2)： 依次输出[1, 2] 、[3, 4] 、 [5]
//当buffer(5, 1)： 依次输出[1,2,3,4,5]、[2,3,4,5]、[3,4,5][4,5]、[5]
//当buffer(2, TimeUnit.SECONDS)：[1,2,3,4,5]
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
- toSortedList—— 收集原始Observable发射的所有数据到一个有序列表，然后返回这个列表。
Observable.just(6,2,3,4,5)
.toSortedList(new Comparator<Integer>() {
@Override
public int compare(Integer integer, Integer t1) {//自定义排序
return integer-t1; //>0 升序 ，<0 降序
}
})
.subscribe(new Consumer<List<Integer>>() {
@Override
public void accept(@NonNull List<Integer> integers) throws Exception {
Log.i("rxjava", integers.toString()); //[2,3,4,5,6]
}
});
-
cast——在发射之前强制将Observable发射的所有数据转换为指定类型
debounce——延时多少秒后执行
- 其他工具集
- delay
.delay(1, TimeUnit.SECONDS)
-
doOnCompleted——注册一个动作，对正常完成的Observable使用
doOnTerminate——注册一个动作，对完成的Observable使用，无论是否发生错误 - timestamp——加上时间戳
Observable.just(3, 4, 5, 6).timestamp().subscribe(new Consumer<Timed<Integer>>() {
@Override
public void accept(@NonNull Timed<Integer> integerTimed) throws Exception {
Log.i("rxjava","time="+integerTimed.time()+",value="+integerTimed.value());
// Log.i("rxjava","time=1500357633032,value=3);
}
}) ;
