開啓RxSwift之旅——開篇

RxSwift 是 ReactiveX 在 Swift 下的實現。ReactiveX 是一個經過使用可觀察序列來組合異步和基於事件的程序的庫。

不少地方一般把 ReactiveX 稱爲 「函數響應式編程」 ，其實這是不恰當的。ReactiveX 能夠是函數式的，能夠是響應式的，可是和「函數響應式編程」是不一樣的概覽。一個主要的不一樣點是「函數響應式編程」是對隨着時間不停變化的值進行操做的，而 ReactiveX 對隨時間發射的離散值進行操做。

咱們先不急着去看 RxSwift 的源碼，在這以前，咱們有必要先了解一下什麼是響應式編程。

"什麼是響應式編程"

響應式編程是一種面向數據流和變化傳播的編程範式。

在某種程度上，這並非什麼新東西。用戶輸入、單擊事件、變量值等均可以看作一個流，你能夠觀察這個流，並基於這個流作一些操做。響應式就是基於這種想法。

一個流就是一個將要發生的以時間爲序的事件序列。它能發射出三種不一樣的東西：一個數據值(某種類型的)，一個錯誤（error）或者一個「完成（completed）」的信號。好比說，當前按鈕所在的窗口或視圖關閉時，「單擊」事件流也就「完成」了。

以一個單擊事件流爲例：定義一個針對數據值的函數，在發出一個值時，該函數就會異步地執行，還有一個針對發出錯誤時的函數，最後還有針對發出‘完成’時的函數。「監聽」流的行爲叫作訂閱。咱們定義的這些函數就是觀察者。這個流就是被觀察的主體(subject)（或「可觀察的(observable)」）。這正是觀察者設計模式。

在你使用 RxSwift 時，你就會發現它正是按照這種模式來進行設計的。在 RxSwift 中，一個流能夠被稱爲序列(Sequences)。序列的生產者就是 Observable 。

在 RxSwift 的 playground 中就有這麼一句話：

Every Observable instance is just a sequence.

Observable

若是你在學習 RxSwift 以前就使用過 ReactiveCocoa 的話，你會發現 RxSwift 和 ReactiveCocoa 徹底是兩個不一樣的物種。在 RxSwift 的世界裏，全部的東西都是 Observable 的。你能夠創造它們、操做它們，而後訂閱它們來響應變化。

理解 Observable 還有一件很重要的事情：

Observables will not execute their subscription closure unless there is a subscriber.

能夠這麼理解，若是你只是調用一個返回一個 Observable 的方法，生成序列不會被執行。Observable 只是一個解釋序列如何被生成和什麼參數被使用於生成元素的定義。生成序列開始於 subscribe 方法被調用的時候。

下面的例子中，Observable 的閉包永遠不會執行：

example("Observable with no subscribers") {
    _ = Observable.create { observer -> Disposable in
        print("This will never be printed")
        observer.on(.next("?"))
        observer.on(.completed)
        return Disposables.create()
    }
}

只有當咱們調用 subscribe(_:) 時，Observable 的閉包纔會執行：

example("Observable with subscriber") {
  _ = Observable.create { observer in
            print("Observable created")
            observer.on(.next("?"))
            observer.on(.completed)
            return Disposables.create()
        }
        .subscribe { event in
            print(event)
    }
}

上面例子中從傳入閉包建立一個 Observable ，到調用 subscribe(_:) 這個過程當中 RxSwift 到底作了什麼？咱們能夠先從簡單的 empty 開始。

empty

empty 就是建立一個空的 sequence, 它只能發出一個 completed 事件。

example(of: "empty") {
    Observable.empty()
        .subscribe({
            print($0)
    }) 
}

// 打印結果
--- Example of: empty ---
completed

上面代碼中經過 Observable 的 empty 方法建立了一個 Observable, 打開 Observable+Creation.swift 文件，能夠看到 empty() 的實現：

public static func empty() -> Observable {
    return EmptyProducer()
}

這裏返回了一個 EmptyProducer 的實例，點進去看看EmptyProducer是個什麼東西：

final class EmptyProducer : Producer {
    override func subscribe(_ observer: O) -> Disposable where O.E == Element {
        observer.on(.completed)
        return Disposables.create()
    }
}

EmptyProducer是 Producer 的子類，重寫了 subscribe(:) 。在 subscribe 方法中，觀察者訂閱了一個完成信號。

當咱們經過 empty() 建立了一個 Observable 後，而後會調用 subscribe(_:)，打開 ObservableType+Extensions.swift 文件, 能夠看到 subscribe 方法的實現：

public func subscribe(_ on: @escaping (Event) -> Void) -> Disposable {
    let observer = AnonymousObserver { e in
        on(e)
    }
    return self.subscribeSafe(observer)
}

subscribe 方法接受了閉包以後，先建立了一個匿名觀察者，subscribe 的閉包參數做爲構造器的參數傳給了 observer。點擊進去 AnonymousObserver源碼：

final class AnonymousObserver : ObserverBase {
    typealias Element = ElementType
    
    typealias EventHandler = (Event) -> Void
    
    private let _eventHandler : EventHandler
    
    init(_ eventHandler: @escaping EventHandler) {
#if TRACE_RESOURCES
        let _ = Resources.incrementTotal()
#endif
        _eventHandler = eventHandler
    }

    override func onCore(_ event: Event) {
        return _eventHandler(event)
    }
    
#if TRACE_RESOURCES
    deinit {
        let _ = Resources.decrementTotal()
    }
#endif
}

AnonymousObserver 的構造器接受一個閉包，而後在 onCore 方法中， 私有的 _eventHandler 會被調用。到這裏爲止，咱們仍是不知道咱們在調用 subscribe(_:) 時傳入的閉包最終的調用時機。不過已經很清楚的知道了，這個閉包在 onCore(:) 中調用了，咱們繼續進入 AnonymousObserver 的父類 ObserverBase 中一探究竟：

class ObserverBase : Disposable, ObserverType {
    typealias E = ElementType

    private var _isStopped: AtomicInt = 0

    func on(_ event: Event) {
        switch event {
        case .next:
            if _isStopped == 0 {
                onCore(event)
            }
        case .error, .completed:
            if AtomicCompareAndSwap(0, 1, &_isStopped) {
                onCore(event)
            }
        }
    }

    func onCore(_ event: Event) {
        rxAbstractMethod()
    }

    func dispose() {
        _ = AtomicCompareAndSwap(0, 1, &_isStopped)
    }
}

這一下就很清楚了，onCore(:) 會被 on(:) 調用。讓咱們再次回到 ObservableType+Extensions.swift 文件中，匿名觀察者(AnonymousObserver)建立完後，調用 subscribeSafe(:) 做爲函數返回值。在文件的最下面能夠看到 subscribeSafe(:) 的實現：

fileprivate func subscribeSafe(_ observer: O) -> Disposable where O.E == E {
    return self.asObservable().subscribe(observer)
}

這裏會調用 subscribe(:) ，注意了，這裏的 subscribe(:) 是 ObservableType 協議中定義的方法：

public protocol ObservableType : ObservableConvertibleType {
    
    associatedtype E
    
    func subscribe(_ observer: O) -> Disposable where O.E == E
}

這裏的參數是一個 ObserverType，也就是一個觀察者，千萬要與 func subscribe(_ on: @escaping (Event) -> Void) -> Disposable 作好區分。

好了， subscribe 方法將建立的匿名觀察者做爲參數，而在 EmptyProducer 中的 subscribe 的實現咱們已經看過了：

override func subscribe(_ observer: O) -> Disposable where O.E == Element {
    observer.on(.completed)
    return Disposables.create()
}

這裏恰好調用了觀察者的 on(:), 在 ObserverBase 中 on 方法會調用 onCore(:), onCore 方法調用了 subscribe(_ on: @escaping (Event) -> Void) -> Disposable 參數中的閉包。因爲 subscribe(_ observer: O) 中觀察者只訂閱了 "completed" 信號，全部閉包不會執行。

至此從建立一個 observable， 到調用 subscribe(_:) 整個過程咱們已經很清楚了。如今也就能明白爲何只是調用一個返回一個 Observable 的方法，生成序列不會被執行了。

小結

最後總結一下調用 subscribe(_:) 後的整個過程：用 subscribe 中的閉包建立一個匿名觀察者（觀察者私有的 _eventHandler 會將閉包保存起來），而後將建立的匿名觀察者做爲參數傳給 subscribeSafe(:) , subscribeSafe(:) 會調用 subscribe(:), 並將匿名觀察者做爲參數。subscribe(:) 會調用 observer 的 on(:), 當 observer 的 on 方法被調用後，最終會調用開始時傳入的閉包。

以上只是分析了一下 empty 的實現，像 of, just, create 等的實如今細節上有一些區別，總的思路是同樣的。在查看源碼時可能會有一點繞，主要是由於繼承太多，不少方法都要到父類中去找，並且 ObservableType 和 ObserverType 的 Extension 太多，代碼分散到各個文件中。

RxSwift 的代碼只看了個開頭，還有不少地方沒有徹底弄明白。在使用 RxSwift 的過程當中你能體會到 "響應式" 和 "函數式" 給咱們的開發帶來的便利性。