RxSwift5.0核心逻辑理解

作者: 数字d | 来源:发表于2020-02-05 17:32 被阅读0次

Demo代码实现内容

//1.创建序列
 let ob = Observable<Any>.create { (obserber) -> Disposable in
            // 3:发送信号
            obserber.onNext("框架")
            obserber.onCompleted()
//            obserber.onError(NSError.init(domain: "coocieeror", code: 10087, userInfo: nil))
            return Disposables.create()
        }
        
        
        // 2:订阅信号
        // AnonymousObserver  - event .next -> onNext()
        // _eventHandler
        // AnonymousObservable._subscribeHandler(observer)
        // 销毁
        let _ = ob.subscribe(onNext: { (text) in
//4.订阅到
            print("订阅到:\(text)")
        }, onError: { (error) in
            print("error: \(error)")
        }, onCompleted: {
            print("完成")
        }) {
            print("销毁")
        }

序列继承链

(class)Observable遵循了(protocol)ObservableType，在Create.swift文件中ObservableType通过扩展(extension)添加了create方法

extension ObservableType {

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

在create方法里，返回一个匿名序列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<O : ObserverType>(_ observer: O, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where O.E == Element {
      //管道，可以理解为一个Manager
        let sink = AnonymousObservableSink(observer: observer, cancel: cancel)
        let subscription = sink.run(self)
        return (sink: sink, subscription: subscription)
    }
}

AnonymousObservable是一个类（class），这个匿名序列继承自Producer

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

//重写父类的subscribe方法，订阅事件
    override func subscribe<O : ObserverType>(_ observer: O) -> Disposable where O.E == Element {
          ...
    }

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

Producer继承于Observable，Observable是可观察序列，遵循ObservableType协议，是所有序列的基类,ObservableType协议定义了一个函数subscribe，所有遵循这个协议的序列都具有subscribe的功能。

public class Observable<Element> : ObservableType {
    public typealias E = Element
    
    init() {
#if TRACE_RESOURCES
        _ = Resources.incrementTotal()//RxSwift自己的引用计数，创建就会增加，销毁就会减少
#endif
    }
    
    public func subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == E {
        rxAbstractMethod()//抽象方法，没有实现，等着被重写
    }
    
    public func asObservable() -> Observable<E> {
        return self
    }
    
    deinit {
#if TRACE_RESOURCES
        _ = Resources.decrementTotal()//根据这个可以知道是否存在内存问题
#endif
    }

    internal func composeMap<R>(_ transform: @escaping (Element) throws -> R) -> Observable<R> {
        return _map(source: self, transform: transform)
    }
}

Observable类本来就是一个可观察序列，为什么突然又出现一个asObservable函数呢？

为了统一管理，使得所有的类型都是序列类。

比如：UISwitch().rx.value，虽然UISwitch也可以.出rx.出value，但是查看value的底层遵循了ObservableType协议，也可以订阅事件，但是它始终不是Observable对象，为了规范统进行转化。“as”类似于将类型强转。


public var value: RxCocoa.ControlProperty<Bool> { get }   //是ControlProperty类型的
 
public struct ControlProperty<PropertyType> : ControlPropertyType {  //遵循了ControlPropertyType协议
    public typealias E = PropertyType
    ...
}
 
public protocol ControlPropertyType : ObservableType, ObserverType {   //遵循了ObservableType协议
 
    /// - returns: `ControlProperty` interface
    func asControlProperty() -> ControlProperty<E>
}

序列Observable提供所有功能，有asObservable（来源于协议），提供subscribe接口，使得所有继承于Observable的序列都具有订阅的功能。

2.png

订阅者继承链和订阅流程

Observable在subscribe订阅时会创建一个observer订阅者

 public func subscribe(onNext: ((E) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil)
        -> Disposable {
            let disposable: Disposable
            
              ...           

        创建一匿名订阅者，类型E是调用者传入的序列类型 
        let observer = AnonymousObserver<E> { event in
       ...
                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源码：

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继承于ObserverBase，ObserverBase的源码如下：

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

    private let _isStopped = AtomicInt(0)

    func on(_ event: Event<E>) { //实现ObserverType协议方法，Event是枚举类型
        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)
    }
}

Event源码：

public enum Event<Element> {
    /// Next element is produced.
    case next(Element)

    /// Sequence terminated with an error.
    case error(Swift.Error)

    /// Sequence completed successfully.
    case completed
}

ObserverBase具有不断发送信号的功能.

ObserverType是一个协议，源码如下

public protocol ObserverType {
    /// The type of elements in sequence that observer can observe.
    associatedtype E  //关联绑定类型

    /// Notify observer about sequence event.
    ///
    /// - parameter event: Event that occurred.
    func on(_ event: Event<E>) //协议方法
}

Producer.subscribe中的“self.asObservable().subscribe(observer)”到底由谁来进行调用呢？

不知道它是什么类型，但是为了统一，所以调用了asObservable，调用subscribe方法。AnonymousObservable里面没有subscribe方法，那么就开始去查找他的父类Producer，Producer继承了Observable，并且实现了subscribe这个抽象类的实现。

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

Producer统一处理避免子类的下发产生冗余代码，且并不影响基类，基类也不需要关心具体的实现。具体的run方法实现应该由子类单独去处理，体现出RxSwift的分工非常的明确。这里使用的AnonymousObersevable，所以具体的run实现源码在AnonymousObersevable里，源码如下：

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

4.png

核心重点

AnonymousObservable会调用父类Producer的subscribe函数，subscribe函数内，会调用AnonymousObservable具体的run方法，Create.swift文件中AnonymousObservable的run函数源码如下：

    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，Sink中就有observer和cancel这两个属性，在此进行保存。

class Sink<O : ObserverType> : Disposable {
    fileprivate let _observer: O
    fileprivate let _cancel: Cancelable
    fileprivate let _disposed = AtomicInt(0)

      ...

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

run函数中，sink调用了本类中的AnonymousObservable.run

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

typealias Parent = AnonymousObservable<E>

因此，AnonymousObservableSink这个管子拥有了订阅者、销毁者、序列、调度环境，处理的就是具体的业务逻辑。

    public static func create(_ subscribe: @escaping (AnyObserver<E>) -> Disposable) -> Observable<E> {
  // 外界传进来的闭包，传递给AnonymousObservable
        return AnonymousObservable(subscribe)
    }


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

    let _subscribeHandler: SubscribeHandler
  //  初始化
    init(_ subscribeHandler: @escaping SubscribeHandler) {
      //属性保存这个闭包
        self._subscribeHandler = subscribeHandler
    }
}

这里的parent就是我们的序列AnonymousObservable，调用_subscribeHandler函数就是我们一开始creat序列的时候外界传入的闭包。

run函数中_subscribeHandler(AnyObserver(self))中的AnyObserver(self)是什么呢？是不是AnonymousObserver呢？

AnyObserver是一个结构体，相当于一个序列化器，源码如下：

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<O : ObserverType>(_ observer: O) where O.E == Element {
//    这里的observer就是我们传进来的sink，当前的observer是一个函数
        self.observer = observer.on
    }
    
    public func on(_ event: Event<Element>) {
        return self.observer(event)
    }

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

observer.on就是调用了管子的on函数，核心逻辑就在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))
    }
}

AnyObserver这个结构体遵循了ObserverType协议，因此它也有onNext这个方法，而onNext这个方法就是调用了on函数，而这个on函数也就是调用的传进来的管子AnonymousObservableSink的on函数块，订阅者已经加载完毕。

extension ObserverType {
        public func onNext(_ element: E) {
        self.on(.next(element))
    }
    public func onCompleted() {
        self.on(.completed)
    }
    public func onError(_ error: Swift.Error) {
        self.on(.error(error))
    }
}

管子中的self.forwardOn(event)函数调用了父类Sink的forwardOn函数：

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

forwardOn函数中的_observer就是从订阅的时候Producer调用subscribe传过来的那个AnonymousObserver。

RxSwiftPic.png

源码地址：https://gitee.com/xgkp/RX2.0login.git

源码解读图片地址：https://gitee.com/xgkp/Rx_5.0.git

RxSwift5.0核心逻辑理解

序列继承链

订阅者继承链和订阅流程

Producer.subscribe中的“self.asObservable().subscribe(observer)”到底由谁来进行调用呢？

核心重点

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读