新手读的懂的RxSwift源码解析（一）-- 事件的产生与监听（

作者: iAllen | 来源:发表于2021-06-17 15:35 被阅读0次

上一篇中，我们了解了RxSwift的核心概念与核心逻辑，有兴趣的读者可以了解一下：
新手读的懂的RxSwift源码解析（零）-- 核心概念与核心逻辑
本篇文章，笔者将与各位读者一起学习一下下一个课题：RxSwift中事件的产生与传递。
我们通过一个最常用的操作符：create来说明。

        Observable<Int>.create { observer in
            //订阅闭包
            observer.onNext(1)
            observer.onNext(2)
            observer.onNext(3)
            observer.onCompleted()
            return Disposables.create()
        }
        .subscribe { value in
            //监听闭包1
            print("Next:\(value)")
        } onError: { error in
            //监听闭包2
            print("Error:\(error)")
        } onCompleted: {
            //监听闭包3
            print("Cpmpleted")
        } onDisposed: {
            //监听闭包
            print("Disposed")
        }
        .disposed(by: disposeBag)

为了便于理解，我们把第一个闭包称之为订阅闭包，后面几个闭包称之为监听闭包。

一、事件的产生

上面的代码中，我们调用了create，传入一个订阅闭包。这个闭包接受一个AnyObserver参数observer，这个AnyObserver是一个遵循了ObserverType协议的struct。在这个订阅闭包闭包中，我们调用了这个observer的onNext方法和onCompleted方法，于是也就产生了一个由三个next事件和一个完成事件的序列。那这个闭包是在什么时候被调用的呢？

先看一下这个create的实现：

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

create接受一个订阅闭包，并且返回一个AnonymousObservable。还是没有看到这个订阅闭包是在什么时候调用的，也就不清楚事件序列是什么时候产生的。那我们只能继续往下看，看看事件的监听过程。

二、事件的监听

通过上一篇，我们已经知道了，我们可以通过调用ObservableType的subscribe方法对事件进行监听。下面我们就来看看这个subscribe方法干了些什么。

根据上面的分析，我们知道，我们通过create创建了一个AnonymousObservable，这正是一个ObservableType。之后我们调用了它的subscribe方法, 但并不是协议定义的:

func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Element

而是协议扩展中提供的一个方法：

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

这个方法主要就是创建了一个AnonymousObserver，并通过self.asObservable().subscribe(observer)订阅了这个AnonymousObserver。由于这里的self是一个AnonymousObservable，而它是继承于Observable的，所以这个asObservable()实际是返回了self。

所以我们大致了解了这个调用链条的第二步干了些什么：通过传入的订阅闭包，创建了一个AnonymousObserver并传给了第一步创建的AnonymousObservable的subscribe方法。那我们接下来就要看一下这个subscrib方法又干了些什么。

为了搞清楚，我们就得进入到这个AnonymousObservable的源码。它是继承Producer类的，而Producer类又是继承于Observable类的，也就是这样：

AnonymousObservable => Producer => Observable

为了搞清楚这个AnonymousObservable是个什么，我们就得先简单介绍一下它的父类Producer。

Producer

Producer 是Observabled的一个子类，它是一个抽象类。从名称来看，我们大概可以猜测，它是负责生产事件序列的类。

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

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

    func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Element {
        rxAbstractMethod()
    }
}

Producer继承Observable,并重写了subscribe方法。我们主要关注这么几行：

    let disposer = SinkDisposer()
    let sinkAndSubscription = self.run(observer, cancel: disposer)
    disposer.setSinkAndSubscription(sink: sinkAndSubscription.sink, subscription: sinkAndSubscription.subscription)

这里首先创建了一个SinkDisposer对象，调用自己的run方法，将这个SinkDisposer以及subscribe方法的参数observer传给run方法，这个SinkDisposer以及run方法的返回值主要是用于资源释放相关的，不属于这篇文章要讨论的重点。
看到这里，我们知道了，通过调用subscribe，我们会调用到AnonymousObservable的run方法，run方法就是核心逻辑的起点，那下面我们就深入到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持有了第一步create中创建的订阅闭包。

其次，AnonymousObservable作为Producer的子类，重写了run方法，在这个run方法中，创建了一个AnonymousObservableSink，并调用了这个AnonymousObservableSink的run方法。

这里又涉及到了另一个概念，叫做Sink。这个Sink其实也是RxSwift里一个比较重要的概念，它主要是作为一个桥梁，连接了Observable和Observer，每个Observable类型，都会通过一个Sink来与它的Observer连接并传递事件。

AnonymousObservableSink

final private class AnonymousObservableSink<Observer: ObserverType>: Sink<Observer>, ObserverType {
  
    ......

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

我们看到，这个run方法主要干了两件事。1，使用self作为参数，创建了一个AnyObserver，2.调用了parent（即AnonymousObservable）的subscribeHandler。至此，我们就知道了，第一步里面创建的订阅闭包是何时调用的了:

在AnonymousObservable.subscribe方法中通过调用self.run，最终调用AnonymousObservableSink.run方法，在run方法中调用了订阅闭包，而在这个订阅闭包中，我们便产生了一些列的事件。

其实，RxSwift中的其他一些Observable，也大部分都是类似的流程，这个我们会在后续文章中一一分析。

接下来还有一个问题，事件产生了，也被监听了，那事件是如何传递到我们的第二步提供的监听闭包的呢？我们之前说了，Observable和Observer通过Sink进行连接，也就是通过Sink进行了事件的传递，那我们就来看一下这个Sink。

Sink

class Sink<Observer: ObserverType>: Disposable {
    fileprivate let observer: Observer
    fileprivate let cancel: Cancelable
    private let disposed = AtomicInt(0)

    #if DEBUG
        private let synchronizationTracker = SynchronizationTracker()
    #endif

    init(observer: Observer, cancel: Cancelable) {
#if TRACE_RESOURCES
        _ = Resources.incrementTotal()
#endif
        self.observer = observer
        self.cancel = cancel
    }

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

    final func forwarder() -> SinkForward<Observer> {
        SinkForward(forward: self)
    }

    final var isDisposed: Bool {
        isFlagSet(self.disposed, 1)
    }

    func dispose() {
        fetchOr(self.disposed, 1)
        self.cancel.dispose()
    }

    deinit {
#if TRACE_RESOURCES
       _ =  Resources.decrementTotal()
#endif
    }
}

我们看到,Sink持有了一个Observer，并在自己的核心方法forwardOn中会将事件传递给所持有的这个observer，这个forwardOn是个final方法，会在不同的子类中被调用。下面我们就再来看一下，前述代码中所产生的AnonymousObservableSink

AnonymousObservableSink

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
        private 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 {
        parent.subscribeHandler(AnyObserver(self))
    }
}

之前我们已经分析过，这个AnonymousObservableSink的run方法中使用self作为参数，创建了一个AnyObserver，并调用了订阅闭包，将这个AnyObserver传给了订阅闭包。所以当订阅闭包中调用observer.onNext和observer.onComplete的时候，实际会调用的这个AnyObserver的on方法，那这个on方法又是怎么传递到后面的监听闭包的呢？

这个得看看AnyObserver的源码：

public struct AnyObserver<Element> : ObserverType {
    /// 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<Observer: ObserverType>(_ observer: Observer) where Observer.Element == Element {
        self.observer = observer.on
    }
    
    /// Send `event` to this observer.
    ///
    /// - parameter event: Event instance.
    public func on(_ event: Event<Element>) {
        self.observer(event)
    }

    /// Erases type of observer and returns canonical observer.
    ///
    /// - returns: type erased observer.
    public func asObserver() -> AnyObserver<Element> {
        self
    }
}

我们看到，前面的AnyObserver(self)（注：这里的self也就是前述的AnonymousObservableSink），实际上会把参数self.on传给这个AnyObserver的observer闭包，而在AnyObserver的on方法中，直接调用了这个observer闭包，于是就将事件传递到了AnonymousObservableSink的on方法中。而
AnonymousObservableSink的on方法会调用forwardOn方法，forwardOn方法又会调用AnonymousObservableSink所持有的observer的on方法：

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

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

而这个self.observer，正是subscribe方法中生成的AnonymousObserver,它的on方法便会把事件一一传递到监听闭包中。

三、总结

至此，我们已经彻底搞清楚了事件的产生、传递以及监听。
1.Observable.create：创建AnonymousObservable
2.Observable.subscribe(onNext:onError:onCompleted:onDisposed:): 创建AnonymousObserver，并调用AnonymousObservable.subscribe(_ observer:)
3.AnonymousObservable.subscribe(_ observer:)：调用Producer.subscribe(_ observer:)
4.Producer.subscribe(_ observer:): 调用AnonymousObservable.run
5.AnonymousObservable.run: 创建AnonymousObservableSink, AnonymousObservableSink.observer = AnonymousObserver, 并调用AnonymousObservableSink.run
6.AnonymousObservableSink.run：创建AnyObserver,调用AnonymousObserver.subscribeHandler，传入AnyObserver
7.AnonymousObserver.subscribeHandler：调用AnyObserver.on
8.AnyObserver.on：调用self.observer，即AnonymousObservableSink.on
9.AnonymousObservableSink.on: 调用Sink.forwardOn
10.Sink.forwardOn: 调用AnonymousObservableSink.observer.on，即AnonymousObserver.on
11.AnonymousObserver.on：调用监听闭包

码字不易，若有错漏，欢迎指正。若有助益，烦请点赞。^ _ ^

新手读的懂的RxSwift源码解析（一）-- 事件的产生与监听（

一、事件的产生

二、事件的监听

Producer

AnonymousObservable

AnonymousObservableSink

Sink

AnonymousObservableSink

三、总结

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

RxSwift