mojito: 麻烦给我的爱人来一份 RxSwift

序

学过 Swift 的同学都知道， RxSwift 宛如周董的 mojito

开始微醺

再者上头

为什么要学习 RxSwift ？

卡蜜尔说过

优雅，永不过时

麻烦给我的爱人来一份 RxSwift

RxSwift 是 Rx 系列的 Swift 版本，相较于 OC 版的 ReactiveCocoa

它们有着异曲同工之妙 函数响应式编程(FRP)

什么是函数响应式编程？

函数式:

函数式编程的核心思想是 stateless，无状态。函数本身并不关心外界输入的值

它只是在函数内部，将输入的值和输出的值完成一种映射，即 input => output

比如:

func changeNum(input: Int) -> Int {return input * 3
}// changeNum 并不会对input 产生改变，只是将运算之后的值 输出

无状态 意味着函数本身，不会改变外部的状态，也不会改变输入的值的状态

再比如

Q:将数组 [1,2,3,4] 里的元素都乘以2，返回新的数组

那么一般的做法可能是:

命令式编程

let array = [1,2,3,4]
var newArray: [Int] = []for item in array {var num = itemnum *= 2newArray.append(num)
}
// [2,4,6,8]

命令式编程倾向于怎么做，具体是怎么把每个数都 * 2 的，那么这里涉及到了 可变数组 newArray

如果某一个时刻，newArray 被某个地方改变了，都会达到 非预期的效果

那么函数式编程会怎么做呢？

let array = [1,2,3,4]let newArray = array.compactMap {return $0 * 2
}
// [2,4,6,8]

函数式编程 和 申明式编程的思想大体一致

它们都只关注 做什么，而不是上面的 怎么做？

函数式编程：倾向于做什么，省去其繁琐的过程，一种更为安全，直观，易懂的编程方式

响应式:

一种抽象的事件流的异步编程方法

比如：用户点击一个按钮，发送网络请求，并将结果展示在label 上

这里网络请求是异步的

想要展示在label 上，就要拿到网络请求的回调，进一步展示

形成事件流写法就是

button.rx.tap.subscribe(onNext: {              // 点击按钮HomeApi.getTitle().asObservable() // 发起网络请求.map { (title) -> String in       // 拿到回调进行 map("我是 \(title)")} .bind(to: (titleLabel?.rx.text)!) // 绑定给 label.disposed(by: rx.disposeBag)      // 管理生命周期
})

这么一个较为复杂的操作，且包含异步操作的流程

在 RxSwift 的调整之后，是不是更为 简单易懂 ？事件的分发以及维护，可以在一个地方完成

大大提高了代码的可读性，以及维护成本

整个事件流的过程如下：

我们不用去关心 序列中的每一个元素，是异步的还是同步的，线程是否安全

只有当我们点击按钮，发送信号 之后，代码块内的函数体才会执行，整个一系列的事件流才会产生

这使得我们更加面向业务逻辑

而不是每一步的具体操作

那么具体 RxSwift 是怎么做到的呢？

喝完 mojito 你就知道了

我喜欢阅读它时紧皱的眉头

对于初学者来说

RxSwift 的学习曲线确实很陡，它诠释了什么是面向协议编程

过程虽然晦涩

但道阻且长

真正的大师永远怀着一颗学徒的心

在RxSwift 的世界里，万物皆 rx，到处是序列(sequence)

听着像不像 iOS 的万物皆对象

是的没错，我们来看一下rx Reactive 的定义，首先引入眼帘的是一个叫 ReactiveCompatible 的协议

public protocol ReactiveCompatible {# 关联协议associatedtype ReactiveBase # rx 是  Reactive 类型，并将 ReactiveBase 传入static var rx: Reactive<ReactiveBase>.Type { get set }var rx: Reactive<ReactiveBase> { get set }
}

Reactive 中还对 ReactiveCompatible进行了协议的拓展，在这个扩展中，通过调用rx，返回的是

Reactive 类型 或者Reactive实例

extension ReactiveCompatible {# Reactive 类型public static var rx: Reactive<Self>.Type {get {  return Reactive<Self>.self }}# Reactive 实例public var rx: Reactive<Self> {get {  return Reactive(self) }}
}

在看一下 Reactive 的实现，是一个包含参数泛型 Base 的结构体

public struct Reactive<Base> {public let base: Base# 将 Reactive 的初始化调用者 设置为 basepublic init(_ base: Base) {self.base = base}
}

如上文中点击按钮的 tap，即 button.rx.tap, 类型就是 UIButton 类型，将 UIButton 的实例设置为 base

那么想实现 万物皆rx，只需要简单的一步

extension NSObject: ReactiveCompatible { }

这样就可以让所有继承于 NSObjce 的对象，都遵循 ReactiveCompatible 协议，即 万物皆rx

Observable

Observable 意味着，可被观察的，也就是可观察序列,什么是序列呢？

我理解的就是 具备发出事件能力的的一种信号

比如:

肚子饿了 -> 吃饭

肚子饿了 可以作为一个可观察序列，当我们大脑感知到肚子饿了，就可以执行去吃饭的操作

TextField 输入 -> 显示

TextField 输入操作可以作为一个序列，可以监听到输入的内容

接下来

就开始调试 mojito 了

看一个订阅过程:

# 创建
let observable = Observable<String>.create { (observe) -> Disposable in# 发送observe.onNext("mojito")return Disposables.create()
}
# 订阅
observable.subscribe(onNext: { text inprint(text)
}).disposed(by: rx.disposeBag)// print "mojito"
复制代码

调试开始

Observable 可观察序列

step1

# Observable 继承于  ObservableType
public class Observable<Element> : ObservableType {# 资源的引用计数 +1init() {_ = Resources.incrementTotal()}# 提供被订阅的能力，由子类实现public func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Element {rxAbstractMethod()}# 将 Observable 类转为 Observable 实例public func asObservable() -> Observable<Element> {return self}# 资源的引用计数 -1deinit {_ = Resources.decrementTotal()}
}

可是这里并没有看到序列的创建，但是可以看到一个继承关系: Observable<Element> : ObservableType

进入 ObservableType

step2

# 协议  ObservableType，继承于 ObservableConvertibleType
public protocol ObservableType: ObservableConvertibleType {func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Element
}# ObservableType 扩展
extension ObservableType {# 提供了一个方法，将遵守 ObservableType 协议的对象 转为 Observable 实体public func asObservable() -> Observable<Element> {return Observable.create { o inreturn self.subscribe(o)}}
}

这里还是没有看到 订阅的方法

还发现了自己的爸爸是个协议，还有爷爷 ObservableConvertibleType

持着怀疑的态度，你又点进了 ObservableConvertibleType

step3

# 也是个协议
public protocol ObservableConvertibleType {associatedtype Elementtypealias E = Element# 定义了一个方法，返回类型 Observable 可观察序列func asObservable() -> Observable<Element>
}

可恶

既然这条路走不通，只能先不走了

哪里跌倒

我就躺在哪里

为了达到万物皆序列，我们就要想办法把所有事件转化为序列，asObservable() 即为 RxSwift 的精髓

Observable.create()

点击 creat ，豁然开朗，原来创建是通过 ObservableType 扩展，这也同时证明了 OOP 的好处，可扩展性强

ObservableType 像是一家名叫 ObservableType 的连锁公司

它可以在任何地方开个分店

实现自己公司的业务

step4

#  ObservableType 的扩展
extension ObservableType {public static func create(_ subscribe: @escaping (AnyObserver<Element>) -> Disposable) -> Observable<Element> {# 返回一个 匿名观察序列，将 subscribe 逃逸闭包传入return AnonymousObservable(subscribe)}
}

点击 AnonymousObservable 进入

step5

# 私有方法，外界无法共享
# 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 提供的 run 方法override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Element {# 初始化匿名管道，传入一个订阅者let sink = AnonymousObservableSink(observer: observer, cancel: cancel)let subscription = sink.run(self)return (sink: sink, subscription: subscription)}
}

这里又来了个 Producer，点击 Producer

step6

# Producer 同样继承于 Observable
class Producer<Element> : Observable<Element> {override init() {super.init()}# 这里涉及到了线程，如果  CurrentThreadScheduler 指定了某个线程，那么就会在指定线程中 执行 run
# 否则 就会在当前线程中 执行 run
# SinkDisposer实例 disposer,用来管理资源释放
override func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == 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(()) { _ inlet disposer = SinkDisposer()let sinkAndSubscription = self.run(observer, cancel: disposer)disposer.setSinkAndSubscription(sink: sinkAndSubscription.sink, subscription: sinkAndSubscription.subscription)return disposer}}}
# 抽象方法，子类去实现，也就是匿名序列 AnonymousObservable
func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Element {rxAbstractMethod()}
}

相信到此，你我都已经微醺了

你问我什么是序列 ？

我指着大海的方向

目前也只能先画个图，继续观望

小结

小结
- 我们调用父类协议的 creat 方法，生成匿名观察序列，即Producer 的子类AnonymousObservable
- AnonymousObservable 保存外界传入的闭包
- 负责资源管理，引用计数的是 Observable 抽象类，不实现方法
- Producer 类实现外界 subscribe 方法，并安排线程调度
- 具体的 run，由 AnonymousObservable 实现，父类 Producer不负责

ok ，继续往下走

subscribe(onNext:) 订阅

点击 subscribe 进入，可以看到 ObservableType 的扩展,提供了 subscribe.on 和 subscribe.onNext 2个方法

此处省略了 subscribe.on

step7

extension ObservableType {...public func subscribe(onNext: ((Element) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil)-> Disposable {....# 创建一个 匿名订阅者  AnonymousObserver# 对外界传入的执行闭包 进行保存let observer = AnonymousObserver<Element> { event inswitch 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)}
}

这里将外界需要执行的闭包，即本例中的 print(text)，生成 AnonymousObserver 实例，传入

self.asObservable().subscribe(observer)

也就是说，这个 AnonymousObserver实例，会通过 Producer 调用 subscribe

然后由子类 AnonymousObservable，序列实例去调用 run方法

来到文中，step 5 的 run 方法，如下

# 将外界 需要执行的闭包 ，以及 资源销毁实例 生成的 元祖 传入  AnonymousObservableSink
# 生成 sink 管道实例，并执行 run
# 将run 之后生成的实例，赋值给  subscription，并返回 subscription 和  sinkoverride func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Element {let sink = AnonymousObservableSink(observer: observer, cancel: cancel)let subscription = sink.run(self)return (sink: sink, subscription: subscription)
}

进入 AnonymousObservableSink

step8

final private class AnonymousObservableSink<Observer: ObserverType>: Sink<Observer>, ObserverType {typealias Element = Observer.Element typealias Parent = AnonymousObservable<Element># 调用父类 Sink 的初始化方法，传入  observer 和 cancel，即 管道 AnonymousObservableSink 持有 这2个属性override init(observer: Observer, cancel: Cancelable) {super.init(observer: observer, cancel: cancel)}func on(_ event: Event<Element>) {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()}}}### 熟悉的东西有没有# 这里看到了  _subscribeHandler，也就是 发出的信号，保存的闭包func run(_ parent: Parent) -> Disposable {return parent._subscribeHandler(AnyObserver(self))}
}

到这里，我们就会发现 sink 管道它很重要

它持有了

要销毁的实例，发出序列的闭包，执行序列的闭包

这里的  AnyObserver(self) 是为了 兼容传入 闭包的类型， 本文对应的 是 String

也就是说，一旦外界开始订阅序列

那么 火车序列就开始发动了，但是怎么响应？下一步往哪开？

这就要看 AnyObserver(self)，做了什么，进入 AnyObserver，查看init

 public init<Observer: ObserverType>(_ observer: Observer) where Observer.Element == Element {self.observer = observer.on
}

你会发现这里的 self.observer 保存了自己的 on 方法

也就是保存的了一个 function

即会调用 step8 中的 on, 然后去调用父类Sink 的 forwardOn

step9

# 父类 Sink
class Sink<Observer: ObserverType> : Disposable {final func forwardOn(_ event: Event<Observer.Element>) {#if DEBUGself._synchronizationTracker.register(synchronizationErrorMessage: .default)defer { self._synchronizationTracker.unregister() }#endifif isFlagSet(self._disposed, 1) {return}# 订阅者self._observer.on(event)}
}

在父类 forwardOn中，由订阅者执行 on 事件

可是订阅者 AnonymousObserver 类有没有 on 方法，只有 onCore

所以去 AnonymousObserver 的父类中 ObserverBase 寻找

class ObserverBase<Element> : Disposable, ObserverType {private let _isStopped = AtomicInt(0)func on(_ event: Event<Element>) {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<Element>) {rxAbstractMethod()}
}

最后 AnonymousObserver 调用自己的 onCore 执行 eventHandler 闭包

到此

整个执行的过程算是走完了

关于资源回收的内容，后续文章会写到

到此

mojito 初体验结束

小结

小结
- 通过 AnonymousObservable 保存可观察序列
- 通过 AnonymousObserve 保存执行闭包
- 外界开始订阅，由 Producer 调度线程，执行 subscribe
- 生成 SinkDisposer 以及 observer 实例元祖
- 将元祖注入 Sink 管道
- Sink 处理事件，发出信号，响应序列
- 资源回收

简单的流程图如下

而我的写法，轻松像魔法

有了RxSwift ，日常开发就变得酣畅淋漓了，比如

监听 tableView 的滚动:

 tableView.rx.contentOffset.subscribe(onNext: { contentOffset in/// 修改透明度})
.disposed(by: rx.disposeBag)

监听textField 输入

 textField.rx.text.skip(1).subscribe(onNext: { (text) inprint("输入的是 : \(text!)")}).disposed(by: rx.disposeBag)

按钮点击

 self.messageBtn.rx.tap.subscribe(onNext: { inNavigator.push("")}).disposed(by: rx.disposeBag)

tableView 绑定数据源代理

  # 这里需要导入 RxDataSourcesdataSource = RxTableViewSectionedReloadDataSource(configureCell: { (_, tab, indexPath, item) -> UITableViewCell inlet cell = tab.dequeue(Reusable.settingCell, for: indexPath)cell.bind(to: item)return cell})# 或者let items = Observable.just(["Just","Relay","From","Driver","merge"])items.bind(to: tableView.rx.items) { (tableView,_,element) inlet cell = self.tableView.dequeue(TestV.normalCell)cell?.textLabel?.text = elementreturn cell!}.disposed(by: rx.disposeBag)

tableView 点击代理

 tableView.rx.itemSelected.subscribe(onNext: { indexPath in/// doSomething}).disposed(by: rx.disposeBag)

配合 HandyJSON 转model

extension Response {func mapHandyJsonModel<T: HandyJSON>(_ type: T.Type) -> T {let jsonString = String.init(data: data, encoding: .utf8)if let modelT = JSONDeserializer<T>.deserializeFrom(json: jsonString) {return modelT}return JSONDeserializer<T>.deserializeFrom(json: "{\"msg\":\"解析有误\"}")!}
}extension ObservableType where Element == Response {/// 注释public func mapHandyJsonModel<T: HandyJSON>(_ type: T.Type) -> Observable<T> {return flatMap { response -> Observable<T> inreturn Observable.just(response.mapHandyJsonModel(T.self))}}
}# 配合 Moya
static func getTopList() -> Observable<HomeResponseModel> {return HomeApiProvider.rx.request(.Top).asObservable().mapHandyJsonModel(HomeResponseModel.self)
}

多个请求合并

 Observable.zip(HomeApi.getTopList(), HomeApi.getRecommondList()).subscribe(onNext: { topResponse, recommodResponse in/// 数据处理}).disposed(by: self.rx.disposeBag)

等等....

先介绍一点简单的用法

后续会慢慢更新

这世界因我让你不再受折磨

RxSwift 熟悉了之后，会让我们的代码变得简洁且优雅

它面向协议编程，我们面向业务编程

RxSwift 需要慢慢品味

听一遍 mojito 肯定是不够的

不说了

听歌去了~

对了

没喝过 mojito 的就从这里开始吧~

如何获取？
转发这篇文章，关注我，私信回复“资料”即可获取高清大纲，以上 spring，MyBatis，Netty源码分析，高并发、高性能、分布式、微服务架构的原理，JVM性能优化、分布式架构
如何私信？
关注我后，在手机，点进头像进我的主页，主页上方右上角有个私信，点击私信，如何回复关键字“资料”即可