- Observable的創(chuàng)建操作符
- 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的每個(gè)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操作符——延時(shí)幾秒呜袁，每隔幾秒開始執(zhí)行
Flowable.interval(1, TimeUnit.SECONDS)
Flowable.interval(0, 1, TimeUnit.SECONDS)////參數(shù)一：延時(shí)0秒敌买；參數(shù)二：每隔1秒開始執(zhí)行
- timer——延時(shí)之后發(fā)射數(shù)據(jù)項(xiàng)為0的Observable<Long>
Observable.timer(1000, TimeUnit.MILLISECONDS)
.subscribe(new Consumer<Long>() {
@Override
public void accept(@NonNull Long aLong) throws Exception {
}
});
- range——創(chuàng)建從n開始，個(gè)數(shù)為m的連續(xù)數(shù)字傅寡，比如range(3,4)放妈，就是創(chuàng)建3、4荐操、5芜抒、6的一組數(shù)字
- Observable的聚合操作符
- reduce—— 發(fā)射最終的結(jié)果
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發(fā)射的每一項(xiàng)數(shù)據(jù)應(yīng)用一個(gè)函數(shù)托启，然后按順序依次發(fā)射每一個(gè)值
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——收集數(shù)據(jù)到一個(gè)可變的數(shù)據(jù)結(jié)構(gòu)
Observable.just(1,2,3,4)
.collect(new Callable<List<Integer>>() { //創(chuàng)建數(shù)據(jù)結(jié)構(gòu)
@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——當(dāng)發(fā)射的數(shù)據(jù)滿足某個(gè)條件（包含該數(shù)據(jù)）后終止
takeWhile—— 當(dāng)發(fā)射的數(shù)據(jù)滿足某個(gè)條件（不含該數(shù)據(jù)）后終止
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—— 判斷兩個(gè)Observable發(fā)射的數(shù)據(jù)是否相同（數(shù)據(jù)宅倒，發(fā)射順序）
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——過濾數(shù)據(jù)
.filter(new Predicate<String>() {
@Override
public boolean test(String s) throws Exception {
//true為繼續(xù)執(zhí)行下去，false攔截
return s.trim().length() > 0;
}
})
- take——只發(fā)射開始的N項(xiàng)數(shù)據(jù)或者一定時(shí)間內(nèi)的數(shù)據(jù)
Observable.just(1,2,3,4)
.take(3)//發(fā)射前三個(gè)數(shù)據(jù)項(xiàng)
.take(100, TimeUnit.MILLISECONDS)//發(fā)射100ms內(nèi)的數(shù)據(jù)
.takeLast(3) //只發(fā)射最后的N項(xiàng)數(shù)據(jù)或者一定時(shí)間內(nèi)的數(shù)據(jù)
- first——只發(fā)射第一項(xiàng)（或者滿足某個(gè)條件的第一項(xiàng)）數(shù)據(jù)
- distinct——過濾重復(fù)數(shù)據(jù)
Observable.just(1,2,3,4,3,5,3).distinct().subscribe(item->Log.d("rxjava",item.toString())); //1,2,3,4,5
distinctUntilChanged——過濾掉連續(xù)重復(fù)的數(shù)據(jù)
Observable.just(1,2,3,4,5,3,3).distinctUntilChanged().subscribe(item->Log.d("rxjava",item.toString())); //1,2,3,4,5,3
- ofType——過濾指定類型的數(shù)據(jù)
Observable.just(1,2,"3").ofType(Integer.class).subscribe(item -> Log.d("rxjava",item.toString()));
- skip——跳過開始的N項(xiàng)數(shù)據(jù)或者一定時(shí)間內(nèi)的數(shù)據(jù)
Observable.just(2,3,4,5)
.skip(1)
.subscribe(integer -> Log.d("rxjava",integer.toString()));//3,4,5
- Observable的組合操作符
- concat——按順序連接多個(gè)Observables屯耸。Observable.concat(a,b)等價(jià)于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——在數(shù)據(jù)序列的開頭增加一項(xiàng)數(shù)據(jù)。startWith的內(nèi)部也是調(diào)用了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——按照時(shí)間線來連接
zip——以最少的Observable為標(biāo)準(zhǔn)進(jìn)行壓合
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的轉(zhuǎn)換操作符(
toList疗绣，toMap)
- map——類型轉(zhuǎn)換
.map(new Function<String, String>() {
@Override
public String apply(@NonNull String s) throws Exception {
return s.toUpperCase();//轉(zhuǎn)成大寫字母
}
})
.toList() //轉(zhuǎn)成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產(chǎn)生的結(jié)果轉(zhuǎn)換成多個(gè)Observable线召，然后把這多個(gè)Observable“扁平化”成一個(gè)Observable，并依次提交產(chǎn)生的結(jié)果給訂閱者
http://blog.csdn.net/qq_35064774/article/details/53057359switchMap——與flatMap操作符不同的是多矮，switchMap操作符會保存最新的Observable產(chǎn)生的結(jié)果而舍棄舊的結(jié)果
buffer——定期收集Observable的數(shù)據(jù)放進(jìn)一個(gè)數(shù)據(jù)包裹缓淹，然后發(fā)射這些數(shù)據(jù)包裹哈打，而不是一次發(fā)射一個(gè)值
Observable.range(1, 5)
.buffer(2)//一次訂閱2個(gè)(int count)
//.buffer(5, 1)//每次剔除一個(gè)(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) {
//當(dāng)buffer(2)： 依次輸出[1, 2] 、[3, 4] 讯壶、 [5]
//當(dāng)buffer(5, 1)： 依次輸出[1,2,3,4,5]料仗、[2,3,4,5]、[3,4,5][4,5]伏蚊、[5]
//當(dāng)buffer(2, TimeUnit.SECONDS)：[1,2,3,4,5]
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
- toSortedList—— 收集原始Observable發(fā)射的所有數(shù)據(jù)到一個(gè)有序列表立轧，然后返回這個(gè)列表。
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——在發(fā)射之前強(qiáng)制將Observable發(fā)射的所有數(shù)據(jù)轉(zhuǎn)換為指定類型
debounce——延時(shí)多少秒后執(zhí)行
- 其他工具集
- delay
.delay(1, TimeUnit.SECONDS)
-
doOnCompleted——注冊一個(gè)動作氛改，對正常完成的Observable使用
doOnTerminate——注冊一個(gè)動作，對完成的Observable使用颜阐，無論是否發(fā)生錯(cuò)誤 - timestamp——加上時(shí)間戳
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);
}
}) ;
