原文地址：http://www.reibang.com/p/0e0703466483

作為ReactiveX家族之一的RxSwift在Github截止現(xiàn)在Star:16K.為什么這個(gè)框架如此受歡迎集峦，作為函數(shù)響應(yīng)式框架典型代表，底層實(shí)現(xiàn)又是如何實(shí)現(xiàn)的呢抠刺？這一篇文章全面解密

RxSwift核心流程

RxSwift這個(gè)優(yōu)秀的框架塔淤，設(shè)計(jì)的api也是非常精簡，讓陌生的用戶也能非乘傺快速上手

• 1: 創(chuàng)建序列
• 2: 訂閱序列
• 3:發(fā)送信號

// 1: 創(chuàng)建序列
_ = Observable<String>.create { (obserber) -> Disposable in
    // 3:發(fā)送信號
    obserber.onNext("Cooci -  框架班級")
    return Disposables.create()  // 這個(gè)銷毀不影響我們這次的解讀
    // 2: 訂閱序列
    }.subscribe(onNext: { (text) in
        print("訂閱到:\(text)")
    })

// 控制臺打痈叻洹：“訂閱到:Cooci -  框架班級”

我剛開始在探索的時(shí)候，我是比較好奇的：為什么我們的Cooci - 框架班級這個(gè)字符串會在訂閱序列的subscribe的閉包打印罕容。下面是我的代碼分析

分析代碼：

• 1：創(chuàng)建序列的代碼 Create 后面的 閉包A 里面有 3：發(fā)送信號备恤，如果要執(zhí)行 發(fā)送信號 ，必然要來到這個(gè)閉包A
• 2：我們執(zhí)行 2: 訂閱序列 創(chuàng)建了 閉包B
• 3：通過結(jié)果我們顯然知道锦秒，先執(zhí)行 閉包A 把 Cooci - 框架班級 傳給了 閉包B
• 猜測：代碼里面嵌套了閉包的執(zhí)行調(diào)用露泊！猜測的真實(shí)性，我們開始解讀源碼來驗(yàn)證

PS: 說實(shí)話 RxSwift框架的源碼的確比較復(fù)雜并且很多脂崔，很多基礎(chǔ)薄弱或者耐性不夠的小伙伴很容易放棄。但是你看到這篇博客梧喷，你有福了：我會快速簡短給你介紹砌左，在最后面會附上我繪制的思維導(dǎo)圖！

RxSwift核心邏輯

創(chuàng)建序列

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

大家可以很清晰看到我們的 可觀察序列 的創(chuàng)建是利用協(xié)議拓展功能的create方法實(shí)現(xiàn)的铺敌，里面創(chuàng)建了一個(gè) AnonymousObservable（匿名可觀察序列） 命名還是體現(xiàn)了作者的思維 ：這個(gè)類就是一個(gè)內(nèi)部類汇歹，具備一些通用特性（具有自己功能的類才會命名） 下面我貼出這個(gè)類的繼承關(guān)系

從上面的圖，我們可以清晰的看到的繼承關(guān)系偿凭。那么這么多的內(nèi)容還有那么多層嵌套产弹，這個(gè)地方我們需要掌握什么：

• create 方法的時(shí)候創(chuàng)建了一個(gè)內(nèi)部對象 AnonymousObservable
• AnonymousObservable 保存了外界的閉包
• AnonymousObservable繼承了 Producer 具有非常重要的方法 subscribe

訂閱序列

這里說明這個(gè)訂閱方法 subscribe 和我們上面所說的 subscribe 不是同一個(gè)方法

來自于對 ObservableType 的拓展功能

extension ObservableType {
    public func subscribe(onNext: ((E) -> Void)? = nil, ...) -> Disposable {
           // 因?yàn)槠?省略不影響我們探索的代碼
            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 的關(guān)聯(lián)類型，這個(gè)如果仔細(xì)看過可觀察序列的繼承鏈源碼應(yīng)該不難得出：這個(gè)E 就是我們的 序列類型，我們這里就是String

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

• 創(chuàng)建了一個(gè) AnonymousObserver (匿名內(nèi)部觀察者) 手法和我們的 AnonymousObservable 差不多痰哨，它這里的初始化是閉包參數(shù)胶果，保存了外界的 onNext, onError , onCompleted , onDisposed 的處理回調(diào)閉包的調(diào)用,下面我還是給大家貼出 觀察者 的繼承鏈關(guān)系，幫助大家理解

• self.asObservable() 這個(gè)是我們的 RxSwift 為了保持一致性的寫法
• self.asObservable().subscribe(observer)其實(shí)本質(zhì)就是 self.subscribe(observer),通過可觀察序列的繼承關(guān)系斤斧，我們可以非吃缈伲快速的定位 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
            }
        }
    }

• 關(guān)于銷毀代碼和調(diào)度者代碼這里不分析
• self.run 這個(gè)代碼最終由我們生產(chǎn)者 Producer 延伸到我們具體的事務(wù)代碼 AnonymousObservable.run

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

• sink.run的寫法也是比較好的游昼，業(yè)務(wù)處理的還是下沉了甘苍，讓分工更加明確

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

• parent 就是上面?zhèn)鬟^來的AnonymousObservable對象
• 我們非常興奮的看到 AnonymousObservable._subscribeHandler,從這句代碼我們解惑了為什么我們的序列訂閱的時(shí)候流程會執(zhí)行我們 序列閉包，然后去執(zhí)行 發(fā)送響應(yīng)
• 發(fā)送響應(yīng)的代碼等會分析烘豌，這里還有一個(gè)比較重要的家伙 AnyObserver(self)

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

• 在這個(gè)構(gòu)造方法里面载庭，我們創(chuàng)建了一個(gè)結(jié)構(gòu)體 AnyObserver 保存了一個(gè)信息 AnonymousObservableSink .on 函數(shù)，不是 AnonymousObservableSink,這個(gè)地方一般初次來到這里的人都會犯錯誤扇谣。不知道你是否意識到了昧捷！

發(fā)送響應(yīng)

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

這時(shí)候發(fā)現(xiàn)我們的AnyObserver 是沒有這個(gè)方法罐寨，這很正常靡挥！一般思路，找父類鸯绿，找協(xié)議

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

• 外界 obserber.onNext("Cooci - 框架班級") 再次變形 ：AnyObserver.on(.next("Cooci - 框架班級")) ,這里大家一定要主要跋破，這個(gè)AnyObserver調(diào)用了 on 里面?zhèn)鞯氖?.next函數(shù), .next函數(shù)帶有我們最終的參數(shù)

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 構(gòu)造初始化就是：AnonymousObservableSink .on 函數(shù)
• 看到這里又要變形咯：self.observer(event) -> AnonymousObservableSink .on(event) 其中 event = .next("Cooci - 框架班級") 最終我們的核心邏輯又回到了 sink 這個(gè)神奇的管子，看到這里不禁拍案叫絕瓶蝴， RxSwift這個(gè)設(shè)計(jì)能力毒返，還有誰~~~

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) 這也是執(zhí)行的核心代碼，因?yàn)?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
• 那么我們變形得出本質(zhì)就是：AnonymousObserver.on（.next("Cooci - 框架班級")）,我的天啊! 這里邏輯輾轉(zhuǎn)回到了我們 訂閱序列 時(shí)候創(chuàng)建的 AnonymousObserver 的參數(shù)閉包的調(diào)用拧簸！所有的一切感覺是這樣的啰嗦，但又是這么的順其資源男窟。

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 進(jìn)而調(diào)用 onNext?(value) ,因?yàn)槊杜e的關(guān)聯(lián)值（Swift很強(qiáng)大的功能）value = "Cooci - 框架班級", 接下來就是外界 onNext閉包的調(diào)用傳參盆赤，那么這個(gè)時(shí)候源碼解析到這里，我相信你已經(jīng)完全掌握了RxSwift的核心邏輯歉眷，最后這里附上我們的分析圖解

總結(jié)：RxSwift的結(jié)構(gòu)

• 1：就是序列感念 滿世界都是序列 - 編碼統(tǒng)一 牺六，隨時(shí)隨地享用
• 2：通過函數(shù)式思想吧一些列的需求操作下沉（把開發(fā)者不關(guān)心的東西封裝） - 優(yōu)化代碼，節(jié)省邏輯
• 3：RxSwift最典型的特色就是解決Swift這門靜態(tài)語言的響應(yīng)能力汗捡，利用隨時(shí)間維度序列變化為軸線淑际，用戶訂閱關(guān)心能隨軸線一直保活，達(dá)到訂閱一次春缕，響應(yīng)一直持續(xù)~

對于RxSwift盗胀，有更好的見解，想要更好的探討淡溯，可以進(jìn)入iOS技術(shù)群读整，一起探討交流

• 點(diǎn)擊加入與 2000+iOS開發(fā)者一起交流探討