RxSwift 文档:Observable

Observable - 可被监听的序列


所有的事物都是序列

Observable 可以用于描述元素异步产生的序列。这样我们生活中许多事物都可以通过它来表示。

如何创建序列

创建序列最直接的方法就是调用 Observable.create,然后在构建函数里面描述元素的产生过程。 observer.onNext(0) 就代表产生了一个元素,他的值是 0。后面又产生了 9 个元素分别是 1, 2, ... 8, 9 。最后,用 observer.onCompleted() 表示元素已经全部产生,没有更多元素了。

你可以用这种方式来封装功能组件,例如,闭包回调:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
typealias JSON = Any

let json: Observable<JSON> = Observable.create { (observer) -> Disposable in

let task = URLSession.shared.dataTask(with: ...) { data, _, error in

guard error == nil else {
observer.onError(error!)
return
}

guard let data = data,
let jsonObject = try? JSONSerialization.jsonObject(with: data, options: .mutableLeaves)
else {
observer.onError(DataError.cantParseJSON)
return
}
observer.onNext(jsonObject)
observer.onCompleted()
}

task.resume()

return Disposables.create { task.cancel() }
}

在闭包回调中,如果任务失败,就调用 observer.onError(error!)。如果获取到目标元素,就调用 observer.onNext(jsonObject)。由于我们的这个序列只有一个元素,所以在成功获取到元素后,就直接调用 observer.onCompleted() 来表示任务结束。最后 Disposables.create { task.cancel() } 说明如果数据绑定被清除(订阅被取消)的话,就取消网络请求。

这样一来我们就将传统的闭包回调转换成序列了。然后可以用 subscribe 方法来响应这个请求的结果。

Single


SingleObservable 的另外一个版本。不像 Observable 可以发出多个元素,它要么只能发出一个元素,要么产生一个 error 事件。

  • 发出一个元素,或一个 error 事件
  • 不会共享状态变化

一个比较常见的例子就是执行 HTTP 请求,然后返回一个应答或错误。不过你也可以用 Single 来描述任何只有一个元素的序列。

如何创建 Single

创建 Single 和创建 Observable 非常相似:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
func getRepo(_ repo: String) -> Single<[String: Any]> {

return Single<[String: Any]>.create { single in
let url = URL(string: "https://api.github.com/repos/\(repo)")!
let task = URLSession.shared.dataTask(with: url) {
data, _, error in

if let error = error {
single(.error(error))
return
}

guard let data = data,
let json = try? JSONSerialization.jsonObject(with: data, options: .mutableLeaves),
let result = json as? [String: Any] else {
single(.error(DataError.cantParseJSON))
return
}

single(.success(result))
}

task.resume()

return Disposables.create { task.cancel() }
}
}

订阅提供一个 SingleEvent 的枚举:

1
2
3
4
public enum SingleEvent<Element> {
case success(Element)
case error(Swift.Error)
}

注意 Single 与普通 Observable 序列的事件枚举的不同。

Completable


CompletableObservable 的另外一个版本。不像 Observable 可以发出多个元素,它要么只能产生一个 completed 事件,要么产生一个 error 事件。

  • 发出零个元素
  • 发出一个 completed 事件或者一个 error 事件
  • 不会共享状态变化

Completable 适用于那种你只关心任务是否完成,而不需要在意任务返回值的情况。它和 Observable<Void> 有点相似。

如何创建 Completable

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
func cacheLocally() -> Completable {
return Completable.create { completable in
// Store some data locally
...
...

guard success else {
completable(.error(CacheError.failedCaching))
return Disposables.create {}
}

completable(.completed)
return Disposables.create {}
}
}

订阅提供一个 CompletableEvent 的枚举:

1
2
3
4
public enum CompletableEvent {
case error(Swift.Error)
case completed
}

Maybe


MaybeObservable 的另外一个版本。它介于 SingleCompletable 之间,它要么只能发出一个元素,要么产生一个 completed 事件,要么产生一个 error 事件。

发出一个元素或者一个 completed 事件或者一个 error 事件
不会共享状态变化
如果你遇到那种可能需要发出一个元素,又可能不需要发出时,就可以使用 Maybe

如何创建 Maybe

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
func generateString() -> Maybe<String> {
return Maybe<String>.create { maybe in
maybe(.success("RxSwift"))

// OR

maybe(.completed)

// OR

maybe(.error(error))

return Disposables.create {}
}
}

Driver


Driver 是一个精心准备的特征序列。它主要是为了简化 UI 层的代码。不过如果你遇到的序列具有以下特征,你也可以使用它:

  • 不会产生 error 事件
  • 一定在 MainScheduler 监听(主线程监听)
  • 共享状态变化

这些都是驱动 UI 的序列所具有的特征。

为什么要使用 Driver ?

我们举个例子来说明一下,为什么要使用 Driver

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
let results = query.rx.text
.throttle(0.3, scheduler: MainScheduler.instance)
.flatMapLatest { query in
fetchAutoCompleteItems(query)
}

results
.map { "\($0.count)" }
.bind(to: resultCount.rx.text)
.disposed(by: disposeBag)

results
.bind(to: resultsTableView.rx.items(cellIdentifier: "Cell")) {
(_, result, cell) in
cell.textLabel?.text = "\(result)"
}
.disposed(by: disposeBag)

这段代码的主要目的是:

  • 取出用户输入稳定后的内容
  • 向服务器请求一组结果
  • 将返回的结果绑定到两个 UI 元素上:tableView 和 显示结果数量的label

那么这里存在什么问题?

  • 如果 fetchAutoCompleteItems 的序列产生了一个错误(网络请求失败),这个错误将取消所有绑定,当用户输入一个新的关键字时,是无法发起新的网络请求。
  • 如果 fetchAutoCompleteItems 在后台返回序列,那么刷新页面也会在后台进行,这样就会出现异常崩溃。
  • 返回的结果被绑定到两个 UI 元素上。那就意味着,每次用户输入一个新的关键字时,就会分别为两个 UI 元素发起 HTTP 请求,这并不是我们想要的结果。

一个更好的方案是这样的:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
let results = query.rx.text
.throttle(0.3, scheduler: MainScheduler.instance)
.flatMapLatest { query in
fetchAutoCompleteItems(query)
.observeOn(MainScheduler.instance) // 结果在主线程返回
.catchErrorJustReturn([]) // 错误被处理了,这样至少不会终止整个序列
}
.share(replay: 1) // HTTP 请求是被共享的

results
.map { "\($0.count)" }
.bind(to: resultCount.rx.text)
.disposed(by: disposeBag)

results
.bind(to: resultsTableView.rx.items(cellIdentifier: "Cell")) {
(_, result, cell) in
cell.textLabel?.text = "\(result)"
}
.disposed(by: disposeBag)

以下是使用 Driver 优化后的代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
let results = query.rx.text.asDriver()        // 将普通序列转换为 Driver
.throttle(0.3, scheduler: MainScheduler.instance)
.flatMapLatest { query in
fetchAutoCompleteItems(query)
.asDriver(onErrorJustReturn: []) // 仅仅提供发生错误时的备选返回值
}

results
.map { "\($0.count)" }
.drive(resultCount.rx.text) // 这里改用 `drive` 而不是 `bindTo`
.disposed(by: disposeBag) // 这样可以确保必备条件都已经满足了

results
.drive(resultsTableView.rx.items(cellIdentifier: "Cell")) {
(_, result, cell) in
cell.textLabel?.text = "\(result)"
}
.disposed(by: disposeBag)

任何可被监听的序列都可以被转换为 Driver,只要他满足 3 个条件:

  • 不会产生 error 事件
  • 一定在 MainScheduler 监听(主线程监听)
  • 共享状态变化

那么要如何确定条件都被满足?通过 Rx 操作符来进行转换。asDriver(onErrorJustReturn: []) 相当于以下代码:

1
2
3
4
5
let safeSequence = xs
.observeOn(MainScheduler.instance) // 主线程监听
.catchErrorJustReturn(onErrorJustReturn) // 无法产生错误
.share(replay: 1, scope: .whileConnected)// 共享状态变化
return Driver(raw: safeSequence) // 封装

最后使用 drive 而不是 bindTo

drive 方法只能被 Driver 调用。这意味着,如果你发现代码所存在 drive,那么这个序列不会产生错误事件并且一定在主线程监听。这样你可以安全的绑定 UI 元素。

ControlEvent


ControlEvent 专门用于描述 UI 控件所产生的事件,它具有以下特征:

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