一.buffer操作符
buffer操作符周期性地收集源Observable產(chǎn)生的結(jié)果到列表中，并把這個(gè)列表提交給訂閱者，訂閱者處理后，清空buffer列表，同時(shí)接收下一次收集的結(jié)果并提交給訂閱者，周而復(fù)始。需要注意的是，一旦源Observable在產(chǎn)生結(jié)果的過(guò)程中出現(xiàn)異常，即使buffer已經(jīng)存在收集到的結(jié)果，訂閱者也會(huì)馬上收到這個(gè)異常，并結(jié)束整個(gè)過(guò)程。
//定義郵件內(nèi)容
final String[] mails = new String[]{"Here is an email!", "Another email!", "Yet another email!"};
//每隔1秒就隨機(jī)發(fā)布一封郵件
Observable<String> endlessMail = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
try {
if (subscriber.isUnsubscribed()) return;
Random random = new Random();
while (true) {
String mail = mails[random.nextInt(mails.length)];
subscriber.onNext(mail);
Thread.sleep(1000); }
} catch (Exception ex) {
subscriber.onError(ex);
}
}
}).subscribeOn(Schedulers.io());
//把上面產(chǎn)生的郵件內(nèi)容緩存到列表中，并每隔3秒通知訂閱者
endlessMail.buffer(3, TimeUnit.SECONDS).subscribe(new Action1<List<String>>() {
@Override
public void call(List<String> list) {
System.out.println(String.format("You've got %d new messages! Here they are!", list.size()));
for (int i = 0; i < list.size(); i++)
System.out.println(list.get(i).toString());
}
});
打印出的結(jié)果：
You’ve got 3 new messages! Here they are!(after 3s)
Here is an email!
Another email!
Another email!
You’ve got 3 new messages! Here they are!(after 6s)
Here is an email!
Another email!
Here is an email!
……
二.flatMap操作符
flatMap操作符是把Observable產(chǎn)生的結(jié)果轉(zhuǎn)換成多個(gè)Observable，然后把這多個(gè)Observable“扁平化”成一個(gè)Observable，并依次提交產(chǎn)生的結(jié)果給訂閱者。flatMap操作符通過(guò)傳入一個(gè)函數(shù)作為參數(shù)轉(zhuǎn)換源Observable，在這個(gè)函數(shù)中，你可以自定義轉(zhuǎn)換規(guī)則，最后在這個(gè)函數(shù)中返回一個(gè)新的Observable，然后flatMap操作符通過(guò)合并這些Observable結(jié)果成一個(gè)Observable，并依次提交結(jié)果給訂閱者。
private Observable<File> listFiles(File f){
if(f.isDirectory()){
return Observable.from(f.listFiles())
.flatMap(new Func1<File, Observable<File>>() {
@Override
public Observable<File> call(File file) {
return listFiles(f);
}
});
} else {
return Observable.just(f);
}
}
@Override
public void onClick(View v) {
Observable.just(getApplicationContext().getExternalCacheDir())
.flatMap(new Func1<File, Observable<File>>() {
@Override
public Observable<File> call(File file) {
//參數(shù)file是just操作符產(chǎn)生的結(jié)果，這里判斷file是不是目錄文件，如果是目錄文件，則遞歸查找其子文件flatMap操作符神奇的地方在于，
//返回的結(jié)果還是一個(gè)Observable，而這個(gè)Observable其實(shí)是包含多個(gè)文件的Observable的，輸出應(yīng)該是ExternalCacheDir下的所有文件
return listFiles(file); }
}) .subscribe(new Action1<File>() {
@Override
public void call(File file) {
System.out.println(file.getAbsolutePath());
}
});
}
三.concatMap操作符
cancatMap操作符與flatMap操作符類(lèi)似，都是把Observable產(chǎn)生的結(jié)果轉(zhuǎn)換成多個(gè)Observable，然后把這多個(gè)Observable“扁平化”成一個(gè)Observable，并依次提交產(chǎn)生的結(jié)果給訂閱者。與flatMap操作符不同的是，concatMap操作符在處理產(chǎn)生的Observable時(shí)，采用的是“連接(concat)”的方式，而不是“合并(merge)”的方式，這就能保證產(chǎn)生結(jié)果的順序性，也就是說(shuō)提交給訂閱者的結(jié)果是按照順序提交的，不會(huì)存在交叉的情況。
四.switchMap操作符
switchMap操作符與flatMap操作符類(lèi)似，都是把Observable產(chǎn)生的結(jié)果轉(zhuǎn)換成多個(gè)Observable，然后把這多個(gè)Observable“扁平化”成一個(gè)Observable，并依次提交產(chǎn)生的結(jié)果給訂閱者。與flatMap操作符不同的是，switchMap操作符會(huì)保存最新的Observable產(chǎn)生的結(jié)果而舍棄舊的結(jié)果，舉個(gè)例子來(lái)說(shuō)，比如源Observable產(chǎn)生A、B、C三個(gè)結(jié)果，通過(guò)switchMap的自定義映射規(guī)則，映射后應(yīng)該會(huì)產(chǎn)生A1、A2、B1、B2、C1、C2，但是在產(chǎn)生B2的同時(shí)，C1已經(jīng)產(chǎn)生了，這樣最后的結(jié)果就變成A1、A2、B1、C1、C2，B2被舍棄掉了！
//flatMap操作符的運(yùn)行結(jié)果
Observable.just(10, 20, 30)
.flatMap(new Func1<Integer, Observable<Integer>>() {
@Override
public Observable<Integer> call(Integer integer) {
//10的延遲執(zhí)行時(shí)間為200毫秒、20和30的延遲執(zhí)行時(shí)間為180毫秒
int delay = 200;
if (integer > 10) delay = 180;
return Observable.from(new Integer[]{integer, integer / 2})
.delay(delay, TimeUnit.MILLISECONDS);
} })
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
System.out.println("flatMap Next:" + integer);
} });
//concatMap操作符的運(yùn)行結(jié)果
Observable.just(10, 20, 30)
.concatMap(new Func1<Integer, Observable<Integer>>() {
@Override
public Observable<Integer> call(Integer integer) {
//10的延遲執(zhí)行時(shí)間為200毫秒、20和30的延遲執(zhí)行時(shí)間為180毫秒
int delay = 200;
if (integer > 10) delay = 180;
return Observable.from(new Integer[]{integer, integer / 2})
.delay(delay, TimeUnit.MILLISECONDS); } })
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
System.out.println("concatMap Next:" + integer);
} });
//switchMap操作符的運(yùn)行結(jié)果
Observable.just(10, 20, 30)
.switchMap(new Func1<Integer, Observable<Integer>>() {
@Override
public Observable<Integer> call(Integer integer) {
//10的延遲執(zhí)行時(shí)間為200毫秒、20和30的延遲執(zhí)行時(shí)間為180毫秒
int delay = 200;
if (integer > 10) delay = 180;
return Observable.from(new Integer[]{integer, integer / 2})
.delay(delay, TimeUnit.MILLISECONDS);
} })
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<Integer>() {
@Override
public void call(Integer integer) {
System.out.println("switchMap Next:" + integer); } });
運(yùn)行結(jié)果如下：
flatMap Next:20
flatMap Next:10
flatMap Next:30
flatMap Next:15
flatMap Next:10
flatMap Next:5
switchMap Next:30
switchMap Next:15
concatMap Next:10
concatMap Next:5
concatMap Next:20
concatMap Next:10
concatMap Next:30
concatMap Next:15
五.groupBy操作符
groupBy操作符是對(duì)源Observable產(chǎn)生的結(jié)果進(jìn)行分組，形成一個(gè)類(lèi)型為GroupedObservable的結(jié)果集，GroupedObservable中存在一個(gè)方法為getKey()，可以通過(guò)該方法獲取結(jié)果集的Key值（類(lèi)似于HashMap的key)。值得注意的是，由于結(jié)果集中的GroupedObservable是把分組結(jié)果緩存起來(lái)，如果對(duì)每一個(gè)GroupedObservable不進(jìn)行處理（既不訂閱執(zhí)行也不對(duì)其進(jìn)行別的操作符運(yùn)算），就有可能出現(xiàn)內(nèi)存泄露。因此，如果你對(duì)某個(gè)GroupedObservable不進(jìn)行處理，最好是對(duì)其使用操作符take(0)處理。
