RxSwift学习

“它拓展了观察者模式。使你能够自由组合多个异步事件，而不需要去关心线程，同步，线程安全，并发数据以及I/O阻塞。”

1.2主要具有哪些功能

1.2.1 实现点击按钮方法Target Action

button.rx.tap
    .subscribe(onNext: {
        print("button Tapped")
    })
    .disposed(by: disposeBag)
    
button.rx.tap
    .bind { [weak self] _ -> Void in
        self?.openAppPreferences()
    }
    .disposed(by: disposeBag)
复制代码

1.2.2实现代理

scrollView.rx.contentOffset
    .subscribe(onNext: { contentOffset in
        print("contentOffset: \(contentOffset)")
    })
    .disposed(by: disposeBag)
复制代码

1.2.3.闭包回调

URLSession.shared.rx.data(request: URLRequest(url: url))
    .subscribe(onNext: { data in
        print("Data Task Success with count: \(data.count)")
    }, onError: { error in
        print("Data Task Error: \(error)")
    })
    .disposed(by: disposeBag)
复制代码

1.2.4.通知

NotificationCenter.default.rx
    .notification(.UIApplicationWillEnterForeground)
    .subscribe(onNext: { (notification) in
        print("Application Will Enter Foreground")
    })
    .disposed(by: disposeBag)
不需要去管理观察者的生命周期，这样你就有更多精力去关注业务逻辑
复制代码

1.2.5.KVO

user.rx.observe(String.self, #keyPath(User.name))
    .subscribe(onNext: { newValue in
        print("do something with newValue")
    })
    .disposed(by: disposeBag)
这样实现 KVO 的代码更清晰，更简洁并且更准确。
复制代码

1.2.6.多个任务之间有依赖关系

/// 用 Rx 封装接口
enum Api {

    /// 通过用户名密码取得一个 token
    static func token(username: String, password: String) -> Observable<String> { ... }

    /// 通过 token 取得用户信息
    static func userInfo(token: String) -> Observable<UserInfo> { ... }
}
/// 通过用户名和密码获取用户信息
Api.token(username: "beeth0ven", password: "987654321")
    .flatMapLatest(Api.userInfo)
    .subscribe(onNext: { userInfo in
        print("获取用户信息成功: \(userInfo)")
    }, onError: { error in
        print("获取用户信息失败: \(error)")
    })
    .disposed(by: disposeBag)
你无需嵌套太多层，从而使得代码易读，易维护
复制代码

1.2.7.等待多个并发任务完成后处理结果

// 需要将两个网络请求合并成一个，
/// 用 Rx 封装接口
enum Api {

    /// 取得老师的详细信息
    static func teacher(teacherId: Int) -> Observable<Teacher> { ... }

    /// 取得老师的评论
    static func teacherComments(teacherId: Int) -> Observable<[Comment]> { ... }
}

/// 同时取得老师信息和老师评论
Observable.zip(
      Api.teacher(teacherId: teacherId),
      Api.teacherComments(teacherId: teacherId)
    ).subscribe(onNext: { (teacher, comments) in
        print("获取老师信息成功: \(teacher)")
        print("获取老师评论成功: \(comments.count) 条")
    }, onError: { error in
        print("获取老师信息或评论失败: \(error)")
    })
    .disposed(by: disposeBag)
复制代码

1.3 使用RxSwift的优势

理由：那么为什么要使用 RxSwift

• 复合 - Rx 就是复合的代名词
• 复用 - 因为它易复合
• 清晰 - 因为声明都是不可变更的
• 易用 - 因为它抽象的了异步编程，使我们统一了代码风格
• 稳定 - 因为 Rx 是完全通过单元测试的

1.4 RxSwift原理

函数式编程是种编程范式，它需要我们将函数作为参数传递，或者作为返回值返还。

数据绑定（订阅）关系：

数据绑定
可被监听的序列==========>观察者
复制代码

核心内容

• Observable - 产生事件，所有的事物都是序列，用于描述元素异步产生的序列。
• Observer - 响应事件
• Operator - 创建变化组合事件
• Disposable - 管理绑定（订阅）的生命周期
• Schedulers - 线程队列调配

1.4.1 产生事件

// Observable<String>
let text = usernameOutlet.rx.text.orEmpty.asObservable()

// Observable<Bool>
let passwordValid = text
    // Operator
    .map { $0.characters.count >= minimalUsernameLength }
复制代码

1.4.2 Observer —响应事件,观察者，具体实现

// Observer<Bool>
let observer = passwordValidOutlet.rx.isHidden
复制代码

1.4.3 取消绑定

// Disposable
let disposable = passwordValid
    // Scheduler 用于控制任务在那个线程队列运行
    .subscribeOn(MainScheduler.instance)
    .observeOn(MainScheduler.instance)
    .bind(to: observer)
// 取消绑定，你可以在退出页面时取消绑定
disposable.dispose()
复制代码

2.如何创建事件(序列)

2.1最简单的创建

let numbers: Observable<Int> = Observable.create({ observer -> Disposable in
        observer.onNext(1)    // 产生了一个元素，他的值是 1
        observer.onNext(2)
        observer.onNext(3)
	observer.onCompleted() // 表示元素已经全部产生，没有更多元素了
        return Disposables.create()
    })
复制代码

2.2 决策树

• 产生特定的一个元素：just
• 经过一段延时：timer
• 从一个序列拉取元素：from
• 重复的产生某一个元素：repeatElement
• 存在自定义逻辑：create
• 每次订阅时产生：deferred
• 每隔一段时间，发出一个元素：interval
• 在一段延时后：timer
• 一个空序列，只有一个完成事件：empty
• 一个任何事件都没有产生的序列：never 
我想要创建一个 Observable 通过组合其他的 Observables
• 任意一个 Observable 产生了元素，就发出这个元素：merge
	•	让这些 Observables 一个接一个的发出元素，当上一个 Observable 元素发送完毕后，下一个Observable 才能开始发出元素：concat
	•	组合多个 Observables 的元素
	•	当每一个 Observable 都发出一个新的元素：zip
	•	当任意一个 Observable 发出一个新的元素：combineLatest
我想要转换 Observable 的元素后，再将它们发出来
	•	对每个元素直接转换：map
	•	转换到另一个 Observable：flatMap
	•	只接收最新的元素转换的 Observable 所产生的元素：flatMapLatest
	•	每一个元素转换的 Observable 按顺序产生元素：concatMap
	•	基于所有遍历过的元素： scan
我想要将产生的每一个元素，拖延一段时间后再发出：delay
我想要将产生的事件封装成元素发送出来
	•	将他们封装成 Event<Element>：materialize
	•	然后解封出来：dematerialize
我想要忽略掉所有的 next 事件，只接收 completed 和 error 事件：ignoreElements
我想创建一个新的 Observable 在原有的序列前面加入一些元素：startWith
我想从 Observable 中收集元素，缓存这些元素之后在发出：buffer
我想将 Observable 拆分成多个 Observables：window
	•	基于元素的共同特征：groupBy
我想只接收 Observable 中特定的元素
	•	发出唯一的元素：single
我想重新从 Observable 中发出某些元素
	•	通过判定条件过滤出一些元素：filter
	•	仅仅发出头几个元素：take
	•	仅仅发出尾部的几个元素：takeLast
	•	仅仅发出第 n 个元素：elementAt
	•	跳过头几个元素
	•	跳过头 n 个元素：skip
	•	跳过头几个满足判定的元素：skipWhile，skipWhileWithIndex
	•	跳过某段时间内产生的头几个元素：skip
	•	跳过头几个元素直到另一个 Observable 发出一个元素：skipUntil
	•	只取头几个元素
	•	只取头几个满足判定的元素：takeWhile，takeWhileWithIndex
	•	只取某段时间内产生的头几个元素：take
	•	只取头几个元素直到另一个 Observable 发出一个元素：takeUntil
	•	周期性的对 Observable 抽样：sample
	•	发出那些元素，这些元素产生后的特定的时间内，没有新的元素产生：debounce
	•	直到元素的值发生变化，才发出新的元素：distinctUntilChanged
	•	并提供元素是否相等的判定函数：distinctUntilChanged
	•	在开始发出元素时，延时后进行订阅：delaySubscription
我想要从一些 Observables 中，只取第一个产生元素的 Observable：amb
我想评估 Observable 的全部元素
	•	并且对每个元素应用聚合方法，待所有元素都应用聚合方法后，发出结果：reduce
	•	并且对每个元素应用聚合方法，每次应用聚合方法后，发出结果：scan
我想把 Observable 转换为其他的数据结构：as...
我想在某个 Scheduler 应用操作符：subscribeOn
	•	在某个 Scheduler 监听：observeOn
我想要 Observable 发生某个事件时, 采取某个行动：do
我想要 Observable 发出一个 error 事件：error
	•	如果规定时间内没有产生元素：timeout
我想要 Observable 发生错误时，优雅的恢复
	•	如果规定时间内没有产生元素，就切换到备选 Observable ：timeout
	•	如果产生错误，将错误替换成某个元素 ：catchErrorJustReturn
	•	如果产生错误，就切换到备选 Observable ：catchError
	•	如果产生错误，就重试 ：retry
我创建一个 Disposable 资源，使它与 Observable 具有相同的寿命：using
我创建一个 Observable，直到我通知它可以产生元素后，才能产生元素：publish
	•	并且，就算是在产生元素后订阅，也要发出全部元素：replay
	•	并且，一旦所有观察者取消观察，他就被释放掉：refCount
	•	通知它可以产生元素了：connect
复制代码

2.3 事件 onNext, onError, onCompleted 事件。我们称这些事件为 Event:

public enum Event<Element> {
    case next(Element)        // 序列产生了一个新的元素
    case error(Swift.Error)  // 创建序列时产生了一个错误，导致序列终止
    case completed	          // 序列的所有元素都已经成功产生，整个序列已经完成		 
}
复制代码

2.4.特征序列

• Single // 它要么只能发出一个元素，要么产生一个 error 事件。例子： HTTP 请求，然后返回一个应答或错误
• Completable. // 只能产生一个 completed 事件，要么产生一个 error 事件; 不会共享状态变化
• Maybe. // 发出一个元素或者一个 completed 事件或者一个 error 事件;不会共享状态变化
• Driver // 不会产生 error 事件 ;一定在 MainScheduler 监听（主线程监听）; 共享状态变化； asDriver(onErrorJustReturn: []) 转换drive 而不是 bindTo，drive 方法只能被 Driver 调用
• ControlEvent // 专门用于描述 UI 控件所产生的事件;不会产生 error 事件;一定在 MainScheduler 订阅;一定在 MainScheduler 监听; 共享状态变化

2.4.1 Sigle 事件枚举

public enum SingleEvent<Element> {
    case success(Element)  // 产生一个单独的元素
    case error(Swift.Error)   // 产生一个错误
}
复制代码

创建办法

Single<[String: Any]>.create

对 Observable 调用 .asSingle() 方法，将它转换为 Single

2.4.2 Completable

只关心任务是否完成，而不需要在意任务返回值的情况。它和 Observable 有点相似

2.5 Observer - 观察者. 用来监听事件，然后它需要这个事件做出响应， 响应事件的都是观察者

如：弹出提示框就是观察者，它对点击按钮这个事件做出响应。

2.5.1创建观察者

创建观察者最直接的方法就是在 Observable 的 subscribe 方法后面描述，事件发生时，需要如何做出响应。而观察者就是由后面的 onNext，onError，onCompleted的这些闭包构建出来的。

tap.subscribe(onNext: { [weak self] in
    self?.showAlert()
}, onError: { error in
    print("发生错误： \(error.localizedDescription)")
}, onCompleted: {
    print("任务完成")
})
复制代码

2.5.1特殊观察者

AnyObserver 可以用来描叙任意一种观察者

Binder 有2个特征:不会处理错误事件;确保绑定都是在给定 Scheduler 上执行（默认 MainScheduler）

打印网络请求结果：
URLSession.shared.rx.data(request: URLRequest(url: url))
    .subscribe(onNext: { data in
        print("Data Task Success with count: \(data.count)")
    }, onError: { error in
        print("Data Task Error: \(error)")
    })
    .disposed(by: disposeBag)
    
可以看作是：
let observer: AnyObserver<Data> = AnyObserver { (event) in
    switch event {
    case .next(let data):
        print("Data Task Success with count: \(data.count)")
    case .error(let error):
        print("Data Task Error: \(error)")
    default:
        break
    }
}
复制代码

由于这个观察者是一个 UI 观察者，所以它在响应事件时，只会处理 next 事件，并且更新 UI 的操作需要在主线程上执行。
let observer: Binder<Bool> = Binder(usernameValidOutlet) { (view, isHidden) in
    view.isHidden = isHidden
}

usernameValid
    .bind(to: observer)
    .disposed(by: disposeBag)
复制代码

Binder 可以只处理 next 事件，并且保证响应 next 事件的代码一定会在给定 Scheduler 上执行，这里采用默认的 MainScheduler。

复用页面是否隐藏是一个常用的观察者，所以应该让所有的 UIView 都提供这种观察者：

extension Reactive where Base: UIView {
  public var isHidden: Binder<Bool> {
      return Binder(self.base) { view, hidden in
          view.isHidden = hidden
      }
  }
}

usernameValid
    .bind(to: usernameValidOutlet.rx.isHidden)
    .disposed(by: disposeBag)
    
label 的当前文本 label.rx.text：

extension Reactive where Base: UILabel {
  public var text: Binder<String?> {
      return Binder(self.base) { label, text in
          label.text = text
      }
  }
}
复制代码

2.6 Observable & Observer 既是可被监听的序列也是观察者

// 作为可被监听的序列
let observable = textField.rx.text
observable.subscribe(onNext: { text in show(text: text) })
// 作为观察者
let observer = textField.rx.text
let text: Observable<String?> = ...
text.bind(to: observer)

复制代码

同样具有此特性的UI控件有：switch的开关状态，segmentedControl的选中索引号，datePicker的选中日期, UISlider, UIStepper等等 controlPropertyWithDefaultEvents

其他辅助类也具有可被监听的序列也是观察者的对象

> AsyncSubject // Observable 产生完成事件后，发出最后一个元素（仅仅只有最后一个元素）
> PublishSubject
> ReplaySubject
> BehaviorSubject
> Variable
> ControlProperty
复制代码

2.6.1 AsyncSubject

在源 Observable 产生完成事件后，发出最后一个元素（仅仅只有最后一个元素），如果源 Observable 没有发出任何元素，只有一个完成事件。那 AsyncSubject 也只有一个完成事件；如果源 Observable 因为产生了一个 error 事件而中止，

AsyncSubject

就不会发出任何元素，而是将这个 error 事件发送出来。

let disposeBag = DisposeBag()
let subject = AsyncSubject<String>()

subject
  .subscribe { print("Subscription: 1 Event:", $0) }
  .disposed(by: disposeBag)

subject.onNext(":dog:")
subject.onNext(":cat:")
subject.onNext(":hamster:")
subject.onCompleted()
输出结果：

Subscription: 1 Event: next(:hamster:)
Subscription: 1 Event: completed
复制代码

2.6.2 ReplaySubject

PublishSubject将对观察者发送订阅后产生的元素，而在 订阅前 发出的元素将不会发送给观察者。如果你希望观察者接收到所有的元素，你可以通过使用 Observable 的 create 方法来创建 Observable，或者使用 ReplaySubject 。

如果源 Observable 因为产生了一个 error 事件而中止， PublishSubject 就不会发出任何元素，而是将这个 error

事件发送出来。

let disposeBag = DisposeBag()
let subject = PublishSubject<String>()

subject
  .subscribe { print("Subscription: 1 Event:", $0) }
  .disposed(by: disposeBag)

subject.onNext(":dog:")
subject.onNext(":cat:")

subject
  .subscribe { print("Subscription: 2 Event:", $0) }
  .disposed(by: disposeBag)

subject.onNext(":a:")
subject.onNext(":b:")
输出结果：

Subscription: 1 Event: next(:dog:)
Subscription: 1 Event: next(:cat:)
Subscription: 1 Event: next(:a:)
Subscription: 2 Event: next(:a:)
Subscription: 1 Event: next(:b:)
Subscription: 2 Event: next(:b:)
复制代码

2.6.3 ReplaySubject

ReplaySubject 将对观察者发送全部的元素，无论观察者是何时进行订阅的。

如果把 ReplaySubject 当作观察者来使用，注意不要在多个线程调用 onNext , onError 或 onCompleted

。这样会导致无序调用，将造成意想不到的结果。

let disposeBag = DisposeBag()
let subject = ReplaySubject<String>.create(bufferSize: 1)

subject
  .subscribe { print("Subscription: 1 Event:", $0) }
  .disposed(by: disposeBag)

subject.onNext(":dog:")
subject.onNext(":cat:")

subject
  .subscribe { print("Subscription: 2 Event:", $0) }
  .disposed(by: disposeBag)

subject.onNext(":a:")
subject.onNext(":b:")
输出结果：

Subscription: 1 Event: next(:dog:)
Subscription: 1 Event: next(:cat:)
Subscription: 2 Event: next(:cat:)
Subscription: 1 Event: next(:a:)
Subscription: 2 Event: next(:a:)
Subscription: 1 Event: next(:b:)
Subscription: 2 Event: next(:b:)
复制代码

2.6.4 BehaviorSubject

当观察者对 BehaviorSubject 进行订阅时，它会将源 Observable 中最新的元素发送出来（如果不存在最新的元素，就发出默认元素）。然后将随后产生的元素发送出来。

如果源 Observable 因为产生了一个 error 事件而中止， BehaviorSubject 就不会发出任何元素，而是将这个 error

事件发送出来。

let disposeBag = DisposeBag()
let subject = BehaviorSubject(value: ":red_circle:")

subject
  .subscribe { print("Subscription: 1 Event:", $0) }
  .disposed(by: disposeBag)

subject.onNext(":dog:")
subject.onNext(":cat:")

subject
  .subscribe { print("Subscription: 2 Event:", $0) }
  .disposed(by: disposeBag)

subject.onNext(":a:")
subject.onNext(":b:")

subject
  .subscribe { print("Subscription: 3 Event:", $0) }
  .disposed(by: disposeBag)

subject.onNext(":pear:")
subject.onNext(":tangerine:")
输出结果：

Subscription: 1 Event: next(:red_circle:)
Subscription: 1 Event: next(:dog:)
Subscription: 1 Event: next(:cat:)
Subscription: 2 Event: next(:cat:)
Subscription: 1 Event: next(:a:)
Subscription: 2 Event: next(:a:)
Subscription: 1 Event: next(:b:)
Subscription: 2 Event: next(:b:)
Subscription: 3 Event: next(:b:)
Subscription: 1 Event: next(:pear:)
Subscription: 2 Event: next(:pear:)
Subscription: 3 Event: next(:pear:)
Subscription: 1 Event: next(:tangerine:)
Subscription: 2 Event: next(:tangerine:)
Subscription: 3 Event: next(:tangerine:)
复制代码

2.6.5 Variable

RxSwift 提供的 Variable,可变量的版本

使用 var：

// 在 ViewController 中
var model: Model? = nil {
    didSet { updateUI(with: model) }
}

override func viewDidLoad() {
    super.viewDidLoad()

    model = getModel()
}

func updateUI(with model: Model?) { ... }
func getModel() -> Model { ... }

复制代码

使用 Variable：

// 在 ViewController 中
let model: Variable<Model?> = Variable(nil)

override func viewDidLoad() {
    super.viewDidLoad()

    model.asObservable()
        .subscribe(onNext: { [weak self] model in
            self?.updateUI(with: model)
        })
        .disposed(by: disposeBag)

    model.value = getModel()
}

func updateUI(with model: Model?) { ... }
func getModel() -> Model { ... }
复制代码

第一种使用 var 的方式十分常见，在 ViewController 中监听 Model 的变化，然后刷新页面。

第二种使用 Variable 则是 RxSwift 独有的。Variable 几乎提供了 var 的所有功能。另外，加上一条非常重要的特性，就是可以通过调用 asObservable() 方法转换成序列。然后你可以对这个序列应用操作符，来合成其他的序列。所以，如果我们声明的变量需要提供 Rx 支持，那就选用 Variable 这个类型

** 说明 ** Variable 封装了一个 BehaviorSubject，所以它会持有当前值，并且 Variable 会对新的观察者发送当前值。它不会产生 error 事件。Variable 在 deinit 时，会发出一个 completed 事件

2.6.6 ControlProperty

ControlProperty专门用于描述 UI 控件属性的，它具有以下特征：

• 不会产生 error 事件
• 一定在 MainScheduler 订阅（主线程订阅）
• 一定在 MainScheduler 监听（主线程监听）
• 共享状态变化

2.7操作符

操作符

的目的：创建新的序列，或者变化组合原有的序列，从而生成一个新的序列. 操作符有以下：

2.7.1 filter - 过滤，这个序列只发出温度大于 33 度的元素

// 温度
let rxTemperature: Observable<Double> = ...

// filter 操作符
rxTemperature.filter { temperature in temperature > 33 }
    .subscribe(onNext: { temperature in
        print("高温：\(temperature)度")
    })
    .disposed(by: disposeBag)
复制代码

2.7.2 map - 转换，创建一个新的序列。这个序列将原有的 JSON 转换成 Model 。这种转换实际上就是解析 JSON 。

// JSON
let json: Observable<JSON> = ...

// map 操作符
json.map(Model.init)
    .subscribe(onNext: { model in
        print("取得 Model: \(model)")
    })
    .disposed(by: disposeBag)
复制代码

2.7.3 zip - 配对， 这个序列将汉堡序列的元素和薯条序列的元素配对后，生成一个新的套餐序列。

// 汉堡
let rxHamburg: Observable<Hamburg> = ...
// 薯条
let rxFrenchFries: Observable<FrenchFries> = ...

// zip 操作符
Observable.zip(rxHamburg, rxFrenchFries)
    .subscribe(onNext: { (hamburg, frenchFries) in
        print("取得汉堡: \(hamburg) 和薯条：\(frenchFries)")
    })
    .disposed(by: disposeBag)
复制代码

其他操作符列表

amb
buffer
catchError
combineLatest
concat
concatMap
connect
create
debounce
debug
deferred
delay
delaySubscription
dematerialize
distinctUntilChanged
do
elementAt
empty
error
filter
flatMap
flatMapLatest
from
groupBy
ignoreElements
interval
just
map
merge
materialize
never
observeOn
publish
reduce
refCount
repeatElement
replay
retry
sample
scan
shareReplay
single
skip
skipUntil
skipWhile
startWith
subscribeOn
take
takeLast
takeUntil
takeWhile
timeout
timer
using
window
withLatestFrom
zip
复制代码

2.8 Disposable - 可被清除的资源

1.主动取消订阅

var disposeBag = DisposeBag()
textField.rx.text.orEmpty
        .subscribe(onNext: { text in print(text) })
        .disposed(by: self.disposeBag)
复制代码

disposeBag 和 ViewController 具有相同的生命周期。当退出页面时， ViewController 就被释放，disposeBag 也跟着被释放了，那么这里的 5 次绑定（订阅）也就被取消了。这正是我们所需要的。

2.自动取消订阅

_ = usernameValid
        .takeUntil(self.rx.deallocated)
        .bind(to: passwordOutlet.rx.isEnabled)
复制代码

这将使得订阅一直持续到控制器的 dealloc 事件产生为止。

2.9 Schedulers - 调度器

Schedulers 是 Rx 实现多线程的核心模块，它主要用于控制任务在哪个线程或队列运行。

// 后台取得数据，主线程处理结果
DispatchQueue.global(qos: .userInitiated).async {
    let data = try? Data(contentsOf: url)
    DispatchQueue.main.async {
        self.data = data
    }
}
复制代码

如果用 RxSwift 来实现，大致是这样的：

let rxData: Observable<Data> = ...

rxData
    .subscribeOn(ConcurrentDispatchQueueScheduler(qos: .userInitiated))
    .observeOn(MainScheduler.instance)
    .subscribe(onNext: { [weak self] data in
        self?.data = data
    })
    .disposed(by: disposeBag)
    
复制代码

使用subscribeOn来决定数据序列的构建函数在哪个 Scheduler 上运行以上例子中，由于获取 Data 需要花很长的时间，所以用 subscribeOn 切换到 后台 Scheduler 来获取 Data。这样可以避免主线程被阻塞。

使用 observeOn来决定在哪个 Scheduler 监听这个数据序列通过使用 observeOn 方法切换到主线程来监听并且处理结果。

一个比较典型的例子就是，在后台发起网络请求，然后解析数据，最后在主线程刷新页面。你就可以先用 subscribeOn 切到后台去发送请求并解析数据，最后用 observeOn 切换到主线程更新页面。

• MainScheduler MainScheduler 代表主线程。如果你需要执行一些和 UI 相关的任务，就需要切换到该 Scheduler 运行。

• SerialDispatchQueueScheduler SerialDispatchQueueScheduler 抽象了串行 DispatchQueue 。如果你需要执行一些串行任务，可以切换到这个 Scheduler 运行。

• ConcurrentDispatchQueueScheduler ConcurrentDispatchQueueScheduler 抽象了并行 DispatchQueue 。如果你需要执行一些并发任务，可以切换到这个 Scheduler 运行。

• OperationQueueScheduler OperationQueueScheduler 抽象了 NSOperationQueue 。 它具备 NSOperationQueue 的一些特点，例如，你可以通过设置 maxConcurrentOperationCount ，来控制同时执行并发任务的最大数量。

2.10 Error Handling - 错误处理

一旦序列里面产出了一个 error 事件，整个序列将被终止。 RxSwift 主要有两种错误处理机制：

• retry - 重试
• catch - 恢复

retry 可以让序列在发生错误后重试：

// 请求 JSON 失败时，立即重试，
// 重试 3 次后仍然失败，就将错误抛出

let rxJson: Observable<JSON> = ...

rxJson
    .retry(3)
    .subscribe(onNext: { json in
        print("取得 JSON 成功: \(json)")
    }, onError: { error in
        print("取得 JSON 失败: \(error)")
    })
    .disposed(by: disposeBag)
复制代码

以上的代码非常直接 retry(3) 就是当发生错误时，就进行重试操作，并且最多重试 3 次。

retryWhen如果我们需要在发生错误时，经过一段延时后重试，那可以这样实现： 这个操作符主要描述应该在何时重试，并且通过闭包里面返回的 Observable 来控制重试的时机,当它发出一个 error 或者 completed 事件时，就不会重试，并且将这个事件传递给到后面的观察者.

// 请求 JSON 失败时，等待 5 秒后重试，

let retryDelay: Double = 5  // 重试延时 5 秒

rxJson
    .retryWhen { (rxError: Observable<Error>) -> Observable<Int> in
        return Observable.timer(retryDelay, scheduler: MainScheduler.instance)
    }
    .subscribe(...)
    .disposed(by: disposeBag)
复制代码

// 请求 JSON 失败时，等待 5 秒后重试，
// 重试 4 次后仍然失败，就将错误抛出
// 如果重试超过 4 次，就将错误抛出。如果错误在 4 次以内时，就等待 5 秒后重试

let maxRetryCount = 4       // 最多重试 4 次
let retryDelay: Double = 5  // 重试延时 5 秒

rxJson
    .retryWhen { (rxError: Observable<Error>) -> Observable<Int> in
        return rxError.flatMapWithIndex { (error, index) -> Observable<Int> in
            guard index < maxRetryCount else {
                return Observable.error(error)
            }
            return Observable<Int>.timer(retryDelay, scheduler: MainScheduler.instance)
        }
    }
    .subscribe(...)
    .disposed(by: disposeBag)
复制代码

flatMapWithIndex 这个操作符，因为它可以给我们提供错误的索引数 index。然后用这个索引数判断是否超过最大重试数，如果超过了，就将错误抛出。如果没有超过，就等待 5 秒后重试。

catchError - 恢复以在错误产生时，用一个备用元素或者一组备用元素将错误替换掉

// 先从网络获取数据，如果获取失败了，就从本地缓存获取数据

let rxData: Observable<Data> = ...      // 网络请求的数据
let cahcedData: Observable<Data> = ...  // 之前本地缓存的数据

rxData
   .catchError { _ in cahcedData }
   .subscribe(onNext: { date in
       print("获取数据成功: \(date.count)")
   })
   .disposed(by: disposeBag)
复制代码