RxSwift核心逻辑底层剖析(二)

RXSwift.jpg

注：代码块中的...是一些不需要显示的代码，直接用...替代了，rx的思维导图我后面重新画了一张，老漂亮了，画出经验了。

一.创建序列：

 let ob = Observable<Any>.create { (obserber) -> Disposable in
            obserber.onNext("onNext1")
            return Disposables.create()
        }

Observable遵守了ObservableType协议，create方法是协议的一个扩展方法。

create的具体实现可以在工程目录下搜索create.swift查看。

create方法返回了一个AnonymousObservable(subscribe)类，AnonymousObservable初始化保存了闭包。保存在了self._subscribeHandler属性中，self就是AnonymousObservable，也就是 AnonymousObservable. _subscribeHandler,这个很重要，后面会用到（函数式编程思想）。

final private class AnonymousObservable<Element>: Producer<Element> {
    typealias SubscribeHandler = (AnyObserver<Element>) -> Disposable

    let _subscribeHandler: SubscribeHandler

    init(_ subscribeHandler: @escaping SubscribeHandler) {
        self._subscribeHandler = subscribeHandler
    }

AnonymousObservable,继承于父类Producer，父类Producer继承于Observable，重写了subscribe方法。
下面是父类Observable ：

public class Observable<Element> : ObservableType {
    /// Type of elements in sequence.
    public typealias E = Element
    
    init() {
#if TRACE_RESOURCES
        _ = Resources.incrementTotal()
#endif
    }
    
    public func subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == E {
        rxAbstractMethod()
    }
    
    public func asObservable() -> Observable<E> {
        return self
    }
    
    deinit {
#if TRACE_RESOURCES
        _ = Resources.decrementTotal()
#endif
    }

解释：

func asObservable() ，返回一个self，实现了非序列向序列的转化，这个是万物皆序列的第二种解释，第一种解释前面的博客介绍过了。例如UISwitch().rx.value并不是一个序列，我们可以通过UISwitch().rx.value.asObservable() ,转成一个序列使用。

func subscribe实现的是一个抽象方法，里面没有具体的实现。

这里画了一张序列继承链的思维导图，供大家参考，如有不对的地方望不吝赐教：

序列继承链.png

二.序列执行订阅操作

 let _ = ob.subscribe(onNext: { (text) in
            print("订阅到:\(text)")
        }, onError: { (error) in
            print("error: \(error)")
        }, onCompleted: {
            print("完成")
        }) {
            print("销毁")
        }

ob.subscribe方法中就下面的这段代码需要我们去分析。

  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
            )

首先通过let observer = AnonymousObserver<E>{...}创建了一个观察者。AnonymousObserver<E>中的E来自于public func subscribe(onNext: ((E) -> Void)? = nil, ....)中的E，而这个E（序列类型），要从序列的subscribe方法中寻找，序列的subscribe方法是来自于：class AnonymousObservable<Element> ---->父类Producer ----> Observable<Element>

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

接着说let observer = AnonymousObserver<E>的AnonymousObserver类，let observer = AnonymousObserver<E>{...}这个其实是进行了一个初始化的操作， self._eventHandler = eventHandler保存了尾随闭包：

final class AnonymousObserver<ElementType> : ObserverBase<ElementType> {
    typealias Element = ElementType
    
    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，其实是一个基类订阅者，包含核心方法on， onCore发送消息，和dispose方法执行销毁操作。

这里我们能够看到，如果AnonymousObserver<E>后面的闭包能够被调用，就能实现序列执行订阅操作中尾随闭包的调动，也就完成了序列的响应,也就是说后面我们要搞清楚序列和订阅是如何关联上的。
而observer的调用是在：

 return Disposables.create(
                self.asObservable().subscribe(observer),
                disposable
            )

self.asObservable().subscribe(observer)中 self.asObservable()代表的就是序列,也就变成 ob.subscribe(observer)（ob是刚开始创建的序列）， ob的subscribe，需要到其父类（AnonymousObservable）的父类（Producer）中寻找。

class Producer<Element> : Observable<Element> {
    override init() {
        super.init()
    }

    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
            }
        }
    }
...
}

可以看到其实就是执行了一个 let sinkAndSubscription = self.run(observer, cancel: disposer) 的run方法，同时返回了一个销毁者,run的调用者不是AnonymousObservable就是Producer，所以查找run方法的实现，如下。

run方法的查找

以下是run方法的实现：

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)
    }
}

我们看到run方法中创建了一个let sink = AnonymousObservableSink(observer: observer, cancel: cancel)，保存了observer（订阅者）和cancel（销毁者）。然后调用了sink.run(self)，这里的self就是对应的类AnonymousObservable，也就是sink.run(AnonymousObservable)，也就是sink.run传入了一个序列，这一点非常重要，说明在sink中获得了序列，后面sink会把序列和订阅者关联。这个run方法的执行如下：

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

parent._subscribeHandler(AnyObserver(self)) 其实就是 AnonymousObservable._subscribeHandler(AnyObserver(self)),这个方法是不是很熟悉，是的，就是create初始化保存的代码块。这里我把run方法的整个代码块贴出来，因为非常重要：

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

    // state
    private let _isStopped = AtomicInt(0)

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

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

    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()
            }
        }
    }

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

AnonymousObservable._subscribeHandler(AnyObserver(self)),这个方法是怎么实现订阅者和序列的关联，主要是通过AnyObserver(self)，也就是 AnyObserver(AnonymousObservableSink)，在AnyObserver这个结构体中，自定义属性self.observer = observer.on（注意啦，这里的self.observer是一个结构体的一个function），保存了 observer的on方法的代码块，observer是sink传过来的，也就是

on方法

on 方法(AnonymousObservableSink类内的方法)的内部就是如下（核心逻辑）：

 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()
            }
        }
    }

上面的意思总结起来就是结构体AnyObserver的一个函数AnyObserver. observer保存了一个on函数块
回归到 AnonymousObservable._subscribeHandler(AnyObserver(self))，外界的obserber.onNext("onNext--1")，其实就是是调用的AnyObserver.onNext.而onNext其实就是调用了一个on方法（AnyObserver遵从了ObserverType协议，这个onNext是协议的扩展方法）

extension ObserverType {
    
    public func onNext(_ element: E) {
        self.on(.next(element))
    }
    
    public func onCompleted() {
        self.on(.completed)
    }
    
    public func onError(_ error: Swift.Error) {
        self.on(.error(error))
    }
}

这个on方法，其实就是前面提到的AnonymousObservableSink中的on方法，再上一次代码块：

final private class AnonymousObservableSink<O: ObserverType>: Sink<O>, ObserverType {
   ...
    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()
            }
        }
    }
...
}

走到这里，在执行switch，---> 里面的self.forwardOn(event)，调用的而其实是父类的forwardOn方法：

class Sink<O : ObserverType> : Disposable {
   ...
    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)
    }
   ...
}

最后通过self._observer.on(event)调用我们之前创建订阅者时的闭包（let observer = AnonymousObserver<E> {闭包} ）：

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()
                }
            }

然后就可以快乐的调用 onNext，onError，onCompleted输出print(订阅到....)了，终于完了，好累。以上就是RXSwift底层实现的核心逻辑了。如有不对的地方，烦请不吝赐教，本人不胜感激~~~

二次补充：学习到后面后，回来缕了一下逻辑，画了一张流程简化的新图：

RXSwift核心逻辑简化图.png

RxSwift核心逻辑底层剖析(二)