RxSwift核心邏輯（一）-序列的產(chǎn)生以及訂閱

RxSwift github地址

中文文檔

前言

RxSwift是Swift的一套響應(yīng)式編程框架，如同OC的RAC梨树。
響應(yīng)式編程的核心如下圖：（來源于中文文檔）略水。

描述了序列的產(chǎn)生、序列的訂閱（綁定）劝萤、序列的操作符渊涝，以及訂閱管理器Disposable（管理訂閱的生命周期）。其實(shí)還有個(gè)Schedulers

image

Observable:
- 是一個(gè)可監(jiān)聽序列
- 可以理解為觀察者模式里的被觀察者
- 類圖如下：
  
  image
Observer：
- 序列的監(jiān)聽者
- 可以理解為觀察者模式里的觀察者
- 類圖如下：
  
  image
Disposable：可被清除的資源（管理綁定（訂閱）的生命周期
Operator：序列操作符（這篇文章暫不做分析）
Schedulers:線程床嫌、隊(duì)列的調(diào)度器

demo

看下面的demo跨释，這里描述了，序列的產(chǎn)生和訂閱的一個(gè)簡單過程厌处。

/// 創(chuàng)建一個(gè)序列
let observable = Observable<String>.create { (observe) -> Disposable in
    /// 生產(chǎn)一個(gè)事件
    observe.onNext("1")
    /// 生產(chǎn)一個(gè)事件
    observe.onNext("2")
    /// 發(fā)送一個(gè)錯(cuò)誤事件
    //observe.onError(MyError.error1)
    /// 發(fā)送一個(gè)完成事件
    observe.onCompleted()
    /// 返回一個(gè)資源管理對象
    return Disposables.create()
}.observeOn(ConcurrentDispatchQueueScheduler.init(queue: DispatchQueue.global()))

/// 監(jiān)聽（訂閱）序列    
observable.subscribeOn(MainScheduler.instance).subscribe(onNext: { (str) in
    /// 輸出監(jiān)聽到的事件
    print(str)
}, onError: { ( error) in
    /// 輸出錯(cuò)誤信息
    print(error)
}, onCompleted: {
    print("訂閱完成")
}).disposed(by: disposeBag)
#輸出如下：
1
2
訂閱完成

了解了RxSwift的序列的產(chǎn)生和消費(fèi)之后鳖谈，我們接下來探索下序列是如何產(chǎn)生的，又是如何消費(fèi)的阔涉。

序列的產(chǎn)生

根據(jù)上面的demo缆娃，create創(chuàng)建了一個(gè)可監(jiān)聽序列捷绒。下面是create源碼。

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

由上面源碼看出贯要，返回了一個(gè)AnonymousObservable類, 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)
    }
}

從上面可以看出暖侨，創(chuàng)建了一個(gè)可被監(jiān)聽的序列AnonymousObservable,并且AnonymousObservable保存了，序列監(jiān)聽者的回調(diào)崇渗。

序列的監(jiān)聽（訂閱）

序列的監(jiān)聽：也就是上面的demo的 observable.subscribe...字逗。

源碼如下：

public func subscribe(onNext: ((Element) -> 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<Element> { 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
            )
    }

前面的一些Disposable管理，我們先不管宅广，直接從let observer = AnonymousObserver<Element> { event in 這一行代碼開始看葫掉。

這里創(chuàng)建了一個(gè)序列監(jiān)聽者類AnonymousObserver,并初始化了事件處理的回調(diào)。
回調(diào)里面根據(jù)不同的事件類型跟狱，來執(zhí)行相應(yīng)的事件俭厚。然后用資源管理器Disposables將可監(jiān)聽序列(observable)和序列監(jiān)聽者（observer）關(guān)聯(lián)起來，

那么驶臊，是如何關(guān)聯(lián)的呢挪挤？

關(guān)聯(lián)Observable和Observer

源碼如下：這里只貼相關(guān)的流程代碼

class Producer<Element> : Observable<Element> {
    
    override func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == 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
            }
        }
    }
}

final private class AnonymousObservableSink<Observer: ObserverType>: Sink<Observer>, ObserverType {
    typealias Element = Observer.Element 
    typealias Parent = AnonymousObservable<Element>

    // state
    private let _isStopped = AtomicInt(0)

    #if DEBUG
        fileprivate let _synchronizationTracker = SynchronizationTracker()
    #endif

    override init(observer: Observer, cancel: Cancelable) {
        super.init(observer: observer, cancel: cancel)
    }

    func on(_ event: Event<Element>) {
        #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()
            }
        }
    }

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

public struct AnyObserver<Element> : ObserverType {

    public typealias EventHandler = (Event<Element>) -> Void

    private let observer: EventHandler

    public init(eventHandler: @escaping EventHandler) {
        self.observer = eventHandler
    }
    
    public init<Observer: ObserverType>(_ observer: Observer) where Observer.Element == Element {
        self.observer = observer.on
    }
   
    public func on(_ event: Event<Element>) {
        return self.observer(event)
    }

    public func asObserver() -> AnyObserver<Element> {
        return self
    }
}

簡單梳理下上面的流程

可監(jiān)聽序列被訂閱了，就會調(diào)用Producer的subscribe方法
調(diào)用self的run方法资铡，當(dāng)前我們的self是Producer的子類AnonymousObservable
創(chuàng)建AnonymousObservableSink并將訂閱者observer傳進(jìn)去电禀，接著調(diào)用AnonymousObservableSink實(shí)例的run,參數(shù)是AnonymousObservable實(shí)例幢码。
AnonymousObservableSink實(shí)例的run又調(diào)用AnonymousObservable的_subscribeHandler閉包笤休，參數(shù)是AnyObserver,AnyObserver保存了事件的回調(diào)，而_subscribeHandler閉包也就是可監(jiān)聽序列創(chuàng)建時(shí)傳的閉包症副。

執(zhí)行事件

以next事件為例

如：訂閱onNext事件店雅。內(nèi)部流程如下：

調(diào)用observer的onNext方法，接著調(diào)用observer的on(_ event: Event<Element>)并將next事件傳進(jìn)去
執(zhí)行AnyObserver的observer的事件回調(diào)贞铣。
執(zhí)行AnonymousObservableSink的func on(_ event: Event<Element>)方法闹啦，接著調(diào)用ObserverBase的func on(_ event: Event<Element>)讓真正的Observer類去響應(yīng)事件
調(diào)用AnonymousObserver的onCore方法去執(zhí)行真正的事件回調(diào)。

總結(jié)

流程總結(jié)