RxSwift底層核心邏輯、流程分析

時間  2019-11-08
標籤 rxswift 底層 核心 邏輯 流程 分析 简体版
原文   原文鏈接

介紹

咱們都知道，RxSwift很強，做爲一名開發者，閱讀開源框架的源碼，能讓咱們受益頗多，學習優秀的開發者的思路是頗有必要的。html

當咱們寫這麼一份代碼時，會疑惑爲何他兩個閉包之間會可以聯繫起來。git

Observable<String>.create { ob -> Disposable in
            ob.onNext("tets")
            return Disposables.create()
            }.subscribe(onNext: { str in
                print(str)
            }, onError: { error in
                print(error)
            }, onCompleted: {
                print("complete")
            })
複製代碼

本文結尾項目地址： github.com/GitHubYhb/R…
這是一份幫助閱讀RxSwift底層運行流程原理的RxSwift源碼 我往其中加了一些打印的信息，從create到onNext每一步在哪裏作了什麼都很清楚。 github

下載項目以後 -> 運行 RxExample-iOS

那麼廢話很少說。直接開幹。swift

Observable

Observable 原型是 Observable<Element>api

public class Observable<Element> : ObservableType {
    // ...部分省略
    
    // ObservableType 協議定義的方法。 
    public func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Element {
        rxAbstractMethod()
    }
    
    // ObservableConvertibleType 定義的方法。
    public func asObservable() -> Observable<Element> {
        return self
    }
}
複製代碼

它遵循了ObservableType協議。
在ObservableType協議中，要求遵循者須要有subscribe()方法。bash

public protocol ObservableType: ObservableConvertibleType {
    // ...省略一堆註釋
    func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Element
}
extension ObservableType {
    //遵循ObservableConvertibleType 的方法
    public func asObservable() -> Observable<Element> {
       // ...省略實現代碼
    }
}
複製代碼

同時ObservableType協議，又遵循了ObservableConvertibleType協議。 在ObservableConvertibleType協議中，建立了叫Element的關聯類型。
以及要求遵循者要有 asObservable()方法閉包

public protocol ObservableConvertibleType {
    associatedtype Element
    
    // E 從新命名 Element
    @available(*, deprecated, message: "Use `Element` instead.")
    typealias E = Element

    func asObservable() -> Observable<Element>
}
複製代碼

由下圖咱們能夠看出來這三者之間的關係框架

看懂了Observable,咱們再來看看create()，找到Create.swift文件ide

create()

extension ObservableType {
    // ... 省略一堆註釋
    public static func create(_ subscribe: @escaping (AnyObserver<Element>) -> Disposable) -> Observable<Element> {
        return AnonymousObservable(subscribe)
    }
}
複製代碼

吐槽：這句代碼是真的又長又難看懂。函數

create() 是什麼

首先，create()是ObservableType的擴展
上文已經有提到ObservableType是一個protocol協議

因此，create()也是ObservableType下定義的一個方法，跟subscribe()同樣。

解析

create()中有一個參數，名爲subscribe的閉包，而且用_作了省略。
返回型是Disposable。
看上去就是一長串，很不友好，爲了更好的理解，請看我下面的代碼

// 仿照 ObservableType 建立一個協議，並定義一個create方法
protocol TestProtocol {
    func create(_ subscribeBlock: @escaping (String) -> String) -> UILabel
}
// UILabel遵循這個協議
extension UILabel: TestProtocol{
    func create(_ subscribeBlock: @escaping (String) -> String) -> UILabel {
        //調用block
        let newStr = subscribeBlock("block幹活")
        print("block返回 == " + newStr)
        return UILabel.init()
    }
}
複製代碼

這份代碼中
(String) -> String
------------ ↓ 對應 ------------
(AnyObserver<Element>) -> Disposable
而後來測試一下這段代碼

let lb = UILabel.init()
lb.create { str in
    print("block傳遞出來 == " + str)
    return "test"
}
複製代碼

看打印結果

block傳遞出來 == block幹活
block返回 == test
複製代碼

相信看到這裏，你們內心已經有點明白的意思了。

  1. 在咱們調用完create()以後，閉包先返回給咱們一個AnyObserver<Element>類型的參數
  2. 咱們經過這個參數作了操做以後，就會再返回給閉包一個Disposable，給它操做。

那麼這個AnyObserver<Element> 是什麼呢

AnyObserver<Element>

咱們就看AnyObserver，它遵循了ObserverType協議
注意：這個協議跟上文的ObservableType不同，字母位數都不同的 - 0 -

public struct AnyObserver<Element> : ObserverType {
    //... 省略
}
複製代碼

先看看ObserverType

public protocol ObserverType {
    associatedtype Element
    @available(*, deprecated, message: "Use `Element` instead.")
    typealias E = Element
    func on(_ event: Event<Element>)
}
extension ObserverType {
    public func onNext(_ element: Element) {
        self.on(.next(element))
    }
    public func onCompleted() {
        self.on(.completed)
    }
    public func onError(_ error: Swift.Error) {
        self.on(.error(error))
    }
}
複製代碼

從代碼中咱們能夠看到ObserverType同ObservableType都定義了一個關聯類Element。
不一樣的地方在於定義的方法是func on(_ event: Event<Element>)，而且在拓展中定義了三個方便開發者使用的方法，也就是咱們平時經常使用的onNext()、onCompleted() 、onError()。

也就是說，咱們能夠拐彎抹角的這麼玩。

Observable<String>.create { ob -> Disposable in
    // 下面三行代碼一個意思
    ob.on(Event<String>.next("any"))
    ob.on(.next("test"))
    ob.onNext("tets")
    return Disposables.create()
}
複製代碼

既然已經到了Event了，不如先進去一探究竟

public enum Event<Element> {
    case next(Element)

    case error(Swift.Error)

    case completed
}
複製代碼

Event是一個枚舉，包含了咱們熟悉的三個枚舉值，next/error/completed
還有一系列的擴展方法 好比map()這裏先不一一列舉了。

回到AnyObserver
看看完整的代碼。

public struct AnyObserver<Element> : ObserverType {
    // 聲明閉包 EventHandler
    public typealias EventHandler = (Event<Element>) -> Void
    // 私有常量 observer
    private let observer: EventHandler
    // 初始化方法 1
    public init(eventHandler: @escaping EventHandler) {
        self.observer = eventHandler
    }
    // 初始化方法 2
    public init<Observer: ObserverType>(_ observer: Observer) where Observer.Element == Element {
        self.observer = observer.on
    }
    // on 方法
    public func on(_ event: Event<Element>) {
        return self.observer(event)
    }
    // asObserver 方法
    public func asObserver() -> AnyObserver<Element> {
        return self
    }
}
extension AnyObserver {
    typealias s = Bag<(Event<Element>) -> Void>
}

複製代碼

那麼create()方法聲明的部分就先到這裏。接下來咱們看看，他後續返回了AnonymousObservable(subscribe)

AnonymousObservable()

AnonymousObservable 翻譯過來就是 匿名觀察序列
看下面代碼能發現：

  1. 這是一個Create.swift中的私有類
  2. AnonymousObservable繼承自Producer<Element>
  3. 咱們上文中就是return了AnonymousObservable()初始化方法。
final private class AnonymousObservable<Element>: Producer<Element> {
    ...先不看他都幹了什麼
}
複製代碼

既然是繼承自Producer<Element>，那麼不如先看看Producer<Element>

class Producer<Element> : Observable<Element> {
    ... 也先不看
}
複製代碼

最終，他仍是Observable<Element>，函數要求的返回格式。

回到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)
    }
}
複製代碼

上文有說到create()中return了AnonymousObservable()初始化方法。
那麼咱們把注意力放到初始化方法
這裏他把咱們傳進來的閉包保存了起來。
那麼，咱們是否是能夠猜測，他保存起來以後，是否是能夠想何時用就何時用呢。

因此當前，咱們return了AnonymousObservable()初始化方法以後，create()已經執行完畢。

那麼如今咱們能夠這麼理解，下面這段代碼

Observable<String>
    .create { (ob) -> Disposable in
        ob.on(Event<String>.next("any"))
        return Disposables.create()
    }.subscribe(onNext: { str in
    }, onError: { (error) in
    }, onCompleted: {
    })
複製代碼

等同於

AnonymousObservable(subscribe)
    .subscribe(onNext: { str in
    }, onError: { (error) in
    }, onCompleted: {
    })
複製代碼

下一步就是subscribe()

subscribe()

在說以前，我得先跟你們說一下，接下來咱們會遇到的subscribe()有兩個，如圖所示。

咱們這裏就先把他們標記一下。

而後開始吧。

我作了一些打印，幫助我能更好的理解整個訂閱過程。
這張圖上主要作了兩點。

  1. 建立了AnonymousObserver，這個跟AnonymousObservable不同，AnonymousObserver是觀察者，而AnonymousObservable是被觀察對象。雖然建立出來，可是並無進行調用。
  2. 跟AnonymousObservable同樣，把這個閉包做爲一個臨時變量。經過self.asObservable().subscribe(observer)傳遞出去。

那麼被傳遞到哪裏去了呢。咱們看到self，這是咱們當前的調用對象AnonymousObservable，他這裏調用的subscribe(observer)是咱們上面作過標記的2 。

AnonymousObservable的subscribe()繼承自Producer

這裏主要是調用了run方法，又把傳進來的observer傳到別的地方，而後在當前的Producer裏面的run是個抽象方法，咱們回到AnonymousObservable，他重寫了這個run方法

能夠看到他建立了一個AnonymousObservableSink，初始化時又傳入了observer，一路跟進到Sink的初始化方法能夠看到

跟 AnonymousObservable時的操做相似，把 observer保存了起來。

在初始化AnonymousObservableSink後，有調用了另一個run方法。

在這個AnonymousObservableSink中，他重命名了AnonymousObservable
run方法接收到咱們傳遞過來的Parent(AnonymousObservable)
parent執行先前保存起來的閉包，並將AnyObserver(self)傳遞了進去，這樣才能讓保存起來的閉包有一個觀察者來執行Event。
這裏的self是AnonymousObservableSink，經過AnyObserver轉成觀察者。

能夠這麼理解：經過AnyObserver()將先前保存起來的observer，提取出來
這時候，已經把兩個保存起來的元素串聯起來了。

開始執行閉包中的內容。
這是最開始我寫好的閉包內容。

let ob = Observable<String>.create { ob -> Disposable in
            print("開始在咱們保存的閉包裏面搞事,ob的值已經傳進來了，就是AnyObserver(self)")
            ob.onNext("tets")
            return Disposables.create()
        }
        _ = ob.subscribe(onNext: { str in
            print(str)
        })
複製代碼

那麼，這個onNext是何時執行的呢？

onNext 何時執行

ob.onNext("tets")的本質是AnyObserver(self).onNext("tets")
self是AnonymousObservableSink
咱們須要看看AnyObserver(self)發生了什麼

public struct AnyObserver<Element> : ObserverType {
    public typealias EventHandler = (Event<Element>) -> Void

    private let observer: EventHandler

    public init<Observer: ObserverType>(_ observer: Observer) where Observer.Element == Element {
        print("調用AnyObserver(self)")
        self.observer = observer.on
    }
    ...
}
複製代碼

初始化方法中，將observer.on保存了起來，也就是說
把AnonymousObservableSink的on方法保存了起來
那麼這個on方法都作了什麼呢

func on(_ event: Event<Element>) {
        print("AnonymousObservableSink 的 on 方法 event == \(event.element!)")
        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()
            }
        }
    }
複製代碼

能夠看到on方法又調用了一個forwardOn，再點進去看看

final func forwardOn(_ event: Event<Observer.Element>) {
        //...省略
        self._observer.on(event)
}
複製代碼

記得咱們上文有提到在初始化AnonymousObservableSink時，把在subscribe()中建立的AnonymousObserver保存了起來，這裏他就經過這個_observer，調用了on方法。
這時才跳到了AnonymousObserver。

回到建立AnonymousObserver的地方

let observer = AnonymousObserver<Element> { event in
    ...省略不看
}
複製代碼

看看AnonymousObserver的自己

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

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

複製代碼

AnonymousObserver繼承自ObserverBase，貫徹到底

class ObserverBase<Element> : Disposable, ObserverType {
    private let _isStopped = AtomicInt(0)

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

    func onCore(_ event: Event<Element>) {
        rxAbstractMethod()
    }

    func dispose() {
        fetchOr(self._isStopped, 1)
    }
}
複製代碼

咱們能夠看到ObserverBase有一個on方法，還記得咱們最上面說過的嗎

Observable<String>.create { ob -> Disposable in
    // 下面三行代碼一個意思
    ob.on(Event<String>.next("any"))
    ob.on(.next("test"))
    ob.onNext("tets")
    return Disposables.create()
}
複製代碼

這三行是同一個意思。也就是說，在咱們調用onNext時，會先調用on方法，在調用.next，而後就走到了self.onCore,當前的onCore也是一個抽象方法。
咱們回到AnonymousObserver重寫onCore

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

self._eventHandler(event)就會執行這段紅框代碼

直到這裏，才執行到了onNext，纔會到咱們本身在外部寫的代碼。

看起來可能很亂，這裏作一下

總結

  1. 咱們寫的閉包中ob.onNext()調用時，首先調用了AnonymousObservableSink的on方法
  2. AnonymousObservableSink的on方法調用了self.forwardOn(event)
  3. forwardOn(event)經過先前保存起來的_observer，調用了AnonymousObserver的on方法
  4. AnonymousObserver的on方法調用重寫父類的self.onCore(event)
  5. onCore(event)調用AnonymousObserver的self._eventHandler(event),這個_eventHandler就是上面紅框裏面的內容被保存了下來。
  6. 紅框內接收到self._eventHandler(event)傳遞來的event，判斷event類型，執行咱們本身在外部寫的內容。

或許仍是很亂，我調整了一下個人打印內容，一步一步的都寫得很清楚。

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程