首先看一下RxSwift订阅最简单的代码：代码-001
//代码-001
// 1.创建序列
let ob = Observable<Any>.create { (observer) -> Disposable in
//3.发送信号
observer.onNext("走起")
return Disposables.create()
}
// 2. ob(Observable类型) 订阅序列
ob.subscribe(onNext: { (content) in
//4.响应信
print("订阅的内容：\(content)")
}) {
print("销毁")
}
从这段代码我们可以看出它的流程是：
创建序列->订阅序列->发送信号->响应信号
那么它内部的核心流程是怎样的，下面我们一起来探讨一下。
先来个流程图，后面内容不理解的可以跟着这张图捋捋思路：
一、创建序列
首先我们先进入create函数：代码-002
//代码-002
public static func create(_ subscribe: @escaping (AnyObserver<E>) -> Disposable) -> Observable<E> {
/// AnonymousObservable持有了逃逸性函数subscribe
return AnonymousObservable(subscribe)
}
我们发现create函数内部创建了一个AnonymousObservable类,并且将subscribe传入了该类。
进入AnonymousObservable类：代码-003
//代码-003
final private class AnonymousObservable<Element>: Producer<Element> {
typealias SubscribeHandler = (AnyObserver<Element>) -> Disposable
let _subscribeHandler: SubscribeHandler
init(_ subscribeHandler: @escaping SubscribeHandler) {
// 持有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)
}
}
发现该类保存了subscribeHandler。 它的父类是Producer(类) ->Observable(类)->ObservableType(协议)->ObservableConvertibleType(协议)。
二、序列订阅（绑定）
进入subscribe函数：代码-004
//代码-004
public func subscribe(onNext: ((E) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil)
-> Disposable {
/// 初始化 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()
}
}
return Disposables.create(
/// asObservable() 实例化 序列（Observable）
self.asObservable().subscribe(observer)
disposable
)
}
该函数内创建了一个对象observer 监听者，顾名思义该类就是用来监听信号的，一旦有信号发出该对象就会做出响应。
继续看return，返回值暂时不用关心，关心的是返回值里的参数self.asObservable().subscribe(observer)。代码-005
//代码-005
public class Observable<Element> : ObservableType {
public func asObservable() -> Observable<E> {
return self
}
}
asObservable()就是当前序列（AnonymousObservable类型）。
查看一开始贴出的AnonymousObservable类的代码（代码-003）发现该类内并没有subscribe，而是在他的父类producer内，注意这里传入的参数是observer(AnonymousObserver类型)：代码-006
//代码-006
class Producer<Element> : Observable<Element> {
override func subscribe<O : ObserverType>(_ observer: O) -> Disposable where O.E == Element {
let sinkAndSubscription = self.run(observer, cancel: disposer)
}
}
在当前Producer类subscribe函数内执行了run方法,而在一开始创建的Producer 的子类AnonymousObservable类内实现了该方法（查看代码-003），在该实现里创建了一个AnonymousObservableSink类型的对象sink，sink对象内传入了observer(AnonymousObserver类型)，创建完之后执行了run 函数。
AnonymousObservableSink类中：代码-007
//代码-007
final private class AnonymousObservableSink<O: ObserverType>: Sink<O>, ObserverType {
override init(observer: O, cancel: Cancelable) {
super.init(observer: observer, cancel: cancel)
}
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()
}
}
}
func run(_ parent: Parent) -> Disposable {
return parent._subscribeHandler(AnyObserver(self))
}
}
发现AnonymousObservableSink类 实现了run方法,并且继承之Sink类:代码-008
//代码-008
class Sink<O : ObserverType> : Disposable {
fileprivate let _observer: O
fileprivate let _cancel: Cancelable
fileprivate let _disposed = AtomicInt(0)
init(observer: O, cancel: Cancelable) {
#if TRACE_RESOURCES
_ = Resources.incrementTotal()
#endif
//持有了observer(AnonymousObserver类型)
self._observer = observer
self._cancel = cancel
}
final func forwardOn(_ event: Event<O.E>) {
if isFlagSet(self._disposed, 1) {
return
}
self._observer.on(event)
}
}
从这段代码的初始化方法里看出sink对象持有了observer对象(AnonymousObserver类型)。
继续看AnonymousObservableSink类 内 的run函数，parent就是代码-003中传过来的self(AnonymousObservable类型的对象)，所以_subscribeHandler就是一开始初始化序列时保存的subscribeHandler，终于发现subscribeHandler是在这里被调用了，这唤醒了第三部：发送信号，但是我们发现传入的是AnyObserver(self)而不是AnonymousObserver,self在这里是sink。下面我们来看下AnyObserver类：代码-009
//代码-009
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)
}
}
发现AnyObserver是一个结构体，遵循ObserverType协议。该结构体持有了observer.on闭包函数(此处的observer是sink).
下面我们来看第三步：发送信号
三、发送信号
从上面的分析可以看出代码-001的observer就是AnonymousObservableSink类(代码-007)内传入的AnyObserver，执行了onNext函数,但是该结构体并没有实现onNext函数，而是在ObserverType协议的扩展里面：代码-010
//代码-010
extension ObserverType {
public func onNext(_ element: E) {
self.on(.next(element))
}
}
onNext函数内调用了on并且传入了发送信号的值。因为该self就是AnyObserver类型，所以走到了AnyObserver结构体（代码-009）里面的on函数。
查看（代码-009）突然发现该函数竟然调用了self.observer，它不就是刚刚AnyObserver持有的observer.on（也就是sink.on）的闭包函数吗!所以又走到了AnonymousObservableSink类的on闭包函数(代码-007)。
所以执行到了 self.forwardOn(event),forwardOn的实现在它的父类Sink中(代码-008)。
至此我们发现该实现中执行了self._observer.on(event)代码。哇塞！该_observer不就是Sink持有的observer(AnonymousObserver类型)对象吗！不就是我们在subscribe(代码-004)函数中创建的监听者observer吗！666!!! 绕了一大圈终于回到了监听者的位置，进入AnonymousObserver类代码：代码-011
//代码-011
final class AnonymousObserver<ElementType> : ObserverBase<ElementType> {
typealias EventHandler = (Event<Element>) -> Void
private let _eventHandler : EventHandler
init(_ eventHandler: @escaping EventHandler) {
self._eventHandler = eventHandler
}
override func onCore(_ event: Event<Element>) {
return self._eventHandler(event)
}
}
从这段代码可以看到该类只实现了onCore方法，再进入他的父类ObserverBase,查看代码：代码-012
//代码-012
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)
}
}
}
func onCore(_ event: Event<E>) {
rxAbstractMethod()
}
}
到这可以长舒一口气了！on方法内调用了onCore，肯定进入到他的子类AnonymousObserver中，看代码 代码-011 可以发现 _eventHandler 闭包函数在此被调用，也就是 代码-004 中 AnonymousObserver的闭包得到执行，执行到了onNext闭包函数。
至此subscribe订阅的onNext闭包得到了调用，唤起最后一步：响应信号。
到此我们已经了解了整个流程，下面我们来终结一下：
总结
了解一下用到了哪些类、协议、结构体：
- 序列：
AnonymousObservable(类)->Producer(类)->Observable(类)->ObservableType(协议)->ObservableConvertibleType(协议)。
AnonymousObservableSink(类)->Sink(类)->Disposable(协议)
- 订阅者：
AnonymousObserver(类)->ObserverBase(类)->Disposable(协议)，ObserverType(协议)
AnyObserver(结构体)- >ObserverType(协议)
