RxSwift核心逻辑01 -- Observable Observer

思想

“一千个厨师作鱼香肉丝，会作出一千种味道”——杜甫
作鱼香肉丝的工序是固定的，可是随着条件不一样则会有不一样的结果，甚至出现制做失败的状况（error）。
这个过程用用RxSwift写的话基本是下面这个样子：

let 切菜 = 工序<Any>.create { (厨师) -> 收拾厨房的人 in
            切菜();
            厨师.作.给下一我的("切好的菜");
            厨师.作.完成();
            厨师.作.失败("没有刀！");
            return 收拾厨房的人.create();
        }

        let  炒菜 = 工序<Any>.create { (厨师) -> 收拾厨房的人 in
            炒菜();
            厨师.作.给下一我的("炒好的菜");
            厨师.作.完成();
            厨师.作.失败("没有锅！");
            return 收拾厨房的人.create();
        }
        
       let 鱼香肉丝 = 切菜.flatMap{(Any) -> 工序<Any> in 
               return 炒菜;
        };
        
        鱼香肉丝.制做(给下一我的: { (半成品) in
            装盘(未装盘的菜);
        }, 完成: {
             print(这个菜抄完了);
        }, 失败: {(error) in 
             print(这个菜没作完是由于:\(error));
            // 这个菜没作完是由于：没有刀 or 没有锅
        });
复制代码

这里咱们能够知道RxSwift的一个基本思路是，建立一个任务，建立这个任务的时候要把任务的结果、是否完成 以及 善后考虑好。而后在要作这个任务的时候，直接开启这个任务。就是在开启这个任务的时候，内部会自动分配一个匿名的执行者去执行这个任务。
这里面有3个主要成员：菜，厨师，收拾厨房的人。

而后看一段简单的正宗RxSwift调用
代码1.1：

let requestTask = Observable<Any>.create { (obserber) -> Disposable in
            obserber.on(.next("返回数据1"));
            obserber.onNext("返回数据2");
            return Disposables.create();
        }
        
        
        let _ = requestTask.subscribe(onNext: { (text) in
            print("执行结果:\(text)");
        });
复制代码

结构

这里面有三个核心

图1

*注：该图非官方，是笔者本身画的
绿色：Swift语言中的extension。
红色：枚举类型
蓝色：结构体

经过图1咱们能够看到RxSwift里面主要有三个核心协议，即<ObservableConvertibleType>的子类<ObservableType>，<ObserverType>和< Disposable >。用通俗的话讲就是“须要干的活的协议”，“干活的人的协议” 以及 “善后的人的协议”。

是否是被这么多类吓到了？其实稍微划分一下就很明朗了：

图2

总共有五大部分，每一个部分都有一个主要的类（用红框标出的）。
下面咱们一个一个的来细细分析

注意：请必定要看清楚是observable仍是observer
注意：请必定要看清楚是observable仍是observer
注意：请必定要看清楚是observable仍是observer

分析

0 - asXXXX

整个Rx内部有着大量的asXXX的方法，这里放在前面统一来讲：
随着不停地子类化，到后面履行该协议的类可能不止一个身份，也会有其余身份。为了可读性和逻辑性这里才有了这个方法协议。
举个例子：小明原先是程序员可是他转行成了一名厨师，因此小明如今的身份是一名厨师，厨师敲代码这很变扭可是：

小明.as程序员.敲代码()
复制代码

就很合理了。

1 - Event

经过箭头的指向的个数能够看出这个枚举是一个核心。这个枚举表示了一个事件的三种状况，即next, complete, error。这里面next和error都绑定了关联值。

2.1 -

“干活的人”的基本协议。

1. func on(_ event: Event<E>)
   做为一个响应者（干活的人）须要遵照的最基本协议，这里面只包含了这一个方法,而参数则是上面咱们提到的Event。

2. public func onNext(_ element: E) ，public func onCompleted() ，public func onError(_ error: Swift.Error)
   对于2.1.1的扩展，经过源码咱们能够知道仅仅是在发送相应的枚举值。而且在onNext和onError的时候，把参数赋值到枚举的关联值上。

2.2 - public struct AnyObserver

做为本篇惟一的结构体，之因此用结构体是由于它不须要被继承并且无需作析构处理。这里只有一个属性:(Event<Element>) -> Void类型的observer。

1. public init<O : ObserverType>(_ observer: O) where O.E == Element
   这个方法的实现是：

self.observer = observer.on
复制代码

这里很是巧妙，由于on方法的方法原型和observer的闭包类型都是

(Event<Element>) -> Void
复制代码

因此是能够赋值的。

1. public func on(_ event: Event<Element>)
   这里实现了协议的on方法，就是以event为参数调用本身的闭包对象observer。

3 - final class AnonymousObserver

基本属于2.2的增强版，由于不是结构体而是类，因此有了析构处理，并且自己也是继承自class ObserverBase<ElementType> 类。这里基本逻辑和2.2基本相似，惟一不一样的是他多了<Disposable>协议的实现，咱们会在后面的文章中专门讲<Disposable>的，这里不作过多赘述。

4.1 - < ObservableType >

“须要干的活”的基本协议。

1. func subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == E
   -- 做为全部Observable的最根本的协议，固然包含了最基本的功能subscribe。用上面作菜的例子说就是“一道菜(Observable)最起码是能被一个厨子(Observer)制做(subscribe)出来的,而且返回一个收拾厨房的人(Disposable),在适当的时候收拾残局”

2. public static func create(_ subscribe: @escaping (AnyObserver<E>) -> Disposable) -> Observable<E>
   -- 该协议扩展的简单工厂方法，经过一个闭包来快速建立一个AnonymousObservable类的实例。（AnonymousObservable后面会讲到）

3. public func subscribe(_ on: @escaping (Event<E>) -> Void) -> Disposable
   -- 对于方法1的扩展，经过一个(Event<E>) -> Void的逃逸闭包做为参数，来创造一个实现了ObserverType的类AnonymousObserver的实例，而后调用方法1。

4. public func subscribe(onNext: ((E) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil) -> Disposable
   -- 看似很复杂，可是其实真正的核心代码只有中间的AnonymousObserver建立，就是把做为参数的onNext，onError，onCompleted的闭包在对应枚举值的状况下传入关联值而且回调。而onDisposed闭包则是用来建立对一个的disposable，在适当的时候调用disposable.dispose()。在最后建立Disposables时，将先前的AnonymousObserver的对象做为参数调用本身的subscribe方法。

4.2 - class Producer

1. func run<O : ObserverType>(_ observer: O, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where O.E == Element
   -- 新添加的抽象方法。经过一个ObserverType的子类的对象和一个Cancelable（Disposable的子类）的对象为参数，返回两个Disposable的对象。

2. override func subscribe<O : ObserverType>(_ observer: O) -> Disposable where O.E == Element
   -- 协议方法的具体实现，里要注意的是下面这段代码：

let disposer = SinkDisposer()
let sinkAndSubscription = self.run(observer, cancel: disposer)
复制代码

这个方法本身生成了一个disposer，而后把这个disposer和observer做为参数调用了一下本身的run方法。

4.3 - final private class AnonymousObservable

1. _subscribeHandler 和 init

这个final类添加了一个属性

let _subscribeHandler: (AnyObserver<Element>) -> Disposable
复制代码

而构造方法则是把这个属性赋值

init(_ subscribeHandler: @escaping SubscribeHandler) {
        self._subscribeHandler = subscribeHandler
}
复制代码

再结合4.1.2咱们就能够知道这个_subscribeHandler，就是咱们调用creat的时候传过来的block。也就是代码1.1中的

obserber.on(.next("返回数据1"));
obserber.onNext("返回数据2");
return Disposables.create();
复制代码

1. 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)
复制代码

说明了两点：
1） AnonymousObservable的run方法实际上是调用的AnonymousObservableSink的run方法。
2）第一行说明了AnonymousObservableSink类的实例对象中可能包含了observer以及dispose。第二行说明了AnonymousObservableSink类的实例对象中又可能包含了observeable。
换句话说AnonymousObservableSink类是上面说起到的<Observer Type>,< ObservableType > 以及后面会说起的< Disposable >的汇合点。他是把“要干的活”，“干活的人”以及“处理者”给汇合到一块儿而且协调的那个枢纽

5. - final private class AnonymousObservableSink

当你们看到这个类的两个方法run和on的时候，有没有感受前面的一切都串联到了这里。由于run是Observable才有的方法，而on则是Observer才有的方法，这里一个类就包含了这两个方法，足以说明这个类做为枢纽的身份。

1. func on(_ event: Event<E>)
   --经过父类的方法咱们能够看出来，这个方法本质是调用observer的on，可是会根据上下文来适当的调用dispose。

2. func run(_ parent: Parent) -> Disposable
   --其实能够拆成两行代码

let anyObserver = AnyObserver(self);
parent._subscribeHandler(observer);
复制代码

你们注意，这里的anyObserver和自身属性的observer是有区别的。自身的observer是AnonymousObserver类（3）的对象。而anyObserver则是结构体AnyObserver（2.2）的对象。（具体区别在前边讲AnonymousObserver类的时候有提到）。
这两句代码就是生成一个AnyObserver的对象而且以其做为参数调用上面的_subscribeHandler闭包。
仍是拿代码1.1为例。

let requestTask = Observable<Any>.create { (obserber) -> Disposable in
            obserber.on(.next("返回数据1"));
            obserber.onNext("返回数据2");
            return Disposables.create();
        }
复制代码

这里的obserber就是AnyObserver(self)。

总结

总体RxSwift的核心很绕也很迷人，其实只要记住
Obserable - subscribe(Observer) - run(Observer)；
⬆️
on(Event)---Sink-------run(AnyObserver)
⬆️
Observer - on(Event)；
而且结合源码和UML图仍是相对容易理清思路的。

做为一个开发者，有一个学习的氛围和一个交流圈子特别重要，这是个人 交流群（111），你们有兴趣能够进群里一块儿交流学习

收录：原文地址

若是以为写得不错 给我一个赞谢谢