RXSwift-探索RXSwift核心逻辑（一）

2019-07-29 本文已影响0人 Ldies

随着函数响应式编程成为现在很多编程语言的主流编程思想，我们越发的感叹这种思想的魅力，他极大的减少了我们的胶水代码，使得我们可以更好的编写业务逻辑代码。而作为我们iOS程序员来说，RAC以及RxSwift框架使用的也越来越多，我们不再需要写大量的代理方法来作为方法回调，我们可以将一个功能模块代码写在一起，通过链式来表达，使代码逻辑给人一种清晰可见的感觉。

RxSwift

作为ReactiveX家族之一的RxSwift在Github截止现在Star:16K.为什么这个框架如此受欢迎，作为函数响应式框架典型代表，底层实现又是如何实现的呢？这一篇文章全面解密

RxSwift核心流程

Rxswift的优秀之处在于他为开发人员提供了非常精简的API，让我们可以快速上手来使用，它的使用公有三个步骤：

1.创建序列
2.订阅序列
3.发送序列

            //1.创建序列
            let ob = Observable<Any>.create { (obserber) -> Disposable in
            // 3:发送序列
            obserber.onNext("AAA")
            obserber.onCompleted()
            return Disposables.create()
        }
        //2.订阅序列
        let _ = ob.subscribe(onNext: { (text) in
            print("订阅到:\(text)")
        }, onError: { (error) in
            print("error: \(error)")
        }, onCompleted: {
            print("完成")
        }) {
            print("销毁")
        }

image.png

这里我们肯定很疑惑为什么我们的AAA字符串会被传入subscribe的闭包当中，从这里我们开始一步步的分析它里边到底是如何实现的。

创建序列

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

我们看到 可观察序列的创建是通过ObservableType的协议拓展功能的create方法实现的，并且返回了一个匿名内部类AnonymousObservable,而AnonymousObservable中保存了外界传入的subscribe闭包，下面是一附类的继承关系图

image.png

如上图所示
AnonymousObservable继承了Producer可观察序列生产者

final private class AnonymousObservable<Element>: Producer<Element> {
    typealias SubscribeHandler = (AnyObserver<Element>) -> Disposable
    
    let _subscribeHandler: SubscribeHandler
    //保存外界传入的观察者对象
    init(_ subscribeHandler: @escaping SubscribeHandler) {
        self._subscribeHandler = subscribeHandler
    }
}

Producer类继承自Observable可观察序列,并保存了_subscribeHandler
SubscribeHandler为外部传入的观察者对象AnyObserver<Element>

public class Observable<Element> : ObservableType {
    /// Type of elements in sequence.
    public typealias E = Element
    
    init() {
#if TRACE_RESOURCES
        _ = Resources.incrementTotal()
#endif
    }
    
    public func subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == E {
        rxAbstractMethod()
    }
    
    public func asObservable() -> Observable<E> {
        return self
    }

Observable又遵循了ObservableConvertibleType和ObservableType协议
ObservableType
所以AnonymousObservable匿名内部类拥有Observable非常重要的方法subscrible

订阅序列

我们先看下subscribe内部实现

image.png

        public func subscribe(onNext: ((E) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil)
        -> Disposable {
            //这里省略部分代码，我们只分析observer(观察者的创建)
            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
            )
    }

代码说明：

我们创建了一个AnonymousObserver匿名内部观察者，通过初始化闭包，保存了外部传入的onNext,onError ,onCompleted ,onDisposed事件。我们先看一下观察者的继承链

image.png

self.asObserverable()这个函数来保持一致性的写法
self.asObservable().subscribe(observer)通过可观察序列的继承关系，我们可以快速的定位producer订阅代码
image.png
image.png

image.png

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

这里的self.run代码最终由我们的生产者producer延伸到我们的具体事务代码AnonymousObservable.run中

final private class AnonymousObservable<Element>: Producer<Element> {

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

sink.run（self）将业务处理下沉，让分工更具明确

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

parent为我们传入的AnonymousObservable对象
这里我们就可以清晰的看到AnonymousObservable._subscribeHandler的执行，我们的疑惑终于在此时解开了，这就是为什么我们的subscrible订阅序列会执行我们创建序列里的 observer观察者序列闭包,是的是的绕了一大圈终于回来了。。。

        let ob = Observable<Any>.create { (observer) -> Disposable in
            // 3:发送信号
            observer.onNext("AAA")
            observer.onCompleted()
//            obserber.onError(NSError.init(domain: "coocieeror", code: 10087, userInfo: nil))
            return Disposables.create()
        }

AnyObserver(self)的初始化方法里保存了一个非常重要的函数AnonymousObservableSink .on 函数

public init<O : ObserverType>(_ observer: O) where O.E == Element {
    self.observer = observer.on
}

发送事件响应

从上述流程我们清晰的得知，observer.onNext("AAA")的本质是调用了AnyObserver.onNext("AAA"),我们来看下源码

public struct AnyObserver<Element> : ObserverType {
    public init<O : ObserverType>(_ observer: O) where O.E == Element {
        self.observer = observer.on
    }
    public func on(_ event: Event<Element>) {
        return self.observer(event)
    }
}

self.observer构造初始化就是：AnonymouservableSink.on函数
这样我们就可以将 self.observer(event)变形：->AnonymousObservableSink.on(event)->AnonymousObservableSink.on(.next("AAA"))

*AnonymousObservableSink我们这里把它就先称为管子，来说说管子它都干了些啥，看源码：

class AnonymousObservableSink<O: ObserverType>: Sink<O>, ObserverType {
    func on(_ event: Event<E>) {
        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()
            }
        }
    }
}

self.forwardOn(event)我们先看下这句代码的父类做了什么事

class Sink<O : ObserverType> : Disposable {
    final func forwardOn(_ event: Event<O.E>) {
        if isFlagSet(self._disposed, 1) {
            return
        }
        self._observer.on(event)
    }
}

其中 self._observer就是我们初始化保存的 AnonymousObserver匿名观察者，怎么样？发现了吗？激动吗？我们又回到了订阅序列创建时AnonymousObserver里的闭包调用里

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

通过调用onNext?(value),此时value的值就是我们外界传入的AAA啊。。

绕了这么一大圈，我们通过对可观察序列的创建，订阅序列，发送事件最外层的调用，再到他们内部各种协议的配合，下沉业务逻辑，接口隔离让我们着实体会了RxSwift是多么优秀的框架，相信通过这篇对RxSwift的核心逻辑的探究，会让各位小伙伴对RxSwift的设计思路有一个清晰的认知。最后贴给小伙伴们整个RxSwift的分析图

image.png

RXSwift-探索RXSwift核心逻辑（一）

RxSwift

RxSwift核心流程

创建序列

订阅序列

代码说明：

发送事件响应

猜你喜欢

热点阅读