RxSwift（2）— 核心邏輯源碼分析

就問此時此刻還有誰？45度仰望天空，該死！我這無處安放的魅力！

• RxSwift（1）— 初探
• RxSwift（2）— 核心邏輯源碼分析
• RxSwift（3）— Observable序列的建立方式
• RxSwift（4）— 高階函數（上）
• RxSwift（5）— 高階函數（下）
• RxSwift（6）— 調度者-scheduler源碼解析（上）
• RxSwift（7）— 調度者-scheduler源碼解析（下）
• RxSwift（8）— KVO底層探索（上）
• RxSwift（9）— KVO底層探索（下）
• RxSwift（10）— 場景序列總結
• RxSwift（11）— dispose源碼解析
• RxSwift（12）— Subject即攻也守
• RxSwift（13）— 爬過的坑
• RxSwift（14）— MVVM雙向綁定

---

RxSwift 目錄直通車 --- 和諧學習，不急不躁！

---

做爲ReactiveX家族之一的RxSwift在Github截止如今Star:16K.爲何這個框架如此受歡迎，做爲函數響應式框架典型表明，底層實現又是如何實現的呢？這一篇文章全面解密，但願個人解讀給你帶來更多捷徑

RxSwift核心流程

RxSwift這個優秀的框架，設計的api也是很是精簡，讓陌生的用戶也能很是快速上手

• 1: 建立序列
• 2: 訂閱序列
• 3:發送信號

// 1: 建立序列
_ = Observable<String>.create { (obserber) -> Disposable in
    // 3:發送信號
    obserber.onNext("Cooci - 框架班級")
    return Disposables.create()  // 這個銷燬不影響咱們此次的解讀
    // 2: 訂閱序列
    }.subscribe(onNext: { (text) in
        print("訂閱到:\(text)")
    })

// 控制檯打印：「訂閱到:Cooci - 框架班級」
複製代碼

我剛開始在探索的時候，我是比較好奇的：爲何咱們的Cooci - 框架班級這個字符串會在訂閱序列的subscribe的閉包打印。下面是個人代碼分析

分析代碼：

• 1：建立序列的代碼 Create 後面的 閉包A 裏面有 3：發送信號，若是要執行 發送信號 ，必然要來到這個閉包A
• 2：咱們執行 2: 訂閱序列 跟了 閉包B
• 3：經過結果咱們顯然知道，先執行 閉包A 把 Cooci - 框架班級 傳給了 閉包B
• 猜想：代碼裏面嵌套了閉包的執行調用！猜想的真實性，咱們開始解讀源碼來驗證

PS: 說實話 RxSwift框架的源碼的確比較複雜而且不少，不少基礎薄弱或者耐性不夠的小夥伴很容易放棄。可是你看到這篇博客，你有福了：我會快速簡短給你介紹，在最後面會附上我繪製的思惟導圖！

RxSwift核心邏輯

建立序列

extension ObservableType {
    // MARK: create
    public static func create(_ subscribe: @escaping (AnyObserver<E>) -> Disposable) -> Observable<E> {
        return AnonymousObservable(subscribe)
    }
}
複製代碼

你們能夠很清晰看到咱們的 可觀察序列 的建立是利用協議拓展功能的create方法實現的，裏面建立了一個 AnonymousObservable（匿名可觀察序列） 命名仍是體現了做者的思惟 ：這個類就是一個內部類，具有一些通用特性（具備本身功能的類纔會命名） 下面我貼出這個類的繼承關係

從上面的圖，咱們能夠清晰的看到的繼承關係。那麼這麼多的內容還有那麼多層嵌套，這個地方咱們須要掌握什麼：

• create 方法的時候建立了一個內部對象 AnonymousObservable
• AnonymousObservable 保存了外界的閉包
• AnonymousObservable繼承了 Producer 具備很是重要的方法 subscribe

訂閱序列

這裏說明這個訂閱方法 subscribe 和咱們上面所說的 subscribe 不是同一個方法

來自於對 ObservableType 的拓展功能

extension ObservableType {
    public func subscribe(onNext: ((E) -> Void)? = nil, ...) -> Disposable {
           // 由於篇幅 省略不影響咱們探索的代碼
            let observer = AnonymousObserver<E> { 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()
                }
            }
            return Disposables.create(
                self.asObservable().subscribe(observer),
                disposable
            )
    }
}
複製代碼

代碼說明：

• E 這裏的意思是 Swift 的關聯類型，這個若是仔細看過可觀察序列的繼承鏈源碼應該不可貴出：這個E 就是咱們的 序列類型，咱們這裏就是String

public class Observable<Element> : ObservableType {
    /// Type of elements in sequence.
    public typealias E = Element
複製代碼

• 建立了一個 AnonymousObserver (匿名內部觀察者) 手法和咱們的 AnonymousObservable 差很少，它這裏的初始化是閉包參數，保存了外界的 onNext, onError , onCompleted , onDisposed 的處理回調閉包的調用,下面我仍是給你們貼出 觀察者 的繼承鏈關係，幫助你們理解

• self.asObservable() 這個是咱們的 RxSwift 爲了保持一致性的寫法
• self.asObservable().subscribe(observer)其實本質就是 self.subscribe(observer),經過可觀察序列的繼承關係，咱們能夠很是快速的定位 Producer 訂閱代碼

override func subscribe(_ observer: O) -> Disposable where O.E == Element {
        if !CurrentThreadScheduler.isScheduleRequired {
            // 篇幅緣由，咱們省略一些代碼，方便咱們理解
            ...
            return disposer
        }
        else {
            return CurrentThreadScheduler.instance.schedule(()) { _ in
                let disposer = SinkDisposer()
                let sinkAndSubscription = self.run(observer, cancel: disposer)
              // 篇幅緣由，咱們省略一些代碼，方便咱們理解
              ...
                return disposer
            }
        }
    }
複製代碼

• 關於銷燬代碼和調度者代碼這裏不分析
• self.run 這個代碼最終由咱們生產者 Producer 延伸到咱們具體的事務代碼 AnonymousObservable.run

override func run (...) {
    let sink = AnonymousObservableSink(observer: observer, cancel: cancel)
    let subscription = sink.run(self)
    return (sink: sink, subscription: subscription)
}
複製代碼

• sink.run的寫法也是比較好的，業務處理的仍是下沉了，讓分工更加明確

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

• parent 就是上面傳過來的AnonymousObservable對象
• 咱們很是興奮的看到 AnonymousObservable._subscribeHandler,從這句代碼咱們解惑了爲何咱們的序列訂閱的時候流程會執行咱們 序列閉包，而後去執行 發送響應
• 發送響應的代碼等會分析，這裏還有一個比較重要的傢伙 AnyObserver(self)

public init<O : ObserverType>(_ observer: O) where O.E == Element {
    self.observer = observer.on
}
複製代碼

• 在這個構造方法裏面，咱們建立了一個結構體 AnyObserver 保存了一個信息 AnonymousObservableSink .on 函數，不是 AnonymousObservableSink,這個地方通常初次來到這裏的人都會犯錯誤。不知道你是否意識到了！

發送響應

咱們從上面的分析，很是清晰: obserber.onNext("Cooci - 框架班級") 的本質是： AnyObserver.onNext("Cooci - 框架班級")

這時候發現咱們的AnyObserver 是沒有這個方法，這很正常！通常思路，找父類，找協議

extension ObserverType {
    public func onNext(_ element: E) {
        self.on(.next(element))
    }
}
複製代碼

• 外界 obserber.onNext("Cooci - 框架班級") 再次變形 ：AnyObserver.on(.next("Cooci - 框架班級")) ,這裏你們必定要主要，這個AnyObserver調用了 on 裏面傳的是 .next函數, .next函數帶有咱們最終的參數

public struct AnyObserver<Element> : ObserverType {
    public init<O : ObserverType>(_ observer: O) where O.E == Element {
        self.observer = observer.on
    }
    public func on(_ event: Event<Element>) {
        return self.observer(event)
    }
}
複製代碼

• self.observer 構造初始化就是：AnonymousObservableSink .on 函數
• 看到這裏又要變形咯：self.observer(event) -> AnonymousObservableSink .on(event) 其中 event = .next("Cooci - 框架班級") 最終咱們的核心邏輯又回到了 sink 這個神奇的管子，看到這裏不由拍案叫絕， RxSwift這個設計能力，還有誰~~~

class AnonymousObservableSink<O: ObserverType>: Sink<O>, ObserverType {
    func on(_ event: Event<E>) {
        switch event {
        case .next:
            if load(self._isStopped) == 1 {
                return
            }
            self.forwardOn(event)
        case .error, .completed:
            if fetchOr(self._isStopped, 1) == 0 {
                self.forwardOn(event)
                self.dispose()
            }
        }
    }
}
複製代碼

• self.forwardOn(event) 這也是執行的核心代碼，由於 AnonymousObservableSink 繼承 Sink 這裏還有封裝，請看下面的代碼

class Sink<O : ObserverType> : Disposable {
    final func forwardOn(_ event: Event<O.E>) {
        if isFlagSet(self._disposed, 1) {
            return
        }
        self._observer.on(event)
    }
}
複製代碼

• 其中 self._observer 就是咱們初始化保存的 觀察者：AnonymousObserver
• 那麼咱們變形得出本質就是：AnonymousObserver.on（.next("Cooci - 框架班級")）,個人天啊! 這裏邏輯展轉回到了咱們 訂閱序列 時候建立的 AnonymousObserver 的參數閉包的調用！全部的一切感受是這樣的囉嗦，但又是這麼的順其資源。

let observer = AnonymousObserver<E> { 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()
    }
}
複製代碼

• 判斷 event 進而調用 onNext?(value) ,由於枚舉的關聯值（Swift很強大的功能）value = "Cooci - 框架班級", 接下來就是外界 onNext閉包的調用傳參，那麼這個時候源碼解析到這裏，我相信你已經徹底掌握了RxSwift的核心邏輯，最後這裏附上咱們的分析圖解

總結：RxSwift的結構

• 1：就是序列感念 滿世界都是序列 - 編碼統一 ，隨時隨地享用
• 2：經過函數式思想吧一些列的需求操做下沉（把開發者不關心的東西封裝） - 優化代碼，節省邏輯
• 3：RxSwift最典型的特點就是解決Swift這門靜態語言的響應能力，利用隨時間維度序列變化爲軸線，用戶訂閱關心能隨軸線一直保活，達到訂閱一次，響應一直持續~

就問此時此刻還有誰？45度仰望天空，該死！我這無處安放的魅力！