为什么要用Rx
传统的编程方式大多都是告诉程序需要做什么、怎么做、什么时候做，并通过KVO、Notification、Delegate监听变化，同时又需要告诉系统什么时候会发生变化。ReactiveX可以帮助我们让代码自动相应更新，程序可以对底层数据的变化做出响应。
RxSwift是ReactiveX的Swift版，RxCocoa使用RxSwift对Cocoa APIs响应式编程的封装。
比如让一个Button响应单击事件，非响应式编程是这样：
button.addTarget(self, action: #selector(buttonAction), for: .touchUpInside)
func buttonAction() {
...
}
用了RxCocoa后变成这样：
button.rx.tap.subscribe(onNext: {
...
})
代码变得更加简洁，不需要定义多余的函数，从而可以让开发者更专注于业务逻辑，不再维护中间状态。
Rx的一些重点
Observables和Observers
两个重要的概念：Observable和Observer
- Observable是发生变化的对象
- Observer是接收变化通知的对象
多个Observer可以监听同一个Observable，Observable发生变化时会通知所有订阅的Observer。
Observable也相当于事件管道，会向订阅者发送事件信息。事件分为三种：
- .Next(value) 有新的数据。
- .Completed 管道已经结束。
- .Error 有异常发生导致事件管道结束。
例如一个网络请求，收到服务端返回的数据后会发送.Next，因为请求都是一次性的，所以Next发送完后会马上发送Completed。如果请求超时了则会发送Error。
DisposeBag
DisposeBag是RxSwift提供的用来管理内存的工具。
当带有DisposeBag属性的对象调用deinit()时，bag将被清空，且每一个Observer会自动取消订阅它所观察的对象。理念类似于autoreleasepool。
如果没有DisposeBag，则会产生retain cycle，或者被意外释放掉，导致程序Crash。
let _disposeBag = DisposeBag()
Observable.just("123").subscribe(onNext: { (item) in
}).addDisposableTo(_disposeBag)
Disposables.create
每个Observable都要求Disposable类型的返回值，通过Disposables.create便能创建Disposable的实例，当Observable被释放时会执行Disposable，相当于析构函数。
Observable.create
实际开发中用的最多的函数，创建一个Observable。
Observable.create { (observer) -> Disposable in
...
if (error) {
observer.on(.error(RxSwiftError.unknown))
} else {
observer.on(.next(value))
observer.on(.completed)
}
return Disposables.create {
}
}
Observable.just
just只会发送一条数据，它先发送一次.Next(value)，然后发送.Completed。
Observable.empty
empty是空管道，它只会发送.Completed消息。
Observable.deferred
deferred会等到有Observer的时候再去创建Observable，相当于懒加载，而且每个订阅者订阅的对象都是内容相同但指针不同的Observable。
Subjects
Subjet是observable和Observer之间的桥梁，它既是Obserable又是Observer，既可以发出事件，也可以订阅事件。
PublishSubject只能收到订阅它之后的事件，订阅之前的事件无法收到。
Debug
打印所有订阅、事件、Disposable。
sequenceThatErrors
.retry(3)
.debug()
.subscribe(onNext: { print($0) })
.addDisposableTo(disposeBag)
实践（封装CoreBluetooth）
PTRxBluetooth主要分为三个类：
-
PTCBCentralManager
封装系统原生类CBCentralManager
fileprivate let didUpdateStateSubject = PublishSubject<PTBluetoothState>() extension PTCBCentralManager: CBCentralManagerDelegate { public func centralManagerDidUpdateState(_ central: CBCentralManager) { if let bleState = PTBluetoothState(rawValue: central.state.rawValue) { didUpdateStateSubject.onNext(bleState) } } }第一步：定义一个PublishSubject类型的属性
第二步：实现协议CBCentralManagerDelegate，在centralManagerDidUpdateState回调函数中，调用onNext将蓝牙状态发送给订阅者。
每个回调函数都有一个与之相对应的PublishSubject属性，在函数执行时调用onNext将数据发送给订阅者，上层只需要订阅Subject就能得到数据状态，不再需要代理或callback。
-
PTBluetoothClient
RxBluetooth的主要入口，封装了所有蓝牙操作。上层在检查蓝牙状态、搜寻设备、连接设备时主要与这个类交互。
//监听手机的蓝牙开启状态 public var state: Observable<PTBluetoothState> { return .deferred { return self.base.centralManager.rx.didUpdateState.startWith(self.base.centralManager.state) } }第一步：调用了.deferred，deferred会等到有Observer订阅它的时候去创建，并且给每个Observer都创建一个新对象。
第二步：调用centralMangaer的函数didUpdateState，并使用startWith赋初始值，通常是unknown。
关于startWith可以参考官方文档startWith
//搜寻周围的蓝牙设备 public func scanForPeripherals() -> Observable<PTCBPeripheral> { base.scanPeripheral.removeAll() return .deferred { let observable = Observable.create { (element: AnyObserver<PTCBPeripheral>) -> Disposable in let disposeable = self.base.centralManager.rx.didDiscoverPeripheral .map { return $0 } .subscribe(onNext: { (peripheral) in if self.base.scanPeripheral.contains(peripheral) == false { self.base.scanPeripheral.append(peripheral) element.onNext(peripheral) } }) self.base.centralManager.scanForPeripherals(withServices: nil, options: nil) return Disposables.create { disposeable.dispose() self.base.centralManager.stopScan() } } .subscribeOn(self.base.subscriptionQueue) .publish() .refCount() return self.base.ensureBluetoothState(.poweredOn, observable: observable) } }第一步：创建一个Observable。
第二步：订阅centralManager的didDiscoverPeripheral，当搜寻到新设备时，发送给observable的订阅者。
第三步：调用函数scanForPeripherals开始搜寻蓝牙设备。
第四步：observable释放时，取消订阅didDiscoverPeripheral，并停止搜寻设备。
第五步：调用publish和refcount，这样做是为了将冷信号(Cold Observable)转成热信号(Hot Observable)。
1.Hot Observable是主动的，既是没有订阅者，它依然会发送事件。Cold Observable是被动的，只有出现订阅者后，它才会发送事件。
2.Hot Observable可以有多个订阅者，Cold Observable只能一对一，当出现多个订阅者时，事件会从新发送。
//尝试连接蓝牙设备 public func connect(_ peripheral: PTCBPeripheral, options: [String: Any]? = nil) -> Observable<PTCBPeripheral> { let success = base.centralManager.rx.didConnectPeripheral .filter { $0 == peripheral } .take(1) .map { _ in return peripheral } let error = base.centralManager.rx.didFailConnectPeripheral .filter { $0.0 == peripheral } .take(1) .flatMap { (peripheral, error) -> Observable<PTCBPeripheral> in return .empty() } let observable = Observable<PTCBPeripheral>.create { (observer) -> Disposable in guard peripheral.isConnected == false else { observer.onNext(peripheral) observer.onCompleted() return Disposables.create() } let disposable = success.amb(error).subscribe(observer) self.base.centralManager.connect(peripheral, options: options) return Disposables.create { if peripheral.isConnected == false { self.base.centralManager.cancelPeripheralConnection(peripheral) disposable.dispose() } } } return observable }第一步：订阅连接成功和连接失败的Observable。
第二步：创建一个Observable。
第三步：调用amb将success和error绑定在一起，amb的主要作用就是将多个Observable绑定，但是只有第一个Observable能够发送onNext，其他的Observable只能发送onError和onCompleted。因为连接失败是没有onNext的，所以这里用到了amb。
-
PTCBPeripheral
PTCBPeripheral封装系统类CBPeripheral，主要用来与蓝牙设备交互，比如接收蓝牙数据、发送数据给设备。
extension PTCBPeripheral: CBPeripheralDelegate { public func peripheral(_ peripheral: CBPeripheral, didDiscoverServices error: Error?) { let services = peripheral.services ?? [] services.forEach { peripheralDidDiscoverServicesSubject.onNext($0) } } public func peripheral(_ peripheral: CBPeripheral, didDiscoverCharacteristicsFor service: CBService, error: Error?) { let characteristics = service.characteristics ?? [] peripheralDidDiscoverCharacteristicsSubject.onNext(characteristics) } public func peripheral(_ peripheral: CBPeripheral, didUpdateValueFor characteristic: CBCharacteristic, error: Error?) { peripheralDidUpdateValueSubject.onNext(characteristic) } }实现CBPeripheralDelegate代理，当搜寻到Services和Characteristics时，相对应的PublishSubject会发送数据给订阅者。
上层需要搜寻设备的Service时，只需要这样：
aPeripheral.rx.discoverServices(nil).subscribe(onNext: { (service) in }).addDisposableTo(_disposeBag)
