RxSwift高阶函数merge解读

Posted 2023-02-22

参考技术A 通过使用 merge 操作符你可以将多个 Observables 合并成一个，当某一个 Observable 发出一个元素时，他就将这个元素发出。

如果，某一个 Observable 发出一个 onError 事件，那么被合并的 Observable 也会将它发出，并且立即终止序列。

这只是个最简单的 merge。

首先看下 merge 函数：

返回了一个 Merge 序列对象：

Merge 序列被订阅后，会走 MergeBasicSink （ MergeSink 的子类）的 run 函数。

MergeBasicSink 中只实现了一个 performMap ，还要看它的父类。

MergeSink 的 run 中， source.subscribe(self) ：源序列开始订阅， MergeSink 作为观察者。这里的源序列就是外界的 of 序列，那么 of 序列的响应必然会来到 MergeSink 的 on 函数中。

MergeSink 的 on 函数中，在 .next 类型下，调用 nextElementArrived ，这个函数在这里只做了个计数 self._activeCount += 1 ，为了管理序列的生命周期。因为子类对 performMap 的实现就是直接返回了 element 。

现在是 of 序列 的响应，这里的 element 就是 of 序列 中的两个元素（ PublishSubject 序列）。

接着又调用了 subscribeInner 把 PublishSubject 序列传过去。这里才是 merge 用来合并的重点。这里用了 MergeSinkIter 类，用 PublishSubject 作为源序列去订阅，把 MergeSinkIter 作为它的观察者。这样，等 PublishSubject 有响应的时候就会通知到 MergeSinkIter 的 on 函数中。

MergeSinkIter 的 on 里面直接让它的 _parent（MergeSink）调用 forwardOn ，去让 MergeSink 的观察者响应外界的订阅。这样的话，of 序列中的任意元素（ PublishSubject 序列）发出的信号都可以被 merge 序列的订阅者接收到。