RxSwift原理1 - 订阅与发布

看了几篇相关的讲解，还是有点理不清楚，所以决定自己再梳理一遍。

初识

先看一段简单的调用：

//创建序列
let ob = Observable<Any>.create { (observable) -> Disposable in
    observable.onNext("on Next")
    observable.onCompleted()
    return Disposables.create()
}
//订阅消息
let _ = ob.subscribe { info in
    print("next: \(info)")
} onError: { error in
    print(error)
} onCompleted: {
    print("completed")
} onDisposed: {
    print("销毁")
}

看一下输出：

// next: on Next
// completed
// 销毁

看着还挺简单，参考其他文章，看了20行，还行，看了25行，嗯，看不下去了，脑容量不够了。
所以我决定在深入了解他的实现之前，先查一下他的户口。

查户口

在看其他文章是，我看到最多的就是：

订阅者(观察者)：Observer
发布者(被观察者): Observable

至于其他的亲朋好友，暂且忽略，问就是我脑容量不够，看不过来。
为什么要先查户口呢，因为无论是订阅者还是发布者，其都是多层继承的关系，如果不理清楚父类子类各自做了哪些事，就很难去理解他的实现。

Observable

yuque_mind.jpeg

丑是丑了点，将就着看。至于ObservableType继承自ObservableConvertibleType为什么不画，可以画，但是没必要。

Observer

yuque_mind.jpeg

好了，两个重要人物的户口已经查完了，是不是豁然开朗，哦~，原来是这样啊(ps:好像什么都没讲到啊)
户口查完，就看是看他们如何组成RxSwift这个家庭的吧

交往

发布者创建了AnonymousObservable对象，并保存了对象了发布闭包

//创建序列
let ob = Observable<Any>.create { (observable) -> Disposable in
    observable.onNext("on Next")
    observable.onCompleted()
    return Disposables.create()
}

// 看一下ObservableType中create实现
public static func create(_ subscribe: @escaping (AnyObserver<Element>) -> Disposable) -> Observable<Element> {
    AnonymousObservable(subscribe)
}

//最终是创建了AnonymousObservable实例，并保存了创建序列中的闭包
init(_ subscribeHandler: @escaping SubscribeHandler) {
    self.subscribeHandler = subscribeHandler
}

OK，现在看一下订阅消息做了什么：

创建了AnonymousObserver订阅者，并保存了对应的发布闭包

//订阅消息
// ob是一个AnonymousObservable对象
// 返回值dispose，暂时不管
let _ = ob.subscribe { info in
    print("next: \(info)")
} onError: { error in
    print(error)
} onCompleted: {
    print("completed")
} onDisposed: {
    print("销毁")
}

执行subscribue方法, 创建了订阅者，并保存了订阅闭包

// 稍微看一下subscribe实现(删减版)
public func subscribe(
        onNext: ((Element) -> Void)? = nil,
        onError: ((Swift.Error) -> Void)? = nil,
        onCompleted: (() -> Void)? = nil,
        onDisposed: (() -> Void)? = nil
    ) -> Disposable {
        let disposable: Disposable = ...
        //创建Observer
        let observer = AnonymousObserver<Element> { event in            
            switch event {
            case .next(let value):
                onNext?(value)
            case .error(let error):
                if let onError = onError {
                    onError(error)
                }
                else {
                    ...
                }
                disposable.dispose()
            case .completed:
                onCompleted?()
                disposable.dispose()
            }
        }
        return Disposables.create(
            // self是AnonymousObservable发布者，
            // asOBservable转换是为了调用subscribe方法，本质还是AnonymousObservable
            // AnonymousObservable未实现该方法，因此执行的是父类Producer的subscribe方法
            // subscribe有多个，参数不一样
            self.asObservable().subscribe(observer),
            disposable
        )
}
// 内部创建了AnonymousObserver，可以看到其保存了事件闭包
init(_ eventHandler: @escaping EventHandler) {
#if TRACE_RESOURCES
    _ = Resources.incrementTotal()
#endif
    self.eventHandler = eventHandler
}

AnonymousObservable.asObservable,虽然类型转换，氮气本质还是AnonymousObservable。所以调用subscribe时是调用其父类Producer的实现

在发布者的subscribe中，调用了run方法

//所以这里应该是调用的Producer方法
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
        }
    }
}

run方法中创建了sink管道，sink保存了observer订阅者

注意，虽然是在Producer调用的run方法，但是此时的实例实际是AnonymousObservable，所以我们看到的应该是AnonymousObservable的run方法

override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Element {
    //创建sink管道，并保存对应的observer和dispose
    let sink = AnonymousObservableSink(observer: observer, cancel: cancel)
    let subscription = sink.run(self)
    return (sink: sink, subscription: subscription)
}

调用sink run方法，执行一开始保存的发布闭包，并传入sink作为参数

//sink中的run实现，这里的parent也就是AnonymousObserver订阅者
func run(_ parent: Parent) -> Disposable {
    //subscribeHandler则是一开始创建序列中保存的闭包
    //这里的self，则是上一步创建的sink管道，而sink是实现了ObserverType
    //所以闭包中的onNext，最终都是调用sink的on方法，最终调用到observer的on方法
    parent.subscribeHandler(AnyObserver(self))
}

执行发布闭包中对应的onNext方法，触发on方法，触发sink的on方法

public func onNext(_ element: Element) {
    self.on(.next(element))
}

/// Convenience method equivalent to `on(.completed)`
public func onCompleted() {
    self.on(.completed)
}

/// Convenience method equivalent to `on(.error(Swift.Error))`
/// - parameter error: Swift.Error to send to observer(s)
public func onError(_ error: Swift.Error) {
    self.on(.error(error))
}

最终触发执行到observer订阅闭包

//简单看一下sink的on方法实现
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()
        }
    }
}
final func forwardOn(_ event: Event<Observer.Element>) {
    #if DEBUG
        self.synchronizationTracker.register(synchronizationErrorMessage: .default)
        defer { self.synchronizationTracker.unregister() }
    #endif
    if isFlagSet(self.disposed, 1) {
        return
    }
    可以看到最后触发了订阅者的方法
    self.observer.on(event)
}

见家长

最后把整个流程梳理一下吧

yuque_mind.jpeg

按照黑色的箭头先执行
最后执行绿色箭头

最终发布闭包和订阅闭包最终是通过sink管道来实现通信