随着函数响应式编程成为现在很多编程语言的主流编程思想，我们越发的感叹这种思想的魅力，他极大的减少了我们的胶水代码，使得我们可以更好的编写业务逻辑代码。而作为我们iOS程序员来说，RAC以及RxSwift框架使用的也越来越多，我们不再需要写大量的代理方法来作为方法回调，我们可以将一个功能模块代码写在一起，通过链式来表达，使代码逻辑给人一种清晰可见的感觉。
RxSwift
作为ReactiveX家族之一的RxSwift在Github截止现在Star:16K.为什么这个框架如此受欢迎，作为函数响应式框架典型代表，底层实现又是如何实现的呢？这一篇文章全面解密
RxSwift核心流程
Rxswift的优秀之处在于他为开发人员提供了非常精简的API，让我们可以快速上手来使用，它的使用公有三个步骤：
- 1.创建序列
- 2.订阅序列
- 3.发送序列
//1.创建序列
let ob = Observable<Any>.create { (obserber) -> Disposable in
// 3:发送序列
obserber.onNext("AAA")
obserber.onCompleted()
return Disposables.create()
}
//2.订阅序列
let _ = ob.subscribe(onNext: { (text) in
print("订阅到:\(text)")
}, onError: { (error) in
print("error: \(error)")
}, onCompleted: {
print("完成")
}) {
print("销毁")
}
image.png
这里我们肯定很疑惑为什么我们的AAA字符串会被传入subscribe的闭包当中，从这里我们开始一步步的分析它里边到底是如何实现的。
创建序列
extension ObservableType {
// MARK: create
public static func create(_ subscribe: @escaping (AnyObserver<E>) -> Disposable) -> Observable<E> {
//AnonymousObservable.init(subscribe)
return AnonymousObservable(subscribe)
}
}
我们看到 可观察序列的创建是通过ObservableType的协议拓展功能的create方法实现的，并且返回了一个匿名内部类AnonymousObservable,而AnonymousObservable中保存了外界传入的subscribe闭包，下面是一附类的继承关系图
image.png
如上图所示
AnonymousObservable继承了Producer可观察序列生产者
final private class AnonymousObservable<Element>: Producer<Element> {
typealias SubscribeHandler = (AnyObserver<Element>) -> Disposable
let _subscribeHandler: SubscribeHandler
//保存外界传入的观察者对象
init(_ subscribeHandler: @escaping SubscribeHandler) {
self._subscribeHandler = subscribeHandler
}
}
-
Producer类继承自Observable可观察序列,并保存了_subscribeHandler -
SubscribeHandler为外部传入的观察者对象AnyObserver<Element>
public class Observable<Element> : ObservableType {
/// Type of elements in sequence.
public typealias E = Element
init() {
#if TRACE_RESOURCES
_ = Resources.incrementTotal()
#endif
}
public func subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == E {
rxAbstractMethod()
}
public func asObservable() -> Observable<E> {
return self
}
-
Observable又遵循了ObservableConvertibleType和ObservableType协议
ObservableType
所以AnonymousObservable匿名内部类拥有Observable非常重要的方法subscrible
订阅序列
我们先看下subscribe内部实现
image.png
public func subscribe(onNext: ((E) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil)
-> Disposable {
//这里省略部分代码，我们只分析observer(观察者的创建)
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
)
}
代码说明：
- 我们创建了一个
AnonymousObserver匿名内部观察者，通过初始化闭包，保存了外部传入的onNext,onError,onCompleted,onDisposed事件。 我们先看一下观察者的继承链
image.png
-
self.asObserverable()这个函数来保持一致性的写法 -
self.asObservable().subscribe(observer)通过可观察序列的继承关系，我们可以快速的定位producer订阅代码
image.png
image.png
image.png
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代码最终由我们的生产者producer延伸到我们的具体事务代码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)
}
}
-
sink.run（self）将业务处理下沉，让分工更具明确
func run(_ parent: Parent) -> Disposable {
return parent._subscribeHandler(AnyObserver(self))
}
-
parent为我们传入的AnonymousObservable对象 -
这里我们就可以清晰的看到
AnonymousObservable._subscribeHandler的执行，我们的疑惑终于在此时解开了，这就是为什么我们的subscrible订阅序列会执行我们创建序列里的observer观察者序列闭包,是的是的绕了一大圈终于回来了。。。
let ob = Observable<Any>.create { (observer) -> Disposable in
// 3:发送信号
observer.onNext("AAA")
observer.onCompleted()
// obserber.onError(NSError.init(domain: "coocieeror", code: 10087, userInfo: nil))
return Disposables.create()
}
-
AnyObserver(self)的初始化方法里保存了一个非常重要的函数AnonymousObservableSink .on 函数
public init<O : ObserverType>(_ observer: O) where O.E == Element {
self.observer = observer.on
}
发送事件响应
从上述流程我们清晰的得知，observer.onNext("AAA")的本质是调用了AnyObserver.onNext("AAA"),我们来看下源码
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)
}
}
-
self.observer构造初始化就是：AnonymouservableSink.on函数 - 这样我们就可以将
self.observer(event)变形：->AnonymousObservableSink.on(event)->AnonymousObservableSink.on(.next("AAA"))
*AnonymousObservableSink我们这里把它就先称为管子，来说说管子它都干了些啥，看源码：
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()
}
}
}
}
-
self.forwardOn(event)我们先看下这句代码的父类做了什么事
class Sink<O : ObserverType> : Disposable {
final func forwardOn(_ event: Event<O.E>) {
if isFlagSet(self._disposed, 1) {
return
}
self._observer.on(event)
}
}
- 其中
self._observer就是我们初始化保存的AnonymousObserver匿名观察者，怎么样？发现了吗？激动吗？我们又回到了订阅序列创建时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()
}
}
- 通过调用
onNext?(value),此时value的值就是我们外界传入的AAA啊。。
绕了这么一大圈，我们通过对可观察序列的创建，订阅序列，发送事件最外层的调用，再到他们内部各种协议的配合，下沉业务逻辑，接口隔离让我们着实体会了RxSwift是多么优秀的框架，相信通过这篇对RxSwift的核心逻辑的探究，会让各位小伙伴对RxSwift的设计思路有一个清晰的认知。最后贴给小伙伴们整个RxSwift的分析图
image.png
网友评论