Observable
我们之前说过，RxSwift的核心就是可观察序列Observable，在RxSwift中，事件的订阅、传递都是通过Observable。
public class Observable<Element> : ObservableType {
/// Type of elements in sequence.
public typealias E = Element
public func subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == E {
rxAbstractMethod()
}
public func asObservable() -> Observable<E> {
return self
}
}
通过代码可以看出，Observable定义了一个泛型，也就是说观察序列中的元素是泛型，万物皆观察序列，不论什么类型的元素都可以被观察。
Observable中还定义了一个subscribe的方法，但是该方法是一个抽象方法，需要它的子类来实现。方法中的observer就是观察者，订阅观察者就可以接收此序列的事件。序列可以生产0个或者多个元素，就会有0个或者多个事件发送给观察者。一旦发送Error或者Completed事件，该序列就会终止，内部资源就会被释放。事件可以从不同的线程发出，但是会分前后顺序发送给观察者，也就是可以异步发送，但是只能同步接收。
Observable又分为有限观察序列和无限观察序列。
- 有限观察序列，观察之前就明确观察的次数
func testLimitedSequence() {
let ob = Observable.just([1,2,3,4])
ob.subscribe(onNext: { (num) in
print(num)
}, onCompleted: {
print("订阅完成")
}).disposed(by: disposeBag)
}
控制台输出结果：
[1, 2, 3, 4]
订阅完成
可以看到，有限观察序列在观察次数完成之后，就会收到订阅完成的事件。
- 无限观察序列，即观察的次数不明确。比如下面的定时器，就会一直观察下去。
func testUnlimitedSequence() {
let timer = Observable<Int>.interval(1, scheduler: MainScheduler.instance)
timer.subscribe(onNext: { (num) in
print(num)
})
.disposed(by: disposeBag)
}
事件流程
func testObservable() {
// 1. 创建观察序列
let ob = Observable<Any>.create { (obserber) -> Disposable in
// 3. 发送信号
obserber.onNext("发出信号")
obserber.onCompleted()
return Disposables.create()
}
// 2. 订阅事件
let _ = ob.subscribe(onNext: { (text) in
print("订阅到:\(text)")
}, onError: { (error) in
print("error: \(error)")
}, onCompleted: {
print("完成")
}) {
print("销毁")
}
}
可以看出，整个观察序列的流程是：
- 创建观察序列
- 订阅观察
- 发送事件
- 销毁观察序列
创建观察序列
创建观察序列，参数是一个逃逸闭包，闭包中是订阅观察的实现。返回的是一个AnonymousObservable对象。
public static func create(_ subscribe: @escaping (AnyObserver<E>) -> Disposable) -> Observable<E> {
return AnonymousObservable(subscribe)
}
我们先看看AnonymousObservable的继承关系：
在初始化AnonymousObservable的时候，会把闭包中内容保存到_subscribeHandler的变量中。
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)
}
}
订阅观察
可以看到在订阅观察的时候，会有一个匿名观察者AnonymousObserver在内部处理相关的事件。
public func subscribe(onNext: ((E) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> 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)
}
在AnonymousObserver内部有一个变量_eventHandler，保存了事件处理的block。
final class AnonymousObserver<ElementType> : ObserverBase<ElementType> {
typealias Element = 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)
}
}
我们再来看看AnonymousObserver的继承链：
处理完事件的时候还调用了这么一段代码self.asObservable().subscribe(observer)，self.asObservable().返回的是self，也就是我们的AnonymousObservable，而它调用的正是父类Producer的方法：
class Producer<Element> : Observable<Element> {
override init() {
super.init()
}
override func subscribe<O : ObserverType>(_ observer: O) -> Disposable where O.E == Element {
if !CurrentThreadScheduler.isScheduleRequired {
.......
} else {
return CurrentThreadScheduler.instance.schedule(()) { _ in
let disposer = SinkDisposer()
// self = AnonymousObservable
let sinkAndSubscription = self.run(observer, cancel: disposer)
disposer.setSinkAndSubscription(sink: sinkAndSubscription.sink, subscription: sinkAndSubscription.subscription)
return disposer
}
}
}
}
在订阅的时候，又调用了AnonymousObservable的run方法：
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)
}
}
最终会进入AnonymousObservableSink的run方法：
final private class AnonymousObservableSink<O: ObserverType>: Sink<O>, ObserverType {
// parent = AnonymousObservable
func run(_ parent: Parent) -> Disposable {
return parent._subscribeHandler(AnyObserver(self))
}
}
此时的parent就是AnonymousObservable对象，也就是执行AnonymousObservable._subscribeHandler，调用了我们之前保存的创建观察序列时候的闭包。这时候代码就会执行创建时闭包中的内容，即发送信号。调用闭包的时候，还传递过去了一个参数AnyObserver(self)。
public struct AnyObserver<Element> : ObserverType {
......
public typealias EventHandler = (Event<Element>) -> Void
private let observer: EventHandler
// 传入的observer是AnonymousObservableSink对象
public init<O : ObserverType>(_ observer: O) where O.E == Element {
// 保存的是on方法
self.observer = observer.on
}
.....
}
从代码可以看出，此时保存了AnonymousObservableSink这个对象的on方法。
发送事件
我们在创建观察序列的回调方法里面，观察者发送事件的时候会进入以下方法：
public func onNext(_ element: E) {
self.on(.next(element))
}
上述方法中的self就是AnyObserver，所以会调用下面的方法
public struct AnyObserver<Element> : ObserverType {
public typealias EventHandler = (Event<Element>) -> Void
private let observer: EventHandler
.....
public init<O : ObserverType>(_ observer: O) where O.E == Element {
self.observer = observer.on
}
public func on(_ event: Event<Element>) {
return self.observer(event)
}
}
我们在订阅信号的时候，把AnonymousObservableSink的on方法保存在了AnyObserver的内部变量observer中，此时就会调用到AnonymousObservableSink的on方法。
final private 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()
}
}
}
}
调用父类Sink的forwardOn方法：
class Sink<O : ObserverType> : Disposable {
final func forwardOn(_ event: Event<O.E>) {
if isFlagSet(self._disposed, 1) {
return
}
// self._obse = AnonymousObserver
// ObserverBase
self._observer.on(event)
}
}
调用ObserverBase的on方法：
class ObserverBase<ElementType> : Disposable, ObserverType {
func on(_ event: Event<E>) {
switch event {
case .next:
if load(self._isStopped) == 0 {
// self = AnonymousObserver
self.onCore(event)
}
case .error, .completed:
if fetchOr(self._isStopped, 1) == 0 {
self.onCore(event)
}
}
}
}
调用AnonymousObserver的onCore方法，同样，我们在订阅信号的时候，创建了一个匿名观察者AnonymousObserver的对象，同时，将我们的回调block保存在了_eventHandler中，此时调用_eventHandler，就是调用保存的是block。
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)
}
}
调用订阅信号方法里面的on事件处理，同时把发送的事件传递给外界。
至此，RxSwift基本上就结束了。大概就是我们创建的时候会把发送信号的逻辑写在创建的闭包里，并且将创建的闭包保存；订阅信号的时候，会把订阅的闭包保存，然后在订阅的方法里调用创建的闭包，此时就会发出信号，而在发送信号的过程中又会调用订阅的闭包，这就响应了发出的信号。
总结
总结不多说，直接上图：
