RxSwift學習

RxSwift · 發表 2018-11-21 13:56:56

摘要： “它拓展了觀察者模式。使你能夠自由組合多個非同步事件，而不需要去關心執行緒，同步，執行緒安全，併發資料以及I/O阻塞。” 1.2主要具有哪些功能 1.2.1 實現點選按鈕方法Target Action button.rx.tap .subscribe(onNext: { ...

“它拓展了觀察者模式。使你能夠自由組合多個非同步事件，而不需要去關心執行緒，同步，執行緒安全，併發資料以及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)
複製程式碼