RxSwift核心逻辑（一）-序列的产生以及订阅

RxSwift github地址

中文文档

前言

RxSwift是Swift的一套响应式编程框架，如同OC的RAC。
响应式编程的核心如下图：（来源于中文文档）。

描述了序列的产生、序列的订阅（绑定）、序列的操作符，以及订阅管理器Disposable（管理订阅的生命周期）。其实还有个Schedulers

image

Observable:
- 是一个可监听序列
- 可以理解为观察者模式里的被观察者
- 类图如下：
  
  image
Observer：
- 序列的监听者
- 可以理解为观察者模式里的观察者
- 类图如下：
  
  image
Disposable：可被清除的资源（管理绑定（订阅）的生命周期
Operator：序列操作符（这篇文章暂不做分析）
Schedulers:线程、队列的调度器

demo

看下面的demo，这里描述了，序列的产生和订阅的一个简单过程。

/// 创建一个序列
let observable = Observable<String>.create { (observe) -> Disposable in
    /// 生产一个事件
    observe.onNext("1")
    /// 生产一个事件
    observe.onNext("2")
    /// 发送一个错误事件
    //observe.onError(MyError.error1)
    /// 发送一个完成事件
    observe.onCompleted()
    /// 返回一个资源管理对象
    return Disposables.create()
}.observeOn(ConcurrentDispatchQueueScheduler.init(queue: DispatchQueue.global()))

/// 监听（订阅）序列    
observable.subscribeOn(MainScheduler.instance).subscribe(onNext: { (str) in
    /// 输出监听到的事件
    print(str)
}, onError: { ( error) in
    /// 输出错误信息
    print(error)
}, onCompleted: {
    print("订阅完成")
}).disposed(by: disposeBag)
#输出如下：
1
2
订阅完成

了解了RxSwift的序列的产生和消费之后，我们接下来探索下序列是如何产生的，又是如何消费的。

序列的产生

根据上面的demo，create创建了一个可监听序列。下面是create源码。

extension ObservableType {
    // MARK: create
    public static func create(_ subscribe: @escaping (AnyObserver<Element>) -> Disposable) -> Observable<Element> {
        return AnonymousObservable(subscribe)
    }
}

由上面源码看出，返回了一个AnonymousObservable类, 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)
    }
}

从上面可以看出，创建了一个可被监听的序列AnonymousObservable,并且AnonymousObservable保存了，序列监听者的回调。

序列的监听（订阅）

序列的监听：也就是上面的demo的 observable.subscribe...。

源码如下：

public func subscribe(onNext: ((Element) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil)
        -> Disposable {
            let disposable: Disposable
            
            if let disposed = onDisposed {
                disposable = Disposables.create(with: disposed)
            }
            else {
                disposable = Disposables.create()
            }
            
            #if DEBUG
                let synchronizationTracker = SynchronizationTracker()
            #endif
            
            let callStack = Hooks.recordCallStackOnError ? Hooks.customCaptureSubscriptionCallstack() : []
            
            let observer = AnonymousObserver<Element> { 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()
                }
            }
            return Disposables.create(
                self.asObservable().subscribe(observer),
                disposable
            )
    }

前面的一些Disposable管理，我们先不管，直接从let observer = AnonymousObserver<Element> { event in 这一行代码开始看。

这里创建了一个序列监听者类AnonymousObserver,并初始化了事件处理的回调。
回调里面根据不同的事件类型，来执行相应的事件。然后用资源管理器Disposables将可监听序列(observable)和序列监听者（observer）关联起来，

那么，是如何关联的呢？

关联Observable和Observer

源码如下：这里只贴相关的流程代码

class Producer<Element> : Observable<Element> {
    
    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
            }
        }
    }
}

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

    // state
    private let _isStopped = AtomicInt(0)

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

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

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

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

public struct AnyObserver<Element> : ObserverType {

    public typealias EventHandler = (Event<Element>) -> Void

    private let observer: EventHandler

    public init(eventHandler: @escaping EventHandler) {
        self.observer = eventHandler
    }
    
    public init<Observer: ObserverType>(_ observer: Observer) where Observer.Element == Element {
        self.observer = observer.on
    }
   
    public func on(_ event: Event<Element>) {
        return self.observer(event)
    }

    public func asObserver() -> AnyObserver<Element> {
        return self
    }
}

简单梳理下上面的流程

可监听序列被订阅了，就会调用Producer的subscribe方法
调用self的run方法，当前我们的self是Producer的子类AnonymousObservable
创建AnonymousObservableSink并将订阅者observer传进去，接着调用AnonymousObservableSink实例的run,参数是AnonymousObservable实例。
AnonymousObservableSink实例的run又调用AnonymousObservable的_subscribeHandler闭包，参数是AnyObserver,AnyObserver保存了事件的回调，而_subscribeHandler闭包也就是可监听序列创建时传的闭包。

执行事件

以next事件为例

如：订阅onNext事件。内部流程如下：

调用observer的onNext方法，接着调用observer的on(_ event: Event<Element>)并将next事件传进去
执行AnyObserver的observer的事件回调。
执行AnonymousObservableSink的func on(_ event: Event<Element>)方法，接着调用ObserverBase的func on(_ event: Event<Element>)让真正的Observer类去响应事件
调用AnonymousObserver的onCore方法去执行真正的事件回调。

总结

流程总结