RxJava2源码解读之 Map、FlatMap

RxJava给我们提供了很多变换的操作符，map、flatMap就是比较常用的操作符，一般我们使用的时候，都是看官方文档来了解每个操作符的含义，但是我自己感觉下来，看官方文档使用没问题，但是总有一点隔靴搔痒的意思，所以我还要去RxJava的源码一探究竟，做到心中有数。

我们先从相对简单的 Map 开始

Map

官方定义：transform the items emitted by an Observable by applying a function to each item 拙劣的翻译：应用一个函数 转换所有的被发射的item

官方的图解：

map 图例.png

到这里我们总结一下：

• map 转换是一对一的，原来发射了几个数据，转换之后还是几个
• map 转换可以改变发射的数据类型

这里抛出一个问题，map 调用我们提供的function进行转换，那么这个function在什么时候被调用？在哪个线程被调用？（这个对我们实际工程中使用map有意义，知道代码被执行的线程是必须的）

废话不多说，进入源码

Map源码

Observable类是RxJava的门面，基本上所有的转换符都在这里定义，直接看Map 的方法定义

map 方法.png

可以看到，Function类，泛型有2个参数，第一个是原数据类型，第二个是转换后的数据类型，最终返回的是ObservableMap 类（RxJava的类命名很规范，如果是Observable类型的就是Observable开头 + 具体的操作符名称，如果是Observer类型的 就是 具体的操作符名称 + Observer结尾）我们进入ObservableMap类，Observable类之前的文章有提到过，subscribeActual 是个重要的钩子方法，所以我们直接看ObservableMap如何重写该方法的

ObservaleMap.png

方法代码就一行，调用装饰的Observable的subscribe方法，传递一个MapObserver对象，Observer类我们就比较熟悉了，我们这里主要看onNext方法

map onnext.png

代码也很简单，红框标识的就是 mapper 转换函数被调用的地方，得到转换后的对象v，传递给被装饰的Observer 的onNext方法，到这里，一次数据的map转换就结束了。源码的实现还是很简单的，在我们了解了源码的实现后，思路会更清晰，写代码时也会更有把握。

现在我们来解答前面我们抛出的问题，Function在什么时候被调用？在哪个线程被调用？ Function调用的地方已经清楚了，在ObserverMap 的 onNext方法中，那么调用的线程呢，因为是在Observer方法中被调用，所以如果在map 之前 调用了 ObserverOn 方法设置监听线程，那么就在该监听线程，如果没有设置 ObserverOn 但是设置了 SubscribeOn方法设置发射线程，那么就在该 发射线程，如果SubscribeOn也没有设置，那就在Observable的创建线程。

到此Map 就介绍完了，接下来是Map 的好兄弟 FlatMap，调用逻辑稍微复杂一点点，看官们耐心 -。-

FlatMap

官方定义：transform the items emitted by an Observable into Observables, then flatten the emissions from those into a single Observable 拙劣的翻译：应用一个函数 转换所有的被发射的item从一个Observable转成为多个Observable，并将所有要发射的数据平铺为一个Observable

官方的图解:

flatmap 图例.png

到这里我们总结一下：

• flatmap 转换是一对多的(一对一当然也支持)，原来发射了几个数据，转换之后可以是更多个
• flatMap 转换同样可以改变发射的数据类型
• flatMap 转换后的数据，还是会逐个发射给我们的Observer来接收（就像这些数据是由一个Observable发射的一样，其实是多个Observable发射然后合并的）

这里抛出一个问题，flatMap会将原来的Observable，转换为多个Observable来发射数据，那么这些发射的数据是否会严格按顺序发射然后被Observer接收？

问题先留在这里，进入源码

FlatMap 源码

FlatMap操作符涉及的代码会相对多一些，但是也是有规律可循。 同样到Observable 类中看 flatMap的定义，源码作者为了方便开发者调用，提供了多个方法重载，我们最常用的方法定义如下

flatmap 方法1.png

最终调用的方法是

flatmap 方法.png

跟map 的套路 差不多，我们直接进入 ObservableFlatMap类， 我们还是看它的 subscribeActual 方法实现

ObservableFlatMap.png

可以看到，它给原Observer 装饰后的 Observer 是 MergeObserver，我们再继续看 MergeObserver 的 onNext 方法

MergeObserver onnext.png

由于我们默认调用的flatmap 的 maxConcurrency 大小是 Integer.MAX_VALUE, 所以最终会调用 subscribeInner(p)，注意这里我们的mapper方法以及被调用了，p就是跟我们传入的Function生成的Observable，我们再继续往下看

MergeObserver subscribeInner.png

一般我们传入的Function 生成的Observable 都不是 Callable类型的，所以最终传给Observable p 的 是InnerObserver, 找到了最终元凶，直接去看它的onNext方法实现吧。

MergeObserver onnext.png

funsionMode 默认是 None，走第一个if 逻辑，最终调用的是 上面的MergeObserable 的 tryEmit 方法，继续进去看

MergeObservable tryEmit.png

这里要插一句，MergeObserver 继承了 AtomicInteger，所以这里的tryEmit方法就利用了 AtomicInteger 的同步机制，所以同时只会有一个 value 被 actual Observer 发射，而且这里 刚好 可以解答我们上面留下的 问题，由于 AtomicInteger CAS锁只能保证操作的原子性，并不保证锁的获取顺序，是抢占式的，所以最终数据的发射顺序并不是固定的（同一个Observable发出的数据是有序的）

如果没有获取到锁，就会将要发射的数据放入 队列中，drainLoop 方法会循环去获取队列中的 数据，然后发射，由于篇幅有限，更详细的调用过程大家可以看源码。

dramloop.png

Map 和 FlatMap 二个操作符的 源码就解析到这里，水平有限，有不对的，还望大佬不吝赐教。