RxJava（三） flatMap 操作符用法详解

全栈程序员站长

发布于 2022-09-05 15:08:11

1.7K0

发布于 2022-09-05 15:08:11

大家好，又见面了，我是你们的朋友全栈君。

RxJava 系列文章目录导读：

一、RxJava create 操作符的用法和源码分析二、RxJava map 操作符用法详解三、RxJava flatMap 操作符用法详解四、RxJava concatMap 操作符用法详解五、RxJava onErrorResumeNext 操作符实现 app 与服务器间 token 机制六、RxJava retryWhen 操作符实现错误重试机制七、RxJava 使用 debounce 操作符优化 app 搜索功能八、RxJava concat 操作处理多数据源九、RxJava zip 操作符在 Android 中的实际使用场景十、RxJava switchIfEmpty 操作符实现 Android 检查本地缓存逻辑判断十一、RxJava defer 操作符实现代码支持链式调用十二、combineLatest 操作符的高级使用十三、RxJava 导致 Fragment Activity 内存泄漏问题十四、interval、takeWhile 操作符实现获取验证码功能十五、RxJava 线程的自由切换

flatMap 操作符的作用

官方文档解释：

Returns an Observable that emits items based on applying a function that you supply to each item emitted by the source Observable, where that function returns an Observable, and then merging those resulting Observables and emitting the results of this merger.

官方流程图：

对 Observable 发射的数据都应用(apply)一个函数，这个函数返回一个 Observable，然后合并这些 Observables，并且发送（emit）合并的结果。 flatMap 和 map 操作符很相像，flatMap 发送的是合并后的 Observables，map 操作符发送的是应用函数后返回的结果集。

flatMap 操作符使用示例

继续 map 操作符的案例

还是以上一篇map操作符的例子吧，如果对 map操作符不是很了解的，可以看看我之前的文章。

获取主机的IP地址：

   private Observable<String> processUrlIpByOneFlatMap() {
        return Observable.just(
                "http://www.baidu.com/",
                "http://www.google.com/",
                "https://www.bing.com/")
                .flatMap(new Func1<String, Observable<String>>() {
                    @Override
                    public Observable<String> call(String s) {
                        return createIpObservable(s);
                    }
                })
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1<String>() {
                    @Override
                    public void call(String s) {
                        printLog(tvLogs, "Consume Data <- ", s);
                    }
                }, new Action1<Throwable>() {
                    @Override
                    public void call(Throwable throwable) {
                        printErrorLog(tvLogs, "throwable call()", throwable.getMessage());
                    }
                });
    }
    
    //根据主机获取ip
    private Observable<String> createIpObservable(final String url) {
        return Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                try {
                    String ip = getIPByUrl(url);
                    subscriber.onNext(ip);
                    printLog(tvLogs, "Emit Data -> ",url+" : " +ip);
                } catch (MalformedURLException e) {
                    e.printStackTrace();
                    //subscriber.onError(e);
                    subscriber.onNext(null);
                } catch (UnknownHostException e) {
                    e.printStackTrace();
                    //subscriber.onError(e);
                    subscriber.onNext(null);
                }
                subscriber.onCompleted();
            }
        });
    }

输出结果：

Emit Data -> 'http://www.baidu.com/ : 115.239.211.112'
Main Thread:false, Thread Name:RxCachedThreadScheduler-1
Consume Data <- '115.239.211.112'
Main Thread:true, Thread Name:main

Emit Data -> 'http://www.google.com/ : 216.58.199.100'
Main Thread:false, Thread Name:RxCachedThreadScheduler-1
Consume Data <- '216.58.199.100'
Main Thread:true, Thread Name:main

Emit Data -> 'https://www.bing.com/ : 202.89.233.104'
Main Thread:false, Thread Name:RxCachedThreadScheduler-1
Consume Data <- '202.89.233.104'
Main Thread:true, Thread Name:main

flatMap进阶使用

我们从上面的输出结果可以看出，效果和使用 map操作符 的效果是一样。我们同时也发现线程的名称（Thread Name）都是 RxCachedThreadScheduler-1 ，说明他们是通过一个线程来完成所有的任务的。如果任务很多，仅仅通过一个线程去做，效率上是不是有点低呢？如果我想使用多个线程来完成这些任务该怎么做呢？很简单，只需要在创建 Observable 的时候加上subscribeOn(Schedulers.io()) 即可。完整代码如下：

//根据主机获取ip
    private Observable<String> createIpObservable(final String url) {
        return Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                try {
                    String ip = getIPByUrl(url);
                    subscriber.onNext(ip);
                    printLog(tvLogs, "Emit Data -> ",url+" : " +ip);
                } catch (MalformedURLException e) {
                    e.printStackTrace();
                    //subscriber.onError(e);
                    subscriber.onNext(null);
                } catch (UnknownHostException e) {
                    e.printStackTrace();
                    //subscriber.onError(e);
                    subscriber.onNext(null);
                }
                subscriber.onCompleted();
            }
        })
        .subscribeOn(Schedulers.io());
    }

看下运行效果：

Consume Data <- '202.89.233.103'
Main Thread:true, Thread Name:main
Emit Data -> 'https://www.bing.com/ : 202.89.233.103'
Main Thread:false, Thread Name:RxCachedThreadScheduler-8

Emit Data -> 'http://www.google.com/ : 216.58.203.36'
Main Thread:false, Thread Name:RxCachedThreadScheduler-7
Consume Data <- '216.58.203.36'
Main Thread:true, Thread Name:main
 
Emit Data -> 'http://www.baidu.com/ : 115.239.211.112'
Main Thread:false, Thread Name:RxCachedThreadScheduler-6
Consume Data <- '115.239.211.112'
Main Thread:true, Thread Name:main

从运行可以看出，执行完成任务的不是一个线程了，而是三个不同的线程 RxCachedThreadScheduler-8 、RxCachedThreadScheduler-7、RxCachedThreadScheduler-6 。

但是发现一个问题，输出的结果的顺序乱了，不是我们输入的 baidu.com、google.com、bing.com 顺序了。

那怎么办呢？

这时候 concatMap 操作符就闪亮登场了，下一篇将介绍 concatMap 操作符的用法。