RxJava中map和flatMap的用法区别源码解析

更新时间：2022年09月21日 16:35:14 作者：白瑞德

这篇文章主要为大家介绍了RxJava中map和flatMap的用法区别源码解析，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪

前言：

RxJava中提供了大量的操作符，这大大提高了了我们的开发效率。其中最基本的两个变换操作符就是map和flatMap。而其他变换操作符的原理基本与map类似。

map和flatMap都是接受一个函数作为参数(Func1)并返回一个被观察者Observable
Func1的< I,O >I,O模版分别为输入和输出值的类型，实现Func1的call方法对I类型进行处理后返回O类型数据，只是flatMap中执行的方法的返回类型为Observable类型

作用

map对Observable发射的每一项数据应用一个函数，执行变换操作。对原始的Observable发射的每一项数据应用一个你选择的函数，然后返回一个发射这些结果的Observable。

flatMap将一个发射数据的Observable变换为多个Observables，然后将它们发射的数据合并后放进一个单独的Observable。操作符使用一个指定的函数对原始Observable发射的每一项数据执行变换操作，这个函数返回一个本身也发射数据的Observable，然后FlatMap合并这些Observables发射的数据，最后将合并后的结果当做它自己的数据序列发射

使用方法：

通过代码来看一下两者的使用用方法：

map

Observable.just(new User("白瑞德"))
                  .map(new Function<User, String>() {
                      @Override
                      public String apply(User user) throws Throwable {
                          return user.getName();
                      }
                  })
                  .subscribe(new Consumer<String>() {
                      @Override
                      public void accept(String s) throws Throwable {
                          System.out.println(s);
                      }
                  });
<<<白瑞德

这段代码接受一个User对象，最后打印出User中的name。

flatMap

假设存在一个需求，图书馆要打印每个User借走每一本书的名字： User结构如下：

class User {
    private String       name;
    private List<String> book;
}

我们来看一下map的实现方法：

Observable.fromIterable(userList)
                  .map(new Function<User, List<String>>() {
                      @Override
                      public List<String> apply(User user) throws Throwable {
                          return user.getBook();
                      }
                  })
                  .subscribe(new Consumer<List<String>>() {
                      @Override
                      public void accept(List<String> strings) throws Throwable {
                          for (String s : strings) {
                              System.out.println(s);
                          }
                      }
                  });

可以看到，map的转换总是一对一，只能单一转换。我们不得不借助循环进行打印。下面我们来看一下flatMap的实现方式：

Observable.fromIterable(userList)
                  .flatMap(new Function<User, ObservableSource<String>>() {
                      @Override
                      public ObservableSource<String> apply(User user) throws Throwable {
                          return Observable.fromIterable(user.getBook());
                      }
                  })
                  .subscribe(new Consumer<String>() {
                      @Override
                      public void accept(String  o) throws Throwable {
                          System.out.println(o);
                      }
                  });

flatmap既可以单一转换也可以一对多/多对多转换。flatMap使用一个指定的函数对原始Observable发射的每一项数据之行相应的变换操作，这个函数返回一个本身也发射数据的Observable，因此可以再内部再次进行事件的分发。然后flatMap合并这些Observables发射的数据，最后将合并后的结果当做它自己的数据序列发射。

源码分析

下面我们就结合源码来分析一下这两个操作符。为了降低代码阅读难道，这里只保留核心代码：

map

public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
    //接受一个Function实例，并返回一个ObservableMap
    return new ObservableMap<T, R>(this, mapper);
}
public final class ObservableMap<T, U> extends AbstractObservableWithUpstream<T, U> {
    final Function<? super T, ? extends U> function;
    public ObservableMap(ObservableSource<T> source, Function<? super T, ? extends U> function) {
        //调用用父类构造方法，初始化父类中的downstream
        super(source);
        this.function = function;
    }
    @Override
    public void subscribeActual(Observer<? super U> t) {
        source.subscribe(new MapObserver<T, U>(t, function));
    }
    static final class MapObserver<T, U> extends BasicFuseableObserver<T, U> {
        final Function<? super T, ? extends U> mapper;
        MapObserver(Observer<? super U> actual, Function<? super T, ? extends U> mapper) {
            super(actual);
            this.mapper = mapper;
        }
        @Override
        public void onNext(T t) {
            v = mapper.apply(t);
            downstream.onNext(v);
        }
    }
}

这段代码是去掉map源码中一些校验和其它相关回调后的精简代码。接下来分析一下代码流程：

当在调用map时，map接受一个匿名内部类Function的实例，并返回一个ObservableMap对象。
ObservableMap本质上是一个Observable，也是一个被观察者，其构造方法接受最外层的那个被Observable实例，和Function实例。ObservableMap重写了subscribeActual方法，在subscribeActual中使用新建了一个MapObserver实现了对原始Observable的观察。
原始的Observable中的数据变会被发送到MapObserver的实例中。
MapObserver构造方法接收原始Observable的观察者actual，和Function的实例mapper
MapObserver在其onNext方法中调用mapper的apply方法，获得该方法的返回值v apply方法就是map实例中： public String apply(User user) throws Throwable { return user.getName(); }
调用downstream的onNext方法，并传入实参v。其中downstream是MapObserver父类中定义的变量，在MapObserver构造方法中super(actual);时初始化，其本身就是传入的actual，本质上就是最原始的Observable

整个流程可以概括如下：存在一个原始的ObservableA和一个观察者ObserverA，当原始的被观察者ObservableA调用map,并传入一个匿名内部类实例化的’function‘，map新建并返回了一个被观察者ObservableB，通过subscribe让观察者ObserverA对其进行订阅。并重写subscribeActual方法，在其被订阅时创建一个新的观察者ObserverB其接受的，并用ObserverB对原始的ObservableA进行订阅观察。当原始的ObservableA发出事件，调用ObserverB的onNext方法，subscribeActual接受的观察者便是最原始的观察者ObserverA。ObserverB变执行通过匿名内部类实例化的’function‘的apply方法得到数据v，紧接着调用原始的ObservableA的onNext方法，并传入实参v，ObserverA观察到事件。一句话概括：一个原始的被观察者和观察者，但是让原始的观察者去订阅一个新的观察者，当新的被观察者被订阅的时候，创建一个新的观察者去订阅原始的被观察者，并在监听的事件之后执行指定的操作后再通知原始观察者。所以这里面一共涉及到两对观察者和被观察者，map方法会创建一对新的观察者和被观察者作为原始观察者和被观察者通讯的纽带，并在其中做一些数据变换。

用图片显示流程如下：

蓝色框内就是map创建的观察者和被观察者。实际上我们的原始ObserverA并没有对ObservableA进行订阅。

flatMap

faltMap和map的基本原理类似，其代码如下：

public final <R> Observable<R> flatMap(Function<? super T, ? extends ObservableSource<? extends R>> mapper) {
        return new ObservableFlatMap<T, R>(this, mapper, delayErrors, maxConcurrency, bufferSize);
}
public final class ObservableFlatMap<T, U> extends AbstractObservableWithUpstream<T, U> {
    final Function<? super T, ? extends ObservableSource<? extends U>> mapper;
    final boolean delayErrors;
    final int maxConcurrency;
    final int bufferSize;
    public ObservableFlatMap(ObservableSource<T> source,
            Function<? super T, ? extends ObservableSource<? extends U>> mapper,
            boolean delayErrors, int maxConcurrency, int bufferSize) {
        super(source);
    }
    @Override
    public void subscribeActual(Observer<? super U> t) {
        source.subscribe(new MergeObserver<T, U>(t, mapper, delayErrors, maxConcurrency, bufferSize));
    }
    static final class MergeObserver<T, U> extends AtomicInteger implements Disposable, Observer<T> {
        MergeObserver(Observer<? super U> actual, Function<? super T, ? extends ObservableSource<? extends U>> mapper,
                boolean delayErrors, int maxConcurrency, int bufferSize) {
            ...   
            this.observers = new AtomicReference<InnerObserver<?, ?>[]>(EMPTY);
        }
        @Override
        public void onSubscribe(Disposable d) {
            downstream.onSubscribe(this);
        }
        @Override
        public void onNext(T t) {
            ObservableSource<? extends U> p;
            p = mapper.apply(t);    
            subscribeInner(p);
        }
        @SuppressWarnings("unchecked")
        void subscribeInner(ObservableSource<? extends U> p) {
                InnerObserver<T, U> inner = new InnerObserver<T, U>(this, uniqueId++);
                p.subscribe(inner);
        }
        void drain() {    
            drainLoop();
        }
        void drainLoop() {
            final Observer<? super U> child = this.downstream;
            child.onNext(o);
        }
    }
    static final class InnerObserver<T, U> extends AtomicReference<Disposable>
    implements Observer<U> {
        private static final long serialVersionUID = -4606175640614850599L;
        final long id;
        final MergeObserver<T, U> parent;
        volatile boolean done;
        volatile SimpleQueue<U> queue;
        int fusionMode;
        InnerObserver(MergeObserver<T, U> parent, long id) {
            this.id = id;
            this.parent = parent;
        }
        @Override
        public void onNext(U t) {
            parent.drain();
        }
    }
}

上述代码即是faltMap精简后的源码，其中大部分代码的运作流程和前文中的map源码一致，我们继续延续上文中讲解中的观察者和被观察者。重点关注其不同的地方： faltMap返回一个新的被观察者ObservableB，重写ObservableB的subscribeActual方法在原始的观察者ObserverA对其进行订阅时新建一个观察者ObserverB对原始的ObservableA进行订阅。新的观察者ObserverB持有原始的ObserverA和faltMap接收的匿名对象实例function。当ObserverB监听到原始的被观察者ObservableA的事件时，ObserverB调用function的apply方法获得新新的被观察者ObservableC，再创建一个新的观察者ObserverC对ObservableC进行订阅，ObserverC持有原始的观察者ObserverA，在ObserverC观察到被观察者ObservableC的时间时，调用原始的观察者ObserverA的方法。

概括就是：faltMap方法要求调用者提供一个Observable，最原始的Observable在调用faltMap后，faltMap会创建一个新的Observable，并对原始的进行订阅。当拿到订阅后，会通过flatMap接收的函数拿到调用者传入的Observable，并用最原始的观察者对它进行订阅。这期间涉及三对观察者和被观察者，flatMap会创建一对，同时也接收一对用户创建的。flatMap创建的和Map中的作用一样，不过flatMap连接的是原始的和用户通过flatMap提供的两对观察者和被观察者。而原始的观察者最终是对用户通过flatMap提供的那个观察者进行订阅。

用图片显示流程如下：

和Map的流程很相似，只不过是需要用户再提供一对观察者和被观察者。最终实现对用户提供的被观察者进行订阅。

结语

至此，map和flatMap已基本分析完毕，其中map的代码比较简单易懂，flatMap中还涉及到大量辅助操作，文中并未涉及到其中的合并等操作，阅读起来有些困难。如果仅仅是为了了解二者的原理，可以阅读Single<T>中的代码。其中的代码量远远少于Observable中的代码量。如果对RxJava基本的模式还不了解，可以阅读手写极简版的Rxjava

以上就是RxJava中map和flatMap的用法区别源码解析的详细内容，更多关于RxJava map flatMap区别的资料请关注脚本之家其它相关文章！

您可能感兴趣的文章:

深入解析SpringBatch适配器
Spring Batch是Spring的一个子项目,使用Java语言并基于Spring框架为基础开发,使得已经使用 Spring 框架的开发者或者企业更容易访问和利用企业服务，本文给大家介绍SpringBatch适配器的相关知识，感兴趣的朋友一起看看吧
2021-11-11
RabbitMQ进阶之消息可靠性详解
这篇文章主要介绍了RabbitMQ进阶之消息可靠性详解,abbitmq消息的投递过程中，怎么确保消息能不丢失，这是一个很重要的问题,哪怕我们做了Rabbitmq持久化，也不能保证我们的业务消息不会被丢失,需要的朋友可以参考下
2023-08-08
使用Java8中Optional机制的正确姿势
我们知道 Java 8 增加了一些很有用的 API, 其中一个就是 Optional，下面这篇文章主要给大家介绍了关于如何正确使用Java8中Optional机制的相关资料，文中通过示例代码介绍的非常详细，需要的朋友可以参考借鉴，下面来一起看看吧。
2017-11-11
Springboot实现阿里云通信短信服务有关短信验证码的发送功能
短信验证码是通过发送验证码到手机的一种有效的验证码系统。主要用于验证用户手机的合法性及敏感操作的身份验证。下面通过本文大家分享Springboot实现阿里云通信短信服务有关短信验证码的发送功能，一起看看吧
2017-08-08
Kotlin基础教程之操作符与操作符重载
这篇文章主要介绍了Kotlin基础教程之操作符与操作符重载的相关资料,需要的朋友可以参考下
2017-05-05
FastJson时间格式化问题避坑经验分享
这篇文章主要为大家介绍了FastJson时间格式化问题避坑经验分享，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪
2022-08-08
javaCV开发详解之推流器和录制器的实现
这篇文章主要介绍了javaCV开发详解之推流器和录制器实现，对JavaCV感兴趣的同学，可以参考下
2021-04-04
详解@Autowired(required=false)注入注意的问题
这篇文章主要介绍了@Autowired(required=false)注入注意的问题,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2020-04-04
Spring Cloud Alibaba教程之Sentinel的使用
这篇文章主要介绍了Spring Cloud Alibaba教程之Sentinel的使用，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2019-09-09
webservice实现springboot项目间接口调用与对象传递示例
本文主要介绍了webservice实现springboot项目间接口调用与对象传递示例，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2022-07-07