RxSwift核心原理Observable& Observer(一)

理解函数响应式编程思想
RxSwfit核心内容

按照RxSwfit的说法万物皆序列，所以我们先来探讨Observable序列的产生和Observer监听响应

SimpleCode
let observable = Observable<Any>.create { (observer) -> Disposable in
            observer.onNext("来了")
            observer.onCompleted()
            return Disposables.create()
        }
            
 let _  =  observable.subscribe(onNext: { (text) in
              print("订阅到：\(text)")
         }, onError: { (error) in
              print("error: \(error)")
        }, onCompleted: {
              print("完成")
       }) {
              print("销毁")
      }
#########
订阅到：来了
完成
销毁
#########

我们关注：

  • create闭包什么时候执行
  • subscribe闭包什么时候执行
Observable 继承关系

AnonymousObservable--->Producer--->Observable--->ObservableType--->ObservableConvertibleType

Observer继承关系

AnonymousObserver--->ObserverBase--->Disposable、ObserverType

Sink继承关系

AnonymousObservableSink--->Sink/ObserverType

代码分析

Observable.create

// Create.swift
extension ObservableType {
    // MARK: create

    /**
     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 (AnyObserver<E>) -> Disposable) -> Observable<E> {
        return AnonymousObservable(subscribe)
    }
}

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

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

Observable.create
在 extension ObservableType中我们看到create函数返回一个可观察序列，具体是AnonymousObservable（匿名可观察序列）；
在 class AnonymousObservable中初始化时传入一个逃逸闭包self._subscribeHandler，继续找寻看哪里有调用；
在class AnonymousObservableSink中发现有parent._subscribeHandler(AnyObserver(self))；
在尾随逃逸闭包处发现传入的参数为AnyObserver，所以进入Observable.subscribe继续探寻。

//  ObservableType+Extensions.swift

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

Observable.subscribe
临时变量let observer = AnonymousObserver为观察者,保存闭包AnonymousObserver._eventHandler；
subscribe的返回值中有调用self.asObservable().subscribe(observer)，也就是AnonymousObservable.subscribe(observer),根据继承关系，也就是Producer.subscribe(observer)。

//  Producer.swift
//  class Producer<Element> : Observable<Element>
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(observer, cancel: disposer),AnonymousObservable中有重写

// Create.swift 
//  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)，调用到AnonymousObservableSink类中'run'.

// Create.swift
// final private class AnonymousObservableSink<O: ObserverType>: Sink<O>, ObserverType
func run(_ parent: Parent) -> Disposable {
        return parent._subscribeHandler(AnyObserver(self))
    }

这里传入的参数并不是我们在Observable.subscribe中创建的observer，而是AnyObserver(AnonymousObservableSink),我们发现AnonymousObservableSink同时还继承自Sink<O>,Observable.subscribe中创建的AnonymousObserver传入了Sink

// Create.swift
// final private class AnonymousObservableSink<O: ObserverType>: Sink<O>, ObserverType
override init(observer: O, cancel: Cancelable) {
        super.init(observer: observer, cancel: cancel)
 }

// Sink.swift
// class Sink<O : ObserverType> : Disposable
init(observer: O, cancel: Cancelable) {
#if TRACE_RESOURCES
        _ = Resources.incrementTotal()
#endif
        self._observer = observer
        self._cancel = cancel
}

通过执行parent._subscribeHandler(AnyObserver(self))此时已经来到了Observable.create尾随逃逸闭包的调用

let observable = Observable<Any>.create { (observer) -> Disposable in
    observer.onNext("来了")
    observer.onCompleted()
    return Disposables.create()
}

执行observer.onNext("来了")或者observer.onCompleted()

//  ObserverType.swift

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

/// Convenience API extensions to provide alternate next, error, completed events
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))
    }
}

这里将onNext(element)等方法转换成了self.on(.next(element)),同时上面执行observer.onNext("来了")或者observer.onCompleted()的observer为
AnyObserver(AnonymousObservableSink),来到AnonymousObservableSink
我们发现其实现了protocol方法

// Create.swift
// final private class AnonymousObservableSink<O: ObserverType>: Sink<O>, ObserverType 

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

继续执行self.forwardOn(event)

// Sink.swift
// class Sink<O : ObserverType> : Disposable  

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

    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为Observable.subscribe中创建的AnonymousObserver，所以调用self._observer.on(event)，为AnonymousObserver.on(event),其自身没有实现on方法
来到其父类的ObserverBase.on

//  ObserverBase.swift
//  class ObserverBase<ElementType> : Disposable, ObserverType

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

self.onCore(event)来到AnonymousObserver.onCore(event)

//  AnonymousObserver.swift
//  final class AnonymousObserver<ElementType> : ObserverBase<ElementType> {

init(_ eventHandler: @escaping EventHandler) {
#if TRACE_RESOURCES
        _ = Resources.incrementTotal()
#endif
        self._eventHandler = eventHandler
    }

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

调用self._eventHandler(event)就来到了Observable.subscribe中闭包的调用

let _  =  observable.subscribe(onNext: { (text) in
                print("订阅到：\(text)")
            }, onError: { (error) in
                print("error: \(error)")
            }, onCompleted: {
                print("完成")
            }) {
                print("销毁")
        }

总结

这个篇幅是探讨RxSwift中Observable可观察序列和Observer观察者的调度流程
①创建可观察序列 Observable<Any>.create( subscribe: (AnyObserver<Any>) -> Disposable):
产生一个AnonymousObservable序列，同时把尾随逃逸闭包保存在AnonymousObservable的self._subscribeHandler中;

②发起订阅subscribe(onNext: ((Self.E) -> Void)? = nil, onError: ((Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil):
产生一个AnonymousObserver,通过 ObservableType.asObservable().subscribe(observer)
--->AnonymousObservable.subscribe(observer)
--->Producer.subscribe(observer)
--->AnonymousObservable .run(observer)
--->AnonymousObservableSink .run()/AnonymousObservableSink (observer)
--->AnonymousObservable._subscribeHandler(AnyObserver(AnonymousObservableSink))/Sink.init(observer)
--->来到Observable<Any>.create尾随逃逸闭包的执行/把AnonymousObserver存入Sink._observer中；

③信号发送observer.onNext("来了")：
这里的observer为AnyObserver(AnonymousObservableSink);
observer.onNext(element)
--->AnyObserver.onNext(element)
--->ObserverType.onNext(element)
--->ObserverType.on(.next(element))
--->AnonymousObservableSink.on(.next(element))
--->Sink.forwardOn(event)
--->Sink._observer.on(event)
--->AnonymousObserver.on(event)
--->ObserverBase.on(event)
--->ObserverBase.onCore(event)
--->AnonymousObserver.onCore(event)
--->AnonymousObserver._eventHandler(event)来到Observable.subscribe中闭包的调用

这里Sink在run之前是Obsevable的角色后续是Observer角色

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 213,047评论 6赞 492
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 90,807评论 3赞 386
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 158,501评论 0赞 348
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 56,839评论 1赞 285
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 65,951评论 6赞 386
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 50,117评论 1赞 291
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 39,188评论 3赞 412
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 37,929评论 0赞 268
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 44,372评论 1赞 303
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,679评论 2赞 327
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,837评论 1赞 341
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 34,536评论 4赞 335
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 40,168评论 3赞 317
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,886评论 0赞 21
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,129评论 1赞 267
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 46,665评论 2赞 362
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 43,739评论 2赞 351