RxSwift源碼分析(一)-核心邏輯解析

前言：這幾篇關於RxSwift源碼分析的文章主要是對源碼進行解析，不涉及到RxSwift的具體使用。具體使用能夠查看RxSwift中文文檔或者github。

在RxSwift的使用中，咱們須要先建立序列，而後訂閱信號、發送信號來實現一個完整的流程。先舉一個簡單的例子：

// 1:建立序列
        let observable = Observable<Any>.create { (obserber) -> Disposable in
            // 3:發送信號
            obserber.onNext("發送信號")
            obserber.onCompleted()
//            obserber.onError(NSError.init(domain: "fail", code: 10087, userInfo: nil))
            return Disposables.create()
        }
        
        // 2:訂閱信號
        let _ = observable.subscribe(onNext: { (text) in
            print("訂閱到信號:\(text)")
        }, onError: { (error) in
            print("error: \(error)")
        }, onCompleted: {
            print("完成")
        }) {
            print("銷燬")
        }
複製代碼

咱們須要重點關注的是：

1. 第一行代碼中的create的閉包何時執行
2. 訂閱信號subscribe的閉包何時執行
3. 信號從發送出去到接收流程是怎樣的

下面針對這幾個問題來一一探索

第一步create函數

首先來看看create函數，它返回一個AnonymousObservable對象，AnonymousObservable是匿名可觀察者,用來存儲產生事件的閉包（self._subscribeHandler = subscribeHandler）和激活處理事件閉包的入口（run方法）。

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

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<O : ObserverType>(_ observer: O, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where O.E == Element {
        let sink = AnonymousObservableSink(observer: observer, cancel: cancel)
        let subscription = sink.run(self)
        return (sink: sink, subscription: subscription)
    }
}
複製代碼

AnonymousObservable是Observable的子類，它們的繼承關係是：AnonymousObservable -> Producer -> Observable -> ObservableType -> ObservableConvertibleType

第二步subscribe函數

源碼：

public func subscribe(onNext: ((E) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil)
        -> Disposable {
            let disposable: Disposable
            
            if let disposed = onDisposed {
                disposable = Disposables.create(with: disposed)
            }
            else {
                disposable = Disposables.create()
            }
            
            #if DEBUG
                let synchronizationTracker = SynchronizationTracker()
            #endif
            
            let callStack = Hooks.recordCallStackOnError ? Hooks.customCaptureSubscriptionCallstack() : []
            
            let observer = AnonymousObserver<E> { event in
                
                #if DEBUG
                    synchronizationTracker.register(synchronizationErrorMessage: .default)
                    defer { synchronizationTracker.unregister() }
                #endif
                
                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
            )
    }
複製代碼

從上面的代碼能夠看出，首先建立了一個AnonymousObserver對象，在初始化時傳遞了一個閉包做爲參數而且保存下來self._eventHandler = eventHandler。 AnonymousObserver是匿名觀察者,用於存儲和處理事件的閉包。

而後在最後有self.asObservable().subscribe(observer)這樣一行代碼，asObservable返回的是對象自己，而後調用subscribe這個函數而且把建立的AnonymousObserver對象傳遞過去，會來到AnonymousObservable這個類裏面，可是發現這個類裏面沒有subscribe方法，咱們往父類Producer裏面找到這個方法，源碼以下：

override func subscribe<O : ObserverType>(_ observer: O) -> Disposable where O.E == Element {
        if !CurrentThreadScheduler.isScheduleRequired {
            // The returned disposable needs to release all references once it was disposed.
            let disposer = SinkDisposer()
            let sinkAndSubscription = self.run(observer, cancel: disposer)
            disposer.setSinkAndSubscription(sink: sinkAndSubscription.sink, subscription: sinkAndSubscription.subscription)

            return disposer
        }
        else {
            return CurrentThreadScheduler.instance.schedule(()) { _ in
                let disposer = SinkDisposer()
                let sinkAndSubscription = self.run(observer, cancel: disposer)
                disposer.setSinkAndSubscription(sink: sinkAndSubscription.sink, subscription: sinkAndSubscription.subscription)

                return disposer
            }
        }
    }
複製代碼

在這裏調用了run方法，也就是AnonymousObservable這個類裏面的run方法，把observer做爲參數傳過來。

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

在run方法中，先建立一個AnonymousObservableSink對象並持有observer，而後調用這個對象的run方法把self傳遞過去，也就是把observable做爲參數。 AnonymousObservableSink這個類將可觀察者Observable和觀察者Observer連接起來,實現事件的傳遞，起到一個橋樑的做用。

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

在這裏觸發了_subscribeHandler的調用，這裏的_subscribeHandler就是以前create函數的閉包。到這裏就解決了第一個問題「第一行代碼中的create的閉包何時執行」。 在這個方法中把self轉換成AnyObserver對象，也就是把AnonymousObservableSink對象轉換成AnyObserver對象。那麼，具體是怎麼轉換的呢？咱們來看看源碼

public struct AnyObserver<Element> : ObserverType {
    /// The type of elements in sequence that observer can observe.
    public typealias E = Element
    
    /// Anonymous event handler type.
    public typealias EventHandler = (Event<Element>) -> Void

    private let observer: EventHandler

    /// Construct an instance whose `on(event)` calls `eventHandler(event)`
    ///
    /// - parameter eventHandler: Event handler that observes sequences events.
    public init(eventHandler: @escaping EventHandler) {
        self.observer = eventHandler
    }
    
    /// Construct an instance whose `on(event)` calls `observer.on(event)`
    ///
    /// - parameter observer: Observer that receives sequence events.
    public init<O : ObserverType>(_ observer: O) where O.E == Element {
        self.observer = observer.on
    }
    
    /// Send `event` to this observer.
    ///
    /// - parameter event: Event instance.
    public func on(_ event: Event<Element>) {
        return self.observer(event)
    }

    /// Erases type of observer and returns canonical observer.
    ///
    /// - returns: type erased observer.
    public func asObserver() -> AnyObserver<E> {
        return self
    }
}
複製代碼

能夠看到在構造函數中有一行代碼self.observer = observer.on，就是把AnonymousObservableSink類的on函數賦值給AnyObserver類的observer變量。從這裏就能夠明白爲何這行代碼observer.onNext("發送信號") 最終會觸發AnonymousObservableSink.on事件。

// AnonymousObservableSink類on方法
    func on(_ event: Event<E>) {
        #if DEBUG
            self._synchronizationTracker.register(synchronizationErrorMessage: .default)
            defer { self._synchronizationTracker.unregister() }
        #endif
        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()
            }
        }
    }
複製代碼

而後進入到Sink類的forwardOn方法

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

這裏的_observer就是第二步調用subscribe函數裏面建立的observer對象。 會先進入到父類的ObserverBase的on方法

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

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

這裏的_eventHandler閉包就是第二步調用subscribe函數傳入的閉包。 到這裏就能夠理解第二個和第三個問題了。

總結

從整個流程來看，咱們能夠體會到函數響應式編程的魅力，代碼簡介，思路很是的清晰。

梳理一下整個流程：

tips：因爲在RxSwift中普遍的使用了協議和泛型，還有不少抽象方法和同名方法，致使在剛開始閱讀源碼的時候可能有些不是很習慣，再加上Xcode跟蹤代碼不是很強大，閱讀起來有一些障礙。因此在閱讀源碼的時候須要耐心一點，不要找錯了地方，能夠找一些重點代碼閱讀，不要盲目的從頭看到尾。

有問題或者建議和意見，歡迎你們評論或者私信。

喜歡的朋友能夠點下關注和喜歡，後續會持續更新文章。