RxSwift 核心原理解析

角色定位

观察者（Observer）
被观察者（Observable）
订阅者（Subscriber）事件的最终处理者
管道（Sink） Observer 和 Observable 沟通的桥梁

Simple Example

Observable<String>.create { observer -> Disposable in
  observer.onNext("hello")
  return Disposables.create()
}
.subscribe { event in
  print(event.element)
}

我们重点关注：

create 闭包什么时候执行，
subscribe 闭包又是什么时候执行的

Observable 继承体系

Observable.png

Observable的核心函数:

subscribe 订阅操作, Observable 和 Observer 通过订阅建立联系,
Observable 好比水源， Observer 好比水龙头(永远开着的水龙头)，订阅的过程就是在Observable 和 Observer 之间建立管道，一旦建立管道即是永久性的，只要水源有谁，水龙头就会有水。
run 对用户不可见，隐藏了大量的实现细节，这个函数就是建立水管的过程
asObservable：这个协议的存在使得Observable 的定义变得更加广泛。
示例:

    /// GroupedObservable Class

    /// Converts `self` to `Observable` sequence. 
    public func asObservable() -> Observable<Element> {
        return source
    }

ObservableType 协议扩展

extension ObservableType {
    /**
     Subscribes an event handler to an observable sequence.
     
     - parameter on: Action to invoke for each event in the observable sequence.
     - returns: Subscription object used to unsubscribe from the observable sequence.
     */
    public func subscribe(_ on: @escaping (Event<E>) -> Void)
        -> Disposable {
            let observer = AnonymousObserver { e in
                on(e)
            }
            return self.asObservable().subscribe(observer)
    }
}

这里的GroupedObservable, 本身无法被直接订阅，但是它的属性source可以被订阅，于是只需要在asObservable 函数中返回source, 即可被订阅，这样的设计可以更自由使用组合方式，产生新的类型的Observable
ObservableType: 扩展 subscribe 方法，确保所有的Observable行为一致，都是经由self.asObservable() 获取Observable

Observer 继承体系

ObserverType.png

Observer 核心函数

on: 事件发生器
onCore: 对用户不可见，隐藏具体实现细节

Sink.png

Sink 核心函数

forwardOn
on
run

代码分析

先看看Observable.create 函数

/// static Observable create method
    public static func create(_ subscribe: @escaping (AnyObserver<E>) -> Disposable) -> Observable<E> {
        return AnonymousObservable(subscribe)
    }

该函数返回一个AnonymousObservable实例，这是因为Observable是一个抽象类，需要具体的实类实现功能，而AnonymousObservable是最简单的Observable类

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

    let _subscribeHandler: SubscribeHandler

    init(_ subscribeHandler: @escaping SubscribeHandler) {
        _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 -> Producer -> Observable -> ObservableType -> ObservableConvertibleType
每一层都只处理一点点事情，剩下的交给下一层处理

ObservableConvertibleType：完全的抽象

ObservableType：处理subscribe

Observable: 处理 asObservable

Producer: 重载subscribe

AnonymousObservable：处理run
也就是说如果说如果我们要自定义一个Observable的话，通常只需要继承Producer，并实现run方法。AnonymousObservable做的事情也不多，实现run方法，作为create闭包的持有者。

run方法涉及另外一个类AnonymousObservableSink,Sink作为Observer 和 Observable的衔接的桥梁，之前还在想为什么叫做ObservableSink，现在想明白了。Sink本身遵守ObseverType协议，与此同时实现了run方法，虽然没有实现subscribe方法，但是已经足够了，这样sink从某种程度来说也是Observable，通过sink就可以完成从Observable到Obsever的转变。

subscribe:

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

这里不太明白为什么一定要写成这样：

 let observer = AnonymousObserver { e in
                on(e)
            }

感觉直接写成let observer = AnonymousObserver(on) 会更加清晰。AnonymousObserver 比较简单持有subscribe闭包，并且实现onCore method.

刚才的只是个入口，这个才是核心

// Producer Class
  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 = 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
            }
        }
    }

这个if 和 else 不用管这个是线程安全相关的，不是我关注的重点。这里返回的是 Disposable , Disposable 跟对象的生命周期密切相关。暂且在这打个tag吧。

branch_1: 线程安全
branch_2: 生命周期管理
branch_3: 核心逻辑实现细节

好了我们现在关注的是核心逻辑实现细节，其他有时间再补上。核心语句
let sinkAndSubscription = self.run(observer, cancel: disposer)
我们来看看run的逻辑

// AnonymousObservable Class
    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)
    }

之前提到过AnonymousObservableSink,注意Sink是持有Observer的，从这也可以看出来Observerable的 run方法触发Sink的run方法，接下来就要关注AnonymousObservableSink方法。


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

哇，终于触发subscribeHandler了，这里的subscribeHandler就是之前最开始的闭包。

 Observable<String>.create { observer -> Disposable in
            observer.onNext("hello")
            return Disposables.create()
        }

也就是create闭包，真的是千辛万苦才跟踪到这了，那么现在解决了第一个问题： create闭包是什么执行的
现在就要关注这个闭包执行后会带来什么后果？
目光转到实体类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
    }
}

构造函数的入参是EventHandle,结合上下文 AnyObserver.observer = AnonymousObservableSink.on,那么这下就明白了 observer.onNext("hello") 最终会触发AnonymousObservableSink.on事件

// AnonymousObservableSink.on

    func on(_ event: Event<E>) {
        #if DEBUG
            _synchronizationTracker.register(synchronizationErrorMessage: .default)
            defer { _synchronizationTracker.unregister() }
        #endif
        switch event {
        case .next:
            if _isStopped == 1 {
                return
            }
            forwardOn(event)
        case .error, .completed:
            if AtomicCompareAndSwap(0, 1, &_isStopped) {
                forwardOn(event)
                dispose()
            }
        }
    }

老规矩忽略线程安全问题，关注核心问题

// Sink.forwardOn
    final func forwardOn(_ event: Event<O.E>) {
        #if DEBUG
            _synchronizationTracker.register(synchronizationErrorMessage: .default)
            defer { _synchronizationTracker.unregister() }
        #endif
        if _disposed {
            return
        }
        _observer.on(event)
    }

还记得_observer 的实体是啥，没关系帮你回忆一下：

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

我们最初传进来的subscribe闭包的实体类，理解了这个接下来就简单了

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

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

这里的_eventHandler 就是最初传进来的订阅闭包即：

            .subscribe { event in
               print(event.element)
        }

好了到此为止了，总结一下：
来个堆栈看的清晰些：

Rx subscribe源码分析.png

再来一个类图

Compose Diagram.png

稍微解释一下整个过程分为两个阶段：

Obsevable 构建阶段，这里使用create构造方法构造Obsevable，还有其他各种各样的构造方法，这里不一一赘述。
subscribe 阶段， Obsevable.subscribe ---> Obsevable.run ----> Sink.run ----> AnyObserver.On ----> Sink.on ----> Observer.On ---> Observer.OnCore ----> eventHandler

这个过程也可以分为多个阶段：

Obsevable.subscribe ---> Obsevable.run ----> Sink.run 这个过程我理解为build阶段，通过Sink建立Obsevable和 Observer的联系，在这个阶段Sink扮演还是Obsevable角色。

AnyObserver.On ----> Sink.on ----> Observer.On ---> Observer.OnCore 这个阶段的话Sink扮演的就是 Observer 角色