RxSwift (响应式编程)原理探究

作者: 暴躁的小豆子 | 来源:发表于2020-05-28 15:35 被阅读0次

为什么要学Rxswift：

swift语言的特性，swift是静态语言，静态语言对于我们相应的数值的传递，响应的回调，通讯等等系列会有一定影响，不用oc具备动态运行时，可以动态的发送一些消息
RxSwift响应式函数是编程的思想，将来是大家都需要掌握的东西，swift是ios开发接下来的重点，函数响应式编程也是swift未来的趋势，比如 combine
ReactiveX是综合跨平台框架，学会使用 RxSwift，再去学习 Rx.Net , RxJava 或者 RxJS 等就是小菜一碟，只是语言语法上的差异

RxSwift学习的必要性：

复合 - Rx 就是复合的代名词

复用 - 复用性比较强 - 代码量降低

清晰 - 因为声明都是不可变更，代码函数式编程可读性强
易用 - 理解容易，还抽象了异步编程，统一代码风格

稳定 - 因为Rx是完全通过单元测试的

装逼 - 代码的逼格很明显比原生的高太多了

RxSwfit核心

 //1.创建序列
        let obs = Observable<Any>.create { (obserber) -> Disposable in
           
            // 3 发送信号
            
            obserber.onNext("我是小豆子")
            obserber.onError(NSError.init(domain: "error", code: 10086, userInfo: nil))
//            obserber.onCompleted()
            return Disposables.create()
        }
        
        //2 订阅信号
        let _ =  obs.subscribe(onNext: { (text) in
            
            print("订阅到\(text)")
        }, onError: { (error) in
            
            print("error:\(error)")
        }, onCompleted: {
            print("完成")
        }) {
            print("销毁")
        }
 //输出
/*
订阅到我是小豆子
完成
销毁
*/

这里很明显的可以看到RxSwift的三步曲：

创建序列
订阅序列
发送序列
不难看出创建序列跟订阅序列都是闭包，想要探究原理必定是要追随着两个闭包是怎么调用的，在哪里调用，什么时机等

先来看看创建序列

 extension ObservableType {//ObservableType 协议并遵循ObservableConvertibleType协议
    // MARK: create
public static func create(_ subscribe: @escaping (AnyObserver<Element>) -> Disposable) -> Observable<Element> {
        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
    }

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

可观察序列是通过ObservableType协议拓展功能创建并返回一个匿名内部类AnonymousObservable，AnonymousObservable构造函数保存了subscribeHandler闭包

订阅序列

 public func subscribe(onNext: ((Element) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil)
        -> Disposable {
           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
            )
    }
    
class Producer<Element> : Observable<Element> { 
     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()
    }
}

订阅序列也是通过ObservableType协议拓展功能来实现的，创建一个匿名的观察者（AnonymousObserver）并返回一个垃圾袋

self.asObservable().subscribe(observer)
这个self是AnonymousObservable
subscribe是其父类Producer的方法
asObservable为了保持一致性
Producer类中的run 并没有实现而是需要子类具体实现AnonymousObservable.run(observer, cancel: disposer), run完后会创建一个管子（AnonymousObservableSink），管子的构造函数会存传进来的观察者（其实是调用父类Sink的初始化方法进行保存的），并调用管子的run（sink.run(self)）方法（代码块二中），此时的self是匿名可观察序列AnonymousObservable，可以看一下代码

final private class AnonymousObservableSink<Observer: ObserverType>: Sink<Observer>, ObserverType {
    typealias Element = Observer.Element 
    typealias Parent = AnonymousObservable<Element>
  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 {
        return parent._subscribeHandler(AnyObserver(self))
    }
}

** parent._subscribeHandler(AnyObserver(self)) ==> anonymousObservable. _subscribeHandler(AnyObserver(self)) ==> anonymousObservable. _subscribeHandler(AnyObserver(AnonymousObservableSink类型对象)) 此时就调用了一开始创建的可观察序列，是不是很牛逼，转了一圈又回来了，再往下看**

public struct AnyObserver<Element> : ObserverType {
 public typealias EventHandler = (Event<Element>) -> Void

    private let observer: EventHandler

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

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

从上面的代码段可以看到observer.on是一个函数，所以self.observer保存的是一个函数，AnyObserver遵从ObserverType协议

public protocol ObserverType {
    /// The type of elements in sequence that observer can observe.
    associatedtype Element

    @available(*, deprecated, renamed: "Element")
    typealias E = Element

    /// Notify observer about sequence event.
    ///
    /// - parameter event: Event that occurred.
    func on(_ event: Event<Element>)
}

/// Convenience API extensions to provide alternate next, error, completed events
extension ObserverType {
    
    /// Convenience method equivalent to `on(.next(element: Element))`
    ///
    /// - parameter element: Next element to send to observer(s)
    public func onNext(_ element: Element) {
        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))
    }
}

发送序列

从上端代码可以看到ObserverType协议拓展方法中，有onNext，onCompleted ，onError 其执行方法是协议中的on方法，on方法的实现是在遵循了ObserverType协议的AnyObserver结构体中

public struct AnyObserver<Element> : ObserverType {

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

截屏.png

上面已经说了self.observer保存的是一个函数，根据传进来的event，其实是走上面这个方法，

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 {
    fileprivate let _observer: Observer
  init(observer: Observer, cancel: Cancelable) {

        self._observer = observer
    }

    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.forwardOn(event)==>  self._observer.on(event)  这里的_observer 是订阅序列时创建的AnonymousObserver

AnonymousObserver.on(event) ==>AnonymousObserver.onCore(event)==>self._eventHandler(event), _eventHandler是一开始创建AnonymousObserver时保存的代码块
就是下面这个

 let observer = AnonymousObserver<Element> { 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()
                }
            }

最后贴一个结构图

结构图.png

RxSwift (响应式编程)原理探究

为什么要学Rxswift：

RxSwift学习的必要性：

RxSwfit核心

先来看看创建序列

订阅序列

发送序列

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Swift学习