RxJava 转换操作符 flatMap

flatMap操作符是把Observable产生的结果转换成多个Observable,然后又直接把这些Observable转换一个Observable,并依次提交产生的结果给订阅者

规律:源Observable 传入flatMap对应参数，在flatMap中返回自定义的Observable

例子:扫描文件:

private void flatMap() {
Observable.just(getApplicationContext().getExternalCacheDir()).flatMap(new Func1<File, Observable<File>>() {
@Override
public Observable<File> call(File file) {
return listFiles(file);
}
}).subscribe(new Action1<File>() {
@Override
public void call(File file) {
LogUtils.d("------>:" + file.getAbsolutePath());
}
});
}

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(file);
}
});
} else {
return Observable.just(f);
}
}

扫描结果:

02-24 16:13:02.038 24715-24715/? D/----->: ------>:/storage/sdcard0/Android/data/com.rxandroid.test1/cache/.text

02-24 16:13:02.038 24715-24715/? D/----->: ------>:/storage/sdcard0/Android/data/com.rxandroid.test1/cache/0.text

02-24 16:13:02.038 24715-24715/? D/----->: ------>:/storage/sdcard0/Android/data/com.rxandroid.test1/cache/1.text

02-24 16:13:02.038 24715-24715/? D/----->: ------>:/storage/sdcard0/Android/data/com.rxandroid.test1/cache/2.text

02-24 16:13:02.038 24715-24715/? D/----->: ------>:/storage/sdcard0/Android/data/com.rxandroid.test1/cache/3.text

02-24 16:13:02.038 24715-24715/? D/----->: ------>:/storage/sdcard0/Android/data/com.rxandroid.test1/cache/4.text

02-24 16:13:02.048 24715-24715/? D/----->: ------>:/storage/sdcard0/Android/data/com.rxandroid.test1/cache/5.text

02-24 16:13:02.048 24715-24715/? D/----->: ------>:/storage/sdcard0/Android/data/com.rxandroid.test1/cache/6.text

02-24 16:13:02.048 24715-24715/? D/----->: ------>:/storage/sdcard0/Android/data/com.rxandroid.test1/cache/7.text

02-24 16:13:02.048 24715-24715/? D/----->: ------>:/storage/sdcard0/Android/data/com.rxandroid.test1/cache/8.text

02-24 16:13:02.048 24715-24715/? D/----->: ------>:/storage/sdcard0/Android/data/com.rxandroid.test1/cache/9.text

flatMap和map的区别：map是将参数加工,flatMap是将源Observable变成多个Observable去处理并交给源Observable依次发送给订阅者

注意:这些新的Observable线程是默认伴随源Observable的:

private void flatMap() {
Observable.just(getApplicationContext().getExternalCacheDir()).flatMap(new Func1<File, Observable<File>>() {
@Override
public Observable<File> call(File file) {
LogUtils.d("------>Thread1:" + Thread.currentThread());
return listFiles(file);
}
}).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Action1<File>() {
@Override
public void call(File file) {
LogUtils.d("------>Thread3:" + Thread.currentThread().getName() + "   file:" + file.getAbsolutePath());
}
});
}

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) {
LogUtils.d("------>Thread2:" + Thread.currentThread());
return listFiles(file);
}
});
} else {
return Observable.just(f);
}
}

线程关系:
02-24 16:27:42.118 25657-25835/? D/----->: ------>Thread1:Thread[RxCachedThreadScheduler-2,5,main]

02-24 16:27:42.118 25657-25835/? D/----->: ------>Thread2:Thread[RxCachedThreadScheduler-2,5,main]

02-24 16:27:42.118 25657-25835/? D/----->: ------>Thread2:Thread[RxCachedThreadScheduler-2,5,main]

02-24 16:27:42.118 25657-25835/? D/----->: ------>Thread2:Thread[RxCachedThreadScheduler-2,5,main]

02-24 16:27:42.128 25657-25835/? D/----->: ------>Thread2:Thread[RxCachedThreadScheduler-2,5,main]

02-24 16:27:42.128 25657-25835/? D/----->: ------>Thread2:Thread[RxCachedThreadScheduler-2,5,main]

02-24 16:27:42.128 25657-25835/? D/----->: ------>Thread2:Thread[RxCachedThreadScheduler-2,5,main]

02-24 16:27:42.128 25657-25835/? D/----->: ------>Thread2:Thread[RxCachedThreadScheduler-2,5,main]

02-24 16:27:42.128 25657-25835/? D/----->: ------>Thread2:Thread[RxCachedThreadScheduler-2,5,main]

02-24 16:27:42.138 25657-25835/? D/----->: ------>Thread2:Thread[RxCachedThreadScheduler-2,5,main]

02-24 16:27:42.138 25657-25835/? D/----->: ------>Thread2:Thread[RxCachedThreadScheduler-2,5,main]

02-24 16:27:42.138 25657-25835/? D/----->: ------>Thread2:Thread[RxCachedThreadScheduler-2,5,main]

02-24 16:27:42.138 25657-25657/? D/----->: ------>Thread3:main file:/storage/sdcard0/Android/data/com.rxandroid.test1/cache/.text

02-24 16:27:42.138 25657-25657/? D/----->: ------>Thread3:main file:/storage/sdcard0/Android/data/com.rxandroid.test1/cache/0.text

02-24 16:27:42.138 25657-25657/? D/----->: ------>Thread3:main file:/storage/sdcard0/Android/data/com.rxandroid.test1/cache/1.text

02-24 16:27:42.138 25657-25657/? D/----->: ------>Thread3:main file:/storage/sdcard0/Android/data/com.rxandroid.test1/cache/2.text

02-24 16:27:42.138 25657-25657/? D/----->: ------>Thread3:main file:/storage/sdcard0/Android/data/com.rxandroid.test1/cache/3.text

02-24 16:27:42.148 25657-25657/? D/----->: ------>Thread3:main file:/storage/sdcard0/Android/data/com.rxandroid.test1/cache/4.text

02-24 16:27:42.148 25657-25657/? D/----->: ------>Thread3:main file:/storage/sdcard0/Android/data/com.rxandroid.test1/cache/5.text

02-24 16:27:42.148 25657-25657/? D/----->: ------>Thread3:main file:/storage/sdcard0/Android/data/com.rxandroid.test1/cache/6.text

02-24 16:27:42.148 25657-25657/? D/----->: ------>Thread3:main file:/storage/sdcard0/Android/data/com.rxandroid.test1/cache/7.text

02-24 16:27:42.148 25657-25657/? D/----->: ------>Thread3:main file:/storage/sdcard0/Android/data/com.rxandroid.test1/cache/8.text

02-24 16:27:42.148 25657-25657/? D/----->: ------>Thread3:main file:/storage/sdcard0/Android/data/com.rxandroid.test1/cache/9.text