Swift学习

RxSwift-核心逻辑

2020-06-06  本文已影响0人  恍然如梦_b700

RxSwift核心流程

RxSwift这个优秀的框架，设计的api也是非常精简，让陌生的用户也能非常快速上手

// 1: 创建序列
_ = Observable<String>.create { (obserber) -> Disposable in
    // 3:发送信号
    obserber.onNext("消息字符串")
    return Disposables.create()  // 这个销毁不影响我们这次的解读
    // 2: 订阅序列
    }.subscribe(onNext: { (text) in
        print("订阅到:\(text)")
    })

// 控制台打印：“订阅到:消息字符串”

为什么我们的消息字符串这个字符串会在订阅序列的subscribe的闭包打印。下面是代码分析

RxSwift核心逻辑

创建序列

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

大家可以很清晰看到我们的 可观察序列 的创建是利用协议拓展功能的create方法实现的，里面创建了一个 AnonymousObservable（匿名可观察序列） 命名还是体现了作者的思维 ：这个类就是一个内部类，具备一些通用特性（具有自己功能的类才会命名） 下面我贴出这个类的继承关系。

我们可以清晰的看到的继承关系。那么这么多的内容还有那么多层嵌套，这个地方我们需要掌握什么：

订阅序列

这里说明这个订阅方法 subscribe 和我们上面所说的 subscribe 不是同一个方法

来自于对 ObservableType 的拓展功能

extension ObservableType {
    public func subscribe(onNext: ((E) -> Void)? = nil, ...) -> Disposable {
           // 因为篇幅 省略不影响我们探索的代码
            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
            )
    }
}

代码说明：

E 这里的意思是 Swift 的关联类型，这个如果仔细看过可观察序列的继承链源码应该不难得出：这个E 就是我们的 序列类型，我们这里就是String

public class Observable<Element> : ObservableType {
    /// Type of elements in sequence.
    public typealias E = Element

创建了一个 AnonymousObserver (匿名内部观察者) 手法和我们的 AnonymousObservable 差不多，它这里的初始化是闭包参数，保存了外界的 onNext, onError , onCompleted , onDisposed 的处理回调闭包的调用,下面我还是给大家贴出 观察者 的继承链关系，帮助大家理解

override func subscribe(_ observer: O) -> Disposable where O.E == Element {
        if !CurrentThreadScheduler.isScheduleRequired {
            // 篇幅原因，我们省略一些代码，方便我们理解
            ...
            return disposer
        }
        else {
            return CurrentThreadScheduler.instance.schedule(()) { _ in
                let disposer = SinkDisposer()
                let sinkAndSubscription = self.run(observer, cancel: disposer)
              // 篇幅原因，我们省略一些代码，方便我们理解
              ...
                return disposer
            }
        }
    }

override func run (...) {
    let sink = AnonymousObservableSink(observer: observer, cancel: cancel)
    let subscription = sink.run(self)
    return (sink: sink, subscription: subscription)
}

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

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

发送响应

我们从上面的分析，非常清晰:
obserber.onNext("消息字符串") 的本质是： AnyObserver.onNext("消息字符串")

这时候发现我们的AnyObserver 是没有这个方法，这很正常！一般思路，找父类，找协议

extension ObserverType {
    public func onNext(_ element: E) {
        self.on(.next(element))
    }
}

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

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

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

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

判断 event 进而调用 onNext?(value) ,因为枚举的关联值（Swift很强大的功能）value = "消息字符串", 接下来就是外界 onNext闭包的调用传参，那么这个时候源码解析到这里，我相信你已经完全掌握了RxSwift的核心逻辑，最后这里附上我们的分析图解

image.png

总结：RxSwift的结构

上一篇 下一篇

猜你喜欢

热点阅读