由于项目的需要，笔者最近在学习swift，为了更加高效的撸代码，研究了RxSwift这个牛逼的库，对于它的核心流程，进行了一番探索（实际上耗费了毕生心血），对于下面的叙述，如果错误，还望大家多多留言指正，谢谢了🙏🙏🙏

核心代码块

    override func viewDidLoad() {
        super.viewDidLoad()
        //1、创建序列
        let ob = Observable<Any>.create { (obserber) -> Disposable in
            //3、发送信号
            obserber.onNext("佐笾")
            obserber.onError(NSError.init(domain: "1234", code: 10086, userInfo: nil))
            obserber.onCompleted()
            return Disposables.create()
        }
        
        //2、订阅信号
        let _ = ob.subscribe(onNext: { (text) in
            print("订阅到：\(text)")
        }, onError: { (error) in
            print("错误：\(error)")
        }, onCompleted: {
            print("完成")
        }) {
            print("销毁")
        }
    }

运行代码结果如下：

image.png

核心代码如上代码所示，共三步

• 创建序列
• 订阅信号
• 发送信号

那么对于上面的流程中提出一下几个问题：

• 序列是如何创建？
• 序列是如何接受到发送的信号的？
• 整个流程又是如何实现的？

下面笔者会跟着大家一起进入源码，一探究竟！
想要具体了解原理的小伙伴，最好带着源码进行阅读下面的内容，便于你的理解

源码解析

创建序列

序列的创建很简单，就是下面一行代码

let ob = Observable<Any>.create { 闭包A }

进入查看create，这里的create只能在create.swift文件里查看，不然跟踪不到这个方法😅

可是，为神马会走到这里来呢？

进入对象Observable查看，会得到下面的继承关系

image.png

所以在ObservableType的扩展中查看create方法，具体实现如下

    public static func create(_ subscribe: @escaping (AnyObserver<Element>) -> Disposable) -> Observable<Element> {
        return AnonymousObservable(subscribe)
    }

可以看到，返回了一个AnonymousObservable对象，并且传入了一个参数subscribe，通过创建的代码可以知道，这里的subscribe就是外面我们定义的闭包，后面文章中笔者会称为它为闭包A，也就是发送信号的部分。

进入AnonymousObservable类的查看源码，

final private class AnonymousObservable<Element>: Producer<Element> {
    typealias SubscribeHandler = (AnyObserver<Element>) -> Disposable

    let _subscribeHandler: SubscribeHandler

    init(_ subscribeHandler: @escaping SubscribeHandler) {
        self._subscribeHandler = subscribeHandler
    }

    override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Element {
        let sink = AnonymousObservableSink(observer: observer, cancel: cancel)
        let subscription = sink.run(self)
        return (sink: sink, subscription: subscription)
    }
}

这个类的实现很简单，通过init初始化方法可以看到，闭包A被保存在了_subscribeHandler属性中。 同时这里还提供了一个非常重要run方法，后面会走到这里，所以这里先强调一下。

到此，序列的创建就结束了，下面在看订阅信号的解析

订阅信号

        //2、订阅信号
        let _ = ob.subscribe(onNext: { (text) in
            print("订阅到：\(text)")
        }, onError: { (error) in
            print("错误：\(error)")
        }, onCompleted: {
            print("完成")
        }) {
            print("销毁")
        }

从上面代码可以知道，重点解析在于方法subscribe，进入源码，过滤掉其他无关代码如下

image.png

嗯，不错，就这两行代码，信号的订阅到发送，就结束了。

此刻请允许笔者通过下面内容来表达一下此刻的内心

image.png

嗯。。。耐下心来，慢慢琢磨可以知道，想要突破，那么关键代码就在下面这一行

self.asObservable().subscribe(observer)

通过断点调试，可以知道这里的self表示着AnonymousObservable，或者通过我们写的核心代码块也可以知道，首先创建了一个序列ob，它的类型上面说过了是AnonymousObservable类型，接着订阅信号是直接用ob调用的方法subscribe，所以这里的self表示着AnonymousObservable，它就是创建序列时返回的对象。

但是在上面对AnonymousObservable解析中，并没有发现它拥有一个叫做subscribe的方法，再次回到AnonymousObservable类的源码中，可以发现，它是继承自Producer类，不错，在它的父类Producer中我们发现了subscribe方法。

这里停一下，因为此时传了一个参数observer，它也是一个AnonymousObserver对象，同时带着一个尾随闭包，暂且称这个参数为闭包B，以便后面使用它的时候，我们可以容易想起来。

接着上面的思路，进入Producer类中查看，在subscribe方法中，调用了一个run方法，本类中，并没有对这个方法进行具体实现，而是分发给了它的子类去实现，那么这里又需要回到AnonymousObservable类中，在上面的序列创建的内容中，笔者也就和大家说了这里run方法的重要，下面具体分析。

在此之前，需要注意的是，参数闭包B一直在传递。

    override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Element {
        let sink = AnonymousObservableSink(observer: observer, cancel: cancel)
        let subscription = sink.run(self)
        return (sink: sink, subscription: subscription)
    }

创建了一个实例对象sink，然后调用了一个run方法。这里看不出来什么东西，进入AnonymousObservableSink查看源码。

可以发现下面几点

• 第一，重写了父类的init方法，而传进来的闭包B，传递给了父类Sink，同时赋值给了父类的_observer属性。
• 第二，方法on，参数为event
• 第三，方法run，参数parent

OK，AnonymousObservableSink类的内容，差不多就这么多，回到之前的代码中

let subscription = sink.run(self)

这行代码里的self表示类AnonymousObservable，我想大家应该都没有意见，那么具体看下这个方法的实现。

    func run(_ parent: Parent) -> Disposable {
        return parent._subscribeHandler(AnyObserver(self))
    }

转化一下代码AnonymousObservable._subscribeHandler(AnyObserver(self))，这样看的话，我相信大家应该非常的熟悉了。

AnonymousObservable是什么？ 它是在我们创建序列时的返回的那个对象。
_subscribeHandler是什么？ 它是在我们创建序列时传的参数闭包A。

那么这句话的意思就是执行闭包A，即发送信号

发送信号

分两步来分析AnonymousObservable._subscribeHandler(AnyObserver(self))这行代码；

1、发送信号的闭包

回到发送信号的闭包中，

obserber.onNext("佐笾")

这里只分析onNext()，因为其他原理一样，就不过多描述了啊！

按住command，追踪onNext()方法，可以跳转到ObserverType扩展中

    public func onNext(_ element: Element) {
        self.on(.next(element))
    }

可以通过断点调试知道，这里的self指的是AnyObserver类，或者可以回到create.swift文件中的create方法中查看

image.png

这里已经把对应的obserber的类型已经描述的很清晰了。obserber为AnyObserver类型，且遵循ObserverType协议，调用的onNext()方法就是ObserverType协议里的方法（查看一下AnyObserver类的源码便知）。

回到AnyObserver类中，查看on方法

    public func on(_ event: Event<Element>) {
        return self.observer(event)
    }

可以断点看一下，这里已经把信号传了进来

image.png

那么这里的self.observer()具体是什么呢？请看下面的分析

2、解析 AnyObserver(self)

通过上面流程可以知道，关键代码在于AnyObserver(self)，通过进入源码查看

    public init<Observer: ObserverType>(_ observer: Observer) where Observer.Element == Element {
        self.observer = observer.on
    }

在初始化代码里有这么一个操作，observer.on赋值的操作，这里的observer是通过初始化构造方法传来的，回顾上面的流程可以知道，它是AnonymousObservableSink类。那么它执行on方法，跟着代码走，会调用方法forwardOn，最后会来到Sink类里的forwardOn方法，源码如下

    final func forwardOn(_ event: Event<Observer.Element>) {
        #if DEBUG
            self._synchronizationTracker.register(synchronizationErrorMessage: .default)
            defer { self._synchronizationTracker.unregister() }
        #endif
        if isFlagSet(self._disposed, 1) {
            return
        }
        self._observer.on(event)
    }

关键代码在于self._observer.on(event)，在订阅信号的分析中，特别说明了_observer是闭包B，如果这里有疑问的小伙伴，可会回到前面看一下。

闭包B的具体结构如下（部分代码省略）：

AnonymousObserver<Element> { event in
                switch event {
                case .next(let value):
                    onNext?(value)
                case .error(let error):
                    if let onError = onError {
                        onError(error)
                    }
                    else {
                        Hooks.defaultErrorHandler(callStack, error)
                    }
                    disposable.dispose()
                case .completed:
                    onCompleted?()
                    disposable.dispose()
                }
            }

那么对于self._observer.on(event)，可以理解为类AnonymousObserver调用方法on，进入源码查看，AnonymousObserver类中并无该方法，然后通过它的父类ObserverBase可以找到该方法，具体实现如下：

    func on(_ event: Event<Element>) {
        switch event {
        case .next:
            if load(self._isStopped) == 0 {
                self.onCore(event)
            }
        case .error, .completed:
            if fetchOr(self._isStopped, 1) == 0 {
                self.onCore(event)
            }
        }
    }

紧接着，执行self.onCore(event)，同样的，该方法，父类ObserverBase并没有具体实现，而是分发给子类去实现，回到类AnonymousObserver，查看该方法的实现代码

    override func onCore(_ event: Event<Element>) {
        return self._eventHandler(event)
    }

那么对于self._eventHandler()是个什么东西，可以具体分析一下，在当前类AnonymousObserver中，

    typealias EventHandler = (Event<Element>) -> Void
    
    private let _eventHandler : EventHandler

追溯到枚举Event

public enum Event<Element> {
    /// Next element is produced.
    case next(Element)

    /// Sequence terminated with an error.
    case error(Swift.Error)

    /// Sequence completed successfully.
    case completed
}

到这里，我想大家应该可以知道下面我们应该走到哪里了，没错，就是刚刚说的闭包B，根据event执行不同的方法，比如这里所说的onNext方法，那么就会打印输出订阅到：佐笾

总结

OK，到这里关于RxSwift的核心流程的源码分析已经结束了，了解了其中的原理，笔者相信，对于RxSwift的运用会更加得心应手。

好难，笔者在整理这份笔记的时候，确实花费了不少时间，本想画个流程图来帮助大家去理解，奈何时间有限，后续笔者会整理一份思路较清晰的流程图；

上述的分析流程，如果有什么错误，希望大家给笔者留言指正一下，谢谢！！！

持续学习，希望大家持续关注！！！