RxSwift学习--核心逻辑再探

1,323

前言

通过上一篇内容RxSwift学习--核心逻辑初探,对RxSwift有了些初步的认知,下面通过源码来看一下RxSwift到底有多骚

RxSwift核心逻辑再探

先把上篇中的例子代码搬过来:

//第一步:创建序列
//在create()函数中传入一个闭包,任务是对每一个过来的订阅进行处理
 let ob = Observable<Any>.create { (observer) -> Disposable in
            // 第三步:发送信号(onCompleted和onError只能发送一个)
            observer.onNext("你好啊")
            observer.onCompleted()
//            observer.onError(NSError.init(domain: "loser", code: 10010, userInfo: nil))
            return Disposables.create()
 
//第二步:订阅信息
//当我们订阅了Observable的消息后,只要Observable的事件触发,都会通过onNext这个闭包告诉我们。
 let _ = ob.subscribe(onNext: { (text) in
            print("订阅到:\(text)")    //这里会监听到订阅的Observable事件
        }, onError: { (error) in
            print("error: \(error)")    //当发生错误时,会回调这里
        }, onCompleted: { // 当序列执行完毕时,会回调这里。
            print("完成")
        }) {
            print("销毁") 
        }

下面我们根据代码进行具体分析:

1.创建序列Observable<Any>.create()

在执行let ob = Observable<Any>.create { (obserber) -> Disposable in } 这句代码时,是创建了一个可观察序列,点进create()方法的源码

extension ObservableType {
    /*
     Creates an observable sequence from a specified subscribe method implementation.
    
     - seealso: [create operator on reactivex.io](http://reactivex.io/documentation/operators/create.html)
    
     - parameter subscribe: Implementation of the resulting observable sequence's `subscribe` method.
     - returns: The observable sequence with the specified implementation for the `subscribe` method.
     */

    public static func create(_ subscribe: @escaping (RxSwift.AnyObserver<Self.E>) -> Disposable) -> RxSwift.Observable<Self.E>
}

根据注释路由可知create()方法是在Create.swift文件中实现的

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

create()方法中返回了一个匿名内部类---AnonymousObservable(匿名序列),用来存储产生事件的闭包(self._subscribeHandler = subscribeHandler)和激活处理事件闭包的入口(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)
    }
}

可以看到在匿名序列的里面是把传给create()方法的闭包(也就是序列订阅的回调,self._subscribeHandler = subscribeHandler)保存下来了

2.序列订阅ob.subscribe(onNext:)

同样的,跟进去订阅信号的这个方法的源码, 在这里跟源码,跟进去的是ObserveableType.subscribe这个方法

subscribe.png

public func subscribe(onNext: ((E) -> 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<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()
                }
            }
            return Disposables.create(
                self.asObservable().subscribe(observer),
                disposable
            )
    }

可以看到在调用ob.subscribe()这个方法的时候,onNext, onError,onComplete, onDisposed这四个闭包都是作为参数传递到这个订阅函数中来, 在这段代码里同样的也创建了一个匿名观察者let observer = AnonymousObserver<E>用于存储和处理事件的闭包,而对于这个观察者会传入一个带有event的闭包,在前面已经知道event是一个枚举类型,有三种事件类型.next,.error,.completed,所以只要这个观察者调用了这三种事件,那么相应的就会调用onNext,onError,onCompleted这三个方法的具体实现,可是这里的observer观察者只是一个局部变量,它怎么和外界的观察者联系起来呢?

骚就骚在这个订阅方法的return这句代码:

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

其实这个self.asObservable()就是我们在最开始定义的序列let ob = Observable<Any>,而subscribe()就是回调了我们刚刚创建的observer闭包,而observer就会调用传入的监听序列消息闭包onNext, onError,onCompleted, 那么这个subscribe()是怎么回调了observer呢?

可知self.asObservable()AnonymousObservable类型的,但是通过前面的AnonymousObservable类的源码可以发现,它并没有subscribe()这个方法,但是AnonymousObservable是继承于Producer(生产者)的,下面来看一下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
            }
        }
    }

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

在这里面subscribe()方法会把传入的observer,调用self.run(observer, cancel: disposer);那么再次跟进去看源码;至于这个run()方法具体怎么调用,应该还是交给Producer的子类AnonymousObservable去实现的,那么跟进去,代码再次回到了AnonymousObservable类里面:

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对象并持有observer,然后这个AnonymousObservableSink类会调用它自己的run()方法,并传入self,再次跟进去sink.run(self),发现来到了AnonymousObservableSink类的源码;

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

可以看到AnonymousObservableSinkrun()方法,返回parent._subscribeHandler(),而这个Parent,在这个AnonymousObservableSink类中已经取了别名,那就是typealias Parent = AnonymousObservable<E>;在前面我们已经知道,AnonymousObservable匿名序列的里面是把传给create()函数的闭包保存了下来,那么return parent._subscribeHandler(AnyObserver(self))这个_subscribeHandler就是之前create()函数的闭包,在这个_subscribeHandler(AnyObserver(self))方法中把self转换成AnyObserver对象,也就是把AnonymousObservableSink对象转换成AnyObserver对象.

现在接着来看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
    }
}

通过刚刚的AnyObserver(self)我们应该知道这里的self指的是AnonymousObservableSink,所以在AnyObserver的源码中,可以看到在构造函数中有一行代码self.observer = observer.on,就是把AnonymousObservableSink类的on函数赋值给AnyObserverobserver变量.注意这里保存的是AnonymousObservableSink.on.

看到这里是否能明白,在调用ob.subscribe方法的时候,创建了一个AnonymousObserver对象,并把闭包传入,但是这个对象最终走向的却是继承于ProducerAnonymousObservable类的subscribe()方法,经过调用self.run()之后,最后返回AnonymousObservable._subscribeHandler()方法,也就是ob.create()方法。这部分代码先分析到这里,下面来分析下发送信号的代码。

3.发送信号observer.onNext("你好啊")

在刚开始分析create()方法的时候,就已经知道,observer.onNext("你好啊")中的observerAnyObserver类型的,可是在上面的AnyObserver源码中我们可以清楚地看到它并没有onNext()方法,那就去它的父类ObserverType看一下,

extension ObserverType {
    
    /// Convenience method equivalent to `on(.next(element: E))`
    ///
    /// - parameter element: Next element to send to observer(s)
    public func onNext(_ element: E) {
        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))
    }
}

在这里可以看到调用ObserverTypeonNext方法,返回ObserverTypeon(.next(element)),所以在observer.onNext("你好啊")调用onNext()方法,其实调用的就是observer.on(.next()),在前面的AnyObserver的源码中,已经知道AnyObserverobserver变量保存的是AnonymousObservableSink.on,同样的看一下在AnyObserver的源码中的on()方法

public func on(_ event: Event<Element>) {
        return self.observer(event)
    }

这里返回的是self.observer(event),也就是说observer.on(.next())可以是self.observer(.next()),然而self.observer = AnonymousObservableSink.on,所以self.observer(.next())可以是AnonymousObservableSink.on(.next()),这里又回到了AnonymousObservableSink这个类了,(AnonymousObservableSink的源码在前面已经出现了)这是饶了一圈又回来了,下面把AnonymousObservableSinkon(event)方法单独拿出来:

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

on(event)方法中有句重要的代码self.forwardOn(event),再跟进去self.forwardOn(event)这个方法,可以看到进入到Sink类的方法中,这里AnonymousObservableSink继承于Sink

 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)代码中,这个self._observer就是在初始化AnonymousObservableSink时候传入的observer( let sink = AnonymousObservableSink(observer: observer, cancel: cancel)),然而这个observer是在subscribe()订阅方法中由let observer = AnonymousObserver创建的observer,那么等于是这里又来到了AnonymousObserver中,那么下面就来看一下调用AnonymousObserver.on(event)会发生什么,现在来看一下AnonymousObserver的源码:

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)
    }
    
#if TRACE_RESOURCES
    deinit {
        _ = Resources.decrementTotal()
    }
#endif
}

通过这里的源码可以看到AnonymousObserver类中并没有on方法,但是AnonymousObserver是继承于ObserverBase的,在ObserverBase的源码中我们可以看到:

class ObserverBase<ElementType> : Disposable, ObserverType {
    typealias E = ElementType

    private let _isStopped = AtomicInt(0)

    func on(_ event: Event<E>) {
        switch event {
        case .next:
            if load(self._isStopped) == 0 {
                self.onCore(event)
            }
        case .error, .completed:
            if fetchOr(self._isStopped, 1) == 0 {
                self.onCore(event)
            }
        }
    }

    func onCore(_ event: Event<E>) {
        rxAbstractMethod()
    }

    func dispose() {
        fetchOr(self._isStopped, 1)
    }
}

调用ObserverBaseon()方法会返回self.onCore(event);这里回到子类AnonymousObserver去执行这个方法;

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

可以看到这里返回的是self._eventHandler(event);也就是AnonymousObserver._eventHandler(event)通过源码可知这里的AnonymousObserver._eventHandler(event)就是传入的eventHandler(event),也就是序列订阅方法ob.subscribe()传入的闭包,那么就来看下在subscribe()订阅方法中AnonymousObserver创建的observer到底对这个eventHandler(event)做了什么事情:


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

因为Event是枚举值这里会根据observer传入的event来判断到底该响应onNext(),onError(),onCompleted()的具体方法,这样就是把create()方法中的observer.onNext("你好啊")subscribe()订阅方法中onNext()联系起来了。这样就说明了为什么观察者调用了onNext()方法,序列能够订阅到这个方法里的内容。

总结

啰啰嗦嗦写了这么多,算是对RxSwift的核心逻辑有了个初步的认识,不等不说RxSwift很强大,足够的"骚气"。目前写到这里,对RxSwift仅仅只是略知皮毛,冰上一角罢了,对RxSwift的学习还需要继续努力。下面通过一张图来总结一下:

RxSwift核心逻辑