[Swift]RxSwift常见用法详解

RxSwift 是 ReactiveX API 的 Swift 版。它是一个基于 Swift 事件驱动的库，用于处理异步和基于事件的代码。

GitHub:https://github.com/ReactiveX/RxSwift

一、安装

首先，你需要安装 RxSwift。你可以使用 CocoaPods，Carthage 或者 Swift Package Manager 来安装。这里是一个使用 CocoaPods 的例子：

在您的 Podfile 中添加以下内容：

pod 'RxSwift', '~> 6.0'
pod 'RxCocoa', '~> 6.0'

然后运行 pod install。
 

二、基本方法

首先，你需要在你的文件中导入 RxSwift：

import RxSwift

1.Observables 可观察序列

RxSwift 的核心组成部分是 Observables。一个 Observable 发出事件，这些事件可以被观察者（Observers）捕获并做出响应。

这是如何创建一个 Observable 的例子：

let myObservable = Observable<String>.create { observer inobserver.onNext("Hello, RxSwift!")observer.onCompleted()return Disposables.create()
}

2.Observers 观察者

Observer 是接收并处理 Observable 发出的事件的实体。你可以使用 subscribe 方法来创建一个 Observer 并订阅一个 Observable：

let myObserver = myObservable.subscribe { event inprint(event)
}

这将会打印 "Hello, RxSwift!" 到控制台。

3.Disposing 和 Dispose Bags

当你订阅一个 Observable，subscribe 方法会返回一个 Disposable 对象。当你不再需要 Observable 时，你应该释放它，以避免内存泄露。你可以通过调用 dispose 方法或者将其添加到一个 DisposeBag 来实现释放。

let disposeBag = DisposeBag()myObservable.subscribe { event inprint(event)
}.disposed(by: disposeBag)

4.Subjects

Subjects 是 Observable 和 Observer 的桥梁。它们既可以订阅其他 Observable 的事件，也可以发出事件。

RxSwift 提供了几种类型的 Subjects，包括 PublishSubject、BehaviorSubject、ReplaySubject 和 Variable。

let subject = PublishSubject<String>()subject.onNext("Hello, RxSwift!")let subscriptionOne = subject.subscribe(onNext: { string inprint(string)}).disposed(by: disposeBag)subject.onNext("Hello again, RxSwift!")subscriptionOne.dispose()let subscriptionTwo = subject.subscribe(onNext: { string inprint(string)}).disposed(by: disposeBag)subject.onNext("Hello again and again, RxSwift!")

这将打印两次 "Hello again, RxSwift!" 和一次 "Hello again and again, RxSwift!"。

这只是使用 RxSwift 的基本方法，RxSwift 提供了许多操作符，例如 map、filter、reduce、concat 等等，你可以使用这些操作符来处理和转换 Observable 发出的事件。
 

三、操作符

1.map

对 Observable 发出的每个元素应用一个转换函数，返回一个经过转换后的新 Observable。

let numbers = Observable.of(1, 2, 3)numbers.map { $0 * 2 } // 将每个元素乘以 2.subscribe(onNext: { value inprint(value) // 输出：2, 4, 6}).disposed(by: disposeBag)

在示例中，`map` 操作符将每个元素乘以 2。

2.flatMap

对 Observable 发出的每个元素应用一个转换函数，返回一个新的 Observable，然后将这些 Observables 合并为一个单一的 Observable。

let numbers = Observable.of(1, 2, 3)numbers.flatMap { value inObservable.of(value, value * 2) // 将每个元素转换为两个元素的 Observable}.subscribe(onNext: { value inprint(value) // 输出：1, 2, 2, 4, 3, 6}).disposed(by: disposeBag)

在示例中，`flatMap` 操作符将每个元素转换为两个元素的 Observable，并将这些 Observable 合并成一个 Observable。

3.compactMap

对 Observable 发出的每个元素应用一个转换函数，过滤掉转换结果为 nil 的元素，并返回一个新的 Observable。

let numbers = Observable.of(1, 2, 3, nil, 4, nil, 5)numbers.compactMap { $0 } // 过滤掉为 nil 的元素.subscribe(onNext: { value inprint(value) // 输出：1, 2, 3, 4, 5}).disposed(by: disposeBag)

在示例中，`compactMap` 操作符过滤掉为 nil 的元素。

4.filter

过滤掉不符合特定条件的元素，只保留符合条件的元素，并返回一个新的 Observable。

let numbers = Observable.of(1, 2, 3, 4, 5)numbers.filter { $0 % 2 == 0 } // 过滤偶数.subscribe(onNext: { value inprint(value) // 输出：2, 4}).disposed(by: disposeBag)

在示例中，`filter` 操作符只发出偶数元素。

5.scan

对 Observable 发出的每个元素应用一个聚合函数，返回一个逐步累积的结果序列。

let numbers = Observable.of(1, 2, 3, 4, 5)numbers.scan(0) { accumulated, value inaccumulated + value}.subscribe(onNext: { value inprint(value) // 输出：1, 3, 6, 10, 15}).disposed(by: disposeBag)

在示例中，`scan` 操作符对每个元素进行累积操作，并发出每次累积的结果。

6.reduce

对 Observable 发出的每个元素应用一个聚合函数，返回一个最终的聚合结果。

 let numbers = Observable.of(1, 2, 3, 4, 5)numbers.reduce(0) { accumulated, value inaccumulated + value}.subscribe(onNext: { value inprint(value) // 输出：15}).disposed(by: disposeBag)

在示例中，`reduce` 操作符对每个元素进行累积操作，最后输出累积的结果。

7.take

从 Observable 中取前 n 个元素，并返回一个新的 Observable。

let numbers = Observable.of(1, 2, 3, 4, 5)numbers.take(3) // 只发出前 3 个元素.subscribe(onNext: { value inprint(value) // 输出：1, 2, 3}).disposed(by: disposeBag)

在示例中，`take` 操作符只发出前 3 个元素。

8.takeWhile

从 Observable 中取元素，直到某个条件不再成立为止，并返回一个新的 Observable。

let numbers = Observable.of(1, 2, 3, 4, 5)numbers.takeWhile { $0 < 4 } // 只发出小于 4 的元素.subscribe(onNext: { value inprint(value) // 输出：1, 2, 3}).disposed(by: disposeBag)

在示例中，`takeWhile` 操作符只发出小于 4 的元素。

9.skip

跳过 Observable 中的前 n 个元素，并返回一个新的 Observable。

let numbers = Observable.of(1, 2, 3, 4, 5)numbers.skip(2) // 跳过前 2 个元素.subscribe(onNext: { value inprint(value) // 输出：3, 4, 5}).disposed(by: disposeBag)

在示例中，`skip` 操作符跳过前 2 个元素，只发出剩余的元素。

10.skipWhile

跳过 Observable 中的元素，直到某个条件不再成立为止，并返回一个新的 Observable。

let numbers = Observable.of(1, 2, 3, 4, 5)numbers.skipWhile { $0 < 3 } // 跳过小于 3 的元素.subscribe(onNext: { value inprint(value) // 输出：3, 4, 5}).disposed(by: disposeBag)

在示例中，skipWhile 操作符跳过小于 3 的元素，然后发出剩余的元素。

11.distinctUntilChanged

过滤掉连续重复的元素，只保留第一个不重复的元素，并返回一个新的 Observable。

let numbers = Observable.of(1, 1, 2, 2, 3, 3, 4, 4, 5)numbers.distinctUntilChanged() // 只发出连续不重复的元素.subscribe(onNext: { value inprint(value) // 输出：1, 2, 3, 4, 5}).disposed(by: disposeBag)

在示例中，distinctUntilChanged 操作符只发出连续不重复的元素。

12.ignoreElements

忽略 Observable 发出的所有元素，只关注 Observable 的终止事件。

let numbers = Observable.of(1, 2, 3, 4, 5)numbers.ignoreElements() // 忽略所有元素，只发出 `completed` 或 `error` 事件.subscribe(onCompleted: {print("Completed")}).disposed(by: disposeBag)

在示例中，ignoreElements操作符忽略了所有元素，只发出了completed事件。

13.elementAt

获取 Observable 发出的指定索引处的元素，并返回一个新的 Observable。

let numbers = Observable.of(1, 2, 3, 4, 5)numbers.elementAt(2) // 只发出索引为 2 的元素.subscribe(onNext: { value inprint(value) // 输出：3}).disposed(by: disposeBag)

在示例中，elementAt 操作符只发出索引为 2 的元素。

14.toArray

将 Observable 发出的所有元素收集到一个数组中，并返回一个新的 Observable。

let disposeBag = DisposeBag()Observable.of(1, 2, 3, 4, 5).toArray().subscribe(onNext: { array inprint(array) // 输出：[1, 2, 3, 4, 5]}).disposed(by: disposeBag)

在上面的示例中，toArray操作符将 Observable 发出的所有元素收集到一个数组中，并将该数组作为单个事件发出给下游的订阅者。这对于需要将单个事件中的多个元素作为整体处理的场景非常有用。

15.merge

将多个 Observable 的元素合并成一个单个的 Observable。

let numbers1 = Observable.of(1, 2, 3)
let numbers2 = Observable.of(4, 5, 6)Observable.merge(numbers1, numbers2) // 合并两个 Observable.subscribe(onNext: { value inprint(value) // 输出：1, 2, 3, 4, 5, 6}).disposed(by: disposeBag)

在示例中，merge操作符将两个 Observable 的元素合并成一个单个的 Observable。

16.zip

将多个 Observable 的元素按顺序一对一地进行组合。

let numbers = Observable.of(1, 2, 3)
let letters = Observable.of("A", "B", "C")Observable.zip(numbers, letters) // 按顺序一对一地组合两个 Observable 的元素.subscribe(onNext: { number, letter inprint("\(number)\(letter)") // 输出：1A, 2B, 3C}).disposed(by: disposeBag)

在示例中，zip操作符按顺序一对一地组合两个 Observable 的元素。

17.combineLatest

将多个 Observable 的最新元素进行组合。

let numbers = Observable.of(1, 2, 3)
let letters = Observable.of("A", "B", "C")Observable.combineLatest(numbers, letters) // 组合两个 Observable 的最新元素.subscribe(onNext: { number, letter inprint("\(number)\(letter)") // 输出：3A, 3B, 3C}).disposed(by: disposeBag)

在示例中，combineLatest 操作符将两个 Observable 的最新元素进行组合。

18.switchLatest

将 Observable 发出的 Observable 转换为一个单个的 Observable，并只发出最新的 Observable 发出的元素。

let subject = BehaviorSubject(value: Observable.of(1, 2, 3))subject.switchLatest() // 转换为单个 Observable，只发出最新的 Observable 发出的元素.subscribe(onNext: { value inprint(value) // 输出：1, 2, 3}).disposed(by: disposeBag)subject.onNext(Observable.of(4, 5, 6)) // 切换到新的 Observable// 输出：4, 5, 6

在示例中，switchLatest 操作符将发出的 Observable 转换为一个单个的 Observable，并只发出最新的 Observable 发出的元素。

19.amb

从多个 Observable 中选择首先发出元素的 Observable，并忽略其它 Observable。

let numbers1 = Observable<Int>.interval(1, scheduler: MainScheduler.instance).take(3)
let numbers2 = Observable<Int>.interval(0.5, scheduler: MainScheduler.instance).take(5)Observable.amb([numbers1, numbers2]) // 选择首先发出元素的 Observable.subscribe(onNext: { value inprint(value) // 输出：0, 1, 2}).disposed(by: disposeBag)

在示例中，amb操作符选择首先发出元素的 Observable。

20.catchError

捕获 Observable 发出的错误，并返回一个新的 Observable 或执行错误处理逻辑。

enum CustomError: Error {case error
}let observable = Observable<Int>.create { observer inobserver.onNext(1)observer.onNext(2)observer.onError(CustomError.error)return Disposables.create()
}observable.catchError { error inreturn Observable.just(3) // 捕获错误并返回新的 Observable 发出默认值 3}.subscribe(onNext: { value inprint(value) // 输出结果: 1, 2, 3}).disposed(by: disposeBag)

在示例中，Observable 发出 1 和 2，然后遇到错误，使用 catchError 操作符捕获错误并返回一个新的 Observable，新的 Observable 发出默认值 3。

21.retry

在遇到错误时重新订阅 Observable，可以指定最大重试次数。

var count = 0let observable = Observable<Int>.create { observer inif count < 3 {observer.onError(NSError(domain: "", code: 0, userInfo: nil))count += 1} else {observer.onNext(1)observer.onCompleted()}return Disposables.create()
}observable.retry(2) // 最多重试 2 次.subscribe(onNext: { value inprint(value) // 输出结果: 1}, onError: { error inprint(error) // 不会产生错误}).disposed(by: disposeBag)

在示例中，Observable 遇到错误时使用 retry 操作符重新订阅 Observable，最多重试 2 次，所以总共尝试 3 次，最终成功发出值 1。

22.repeatElement

重复发出同一个元素的 Observable。

Observable.repeatElement(1).take(3) // 只取前 3 个元素.subscribe(onNext: { value inprint(value) // 输出结果: 1, 1, 1}).disposed(by: disposeBag)

在示例中，使用 repeatElement 操作符创建一个重复发出元素 1 的 Observable，然后使用 take 操作符只取前 3 个元素。

23.delay

延迟 Observable 发出的元素。

Observable.of(1, 2, 3).delay(.seconds(1), scheduler: MainScheduler.instance) // 延迟 1 秒.subscribe(onNext: { value inprint(value) // 输出结果: 1, 2, 3 (每个元素延迟 1 秒)}).disposed(by: disposeBag)

在示例中，Observable 发出的元素被延迟了 1 秒后才被订阅者接收到。

24.throttle

在指定时间间隔内，只发出 Observable 第一个元素，并忽略后续的元素。

Observable<Int>.from([1, 2, 3, 4, 5]).throttle(.milliseconds(500), scheduler: MainScheduler.instance) // 在 500 毫秒内只发出第一个元素.subscribe(onNext: { value inprint(value) // 输出结果: 1, 3, 5 (忽略了 2 和 4)}).disposed(by: disposeBag)

在示例中，throttle 操作符在 500 毫秒内只发出第一个元素，因此忽略了 2 和 4。

25.debounce

只在 Observable 发出元素后的指定时间间隔内没有新元素时才发出该元素。

Observable<Int>.from([1, 2, 3, 4, 5]).debounce(.milliseconds(500), scheduler: MainScheduler.instance) // 在 500 毫秒内只发出最后一个元素.subscribe(onNext: { value inprint(value) // 输出结果: 5 (忽略了 1、2、3、4)}).disposed(by: disposeBag)

在示例中，debounce 操作符在 500 毫秒内只发出最后一个元素，因此忽略了 1、2、3、4。

26.timeout

如果 Observable 在指定的时间内没有发出任何元素或完成事件，就产生一个超时错误。

 Observable<Int>.never().timeout(.seconds(2), scheduler: MainScheduler.instance) // 超时时间为 2 秒.subscribe(onNext: { value inprint(value) // 不会输出结果}, onError: { error inprint(error) // 输出结果: RxError.timeout}).disposed(by: disposeBag)

在示例中，使用 timeout 操作符设置超时时间为 2 秒，由于 Observable 是一个无限的空 Observable，所以在 2 秒后会产生一个超时错误。

27.startWith

在 Observable 发出的元素序列前插入一个指定的元素。

let numbers = Observable.of(1, 2, 3)numbers.startWith(0) // 在序列前插入元素 0.subscribe(onNext: { value inprint(value) // 输出：0, 1, 2, 3}).disposed(by: disposeBag)

在示例中，startWith 操作符在序列前插入元素 0。

28.endWith

在 Observable 发出的元素序列后追加一个指定的元素。

Observable.of(1, 2, 3).endWith(4) // 在 Observable 完成之前先发出结束元素 4.subscribe(onNext: { value inprint(value) // 输出结果: 1, 2, 3, 4}).disposed(by: disposeBag)

在示例中，endWith 操作符在 Observable 完成之前先发出结束元素 4。

29.concat

按顺序连接多个 Observable，当前一个 Observable 完成后，才订阅下一个 Observable。

let observable1 = Observable.of(1, 2)
let observable2 = Observable.of(3, 4)Observable.concat([observable1, observable2]).subscribe(onNext: { value inprint(value) // 输出结果: 1, 2, 3, 4}).disposed(by: disposeBag)

在示例中，使用 concat 操作符将两个 Observables 按顺序连接起来，observable1 先发出 1 和 2，然后 observable2 发出 3 和 4。

30.concatMap

对 Observable 发出的每个元素应用一个转换函数，返回一个新的 Observable，并按顺序连接这些 Observables。

let observable = Observable.of(1, 2, 3)observable.concatMap { value inreturn Observable.of(value * 2, value * 3) // 将每个元素乘以 2 和 3，并按顺序连接输出}.subscribe(onNext: { value inprint(value) // 输出结果: 2, 3, 4, 6, 6, 9}).disposed(by: disposeBag)

在示例中，使用 concatMap 操作符将每个元素乘以 2 和 3，并按顺序连接输出。

31.switchMap

switchMap 操作符将源 Observable 的每个元素转换为一个新的 Observable，然后订阅这个新的 Observable，并只发出这个新的 Observable 的元素。

let subject = PublishSubject<String>()
let newSubject = PublishSubject<String>()subject.switchMap { _ in newSubject }.subscribe(onNext: { print($0) }).disposed(by: disposeBag)subject.onNext("Hello")
newSubject.onNext("World") // Outputs: World

在上述示例中，switchMap 操作符订阅了 newSubject 并输出了它的元素。

32.materialize

将 Observable 发出的元素和事件转换为元素类型为 Event 的 Observable。

let subject = PublishSubject<String>()subject.materialize().subscribe(onNext: { print($0) }).disposed(by: disposeBag)subject.onNext("Hello") // Outputs: next(Hello)
subject.onCompleted() // Outputs: completed

在上述示例中，materialize 操作符将源 Observable 的发射物和通知转换为元素。

33.dematerialize

将 Observable 发出的 Event 元素转换回原始的元素和事件类型的 Observable。

let subject = PublishSubject<Event<String>>()subject.dematerialize().subscribe(onNext: { print($0) }).disposed(by: disposeBag)subject.onNext(Event.next("Hello")) // Outputs: Hello
subject.onNext(Event.completed) // Nothing is printed

在上述示例中，dematerialize 操作符将 materialize 转换的元素还原为源 Observable 的发射物。

34.share

share 操作符将源 Observable 转换为一个可共享的 Observable。

let source = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance).share()source.subscribe(onNext: { print("Subscriber 1: \($0)") }).disposed(by: disposeBag)DispatchQueue.main.asyncAfter(deadline: .now() + 2) {source.subscribe(onNext: { print("Subscriber 2: \($0)") }).disposed(by: disposeBag)
}// Outputs:
// Subscriber 1: 0
// Subscriber 1: 1
// Subscriber 1: 2
// Subscriber 2: 2
// Subscriber 1: 3
// Subscriber 2: 3

在上述示例中，share 操作符使两个订阅者共享同一个 Observable。

35.shareReplay

shareReplay 操作符使得新订阅者可以接收到最近的 n 个元素。

let source = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance).shareReplay(1)source.subscribe(onNext: { print("Subscriber 1: \($0)") }).disposed(by: disposeBag)DispatchQueue.main.asyncAfter(deadline: .now() + 2) {source.subscribe(onNext: { print("Subscriber 2: \($0)") }).disposed(by: disposeBag)
}// Outputs:
// Subscriber 1: 0
// Subscriber 1: 1
// Subscriber 1: 2
// Subscriber 2: 1
// Subscriber 1: 3
// Subscriber 2: 3

在上述示例中，shareReplay 操作符使得新订阅者可以接收到最近的一个元素。

36.publish

publish 操作符会将源 Observable 转换为一个 ConnectableObservable。只有当 connect 操作符被调用时，它才开始发出元素。

let source = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance).publish()source.subscribe(onNext: { print("Subscriber 1: \($0)") }).disposed(by: disposeBag)DispatchQueue.main.asyncAfter(deadline: .now() + 2) {source.subscribe(onNext: { print("Subscriber 2: \($0)") }).disposed(by: disposeBag)
}source.connect()// Outputs:
// Subscriber 1: 0
// Subscriber 2: 0
// Subscriber 1: 1
// Subscriber 2: 1

在上述示例中，publish 操作符使源 Observable 变成一个 ConnectableObservable，并在 connect 被调用后开始发出元素。

37.multicast

multicast 操作符将源 Observable 转换为一个 ConnectableObservable，并允许你指定一个主题作为中介。只有当 connect 操作符被调用时，它才开始发出元素。

let subject = PublishSubject<Int>()
let source = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance).multicast(subject)source.subscribe(onNext: { print("Subscriber 1: \($0)") }).disposed(by: disposeBag)DispatchQueue.main.asyncAfter(deadline: .now() + 2) {source.subscribe(onNext: { print("Subscriber 2: \($0)") }).disposed(by: disposeBag)
}source.connect()// Outputs:
// Subscriber 1: 0
// Subscriber 2: 0
// Subscriber 1: 1
// Subscriber 2: 1

在上述示例中，multicast 操作符使源 Observable 变成一个 ConnectableObservable，并在 connect 被调用后开始发出元素。

38.refCount

refCount 操作符将 ConnectableObservable 转换为普通的 Observable。当订阅者数量从 0 增加到 1 时，它开始发出元素。当订阅者数量从 1 变为 0 时，它停止发出元素。

let source = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance).publish().refCount()source.subscribe(onNext: { print("Subscriber 1: \($0)") }).disposed(by: disposeBag)DispatchQueue.main.asyncAfter(deadline: .now() + 2) {source.subscribe(onNext: { print("Subscriber 2: \($0)") }).disposed(by: disposeBag)
}// Outputs:
// Subscriber 1: 0
// Subscriber 1: 1
// Subscriber 1: 2
// Subscriber 2: 2
// Subscriber 1: 3
// Subscriber 2: 3

在上述示例中，refCount 操作符使源 Observable 在订阅者数量从 0 增加到 1 时开始发出元素。

39.replay

replay 操作符将源 Observable 转换为一个 ConnectableObservable，并当新的订阅者订阅它时发送最近的 n 个元素。

let source = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance).replay(3)source.subscribe(onNext: { print("Subscriber 1: \($0)") }).disposed(by: disposeBag)DispatchQueue.main.asyncAfter(deadline: .now() + 2) {source.subscribe(onNext: { print("Subscriber 2: \($0)") }).disposed(by: disposeBag)
}source.connect()// Outputs:
// Subscriber 1: 0
// Subscriber 1: 1
// Subscriber 2: 0
// Subscriber 2: 1
// Subscriber 1: 2
// Subscriber 2: 2

在上述示例中，replay 操作符使源 Observable 在新的订阅者订阅时发送最近的 n 个元素。

40.sample

定期从 Observable 中取样并发出最新的元素。

let source = PublishSubject<String>()
let notifier = PublishSubject<Void>()source.sample(notifier).subscribe(onNext: { print($0) }).disposed(by: disposeBag)source.onNext("Hello")
notifier.onNext(()) // Nothing is printed
source.onNext("World")
notifier.onNext(()) // Outputs: World

在示例中，首先没有调用notifier.onNext(())，因此在执行source.onNext("Hello")之后，notifier并未发出任何元素。因此，当sample操作符在没有采样信号的情况下运行时，不会发射任何元素，因此控制台中不会输出任何内容，即输出为"Nothing is printed"。

只有在调用notifier.onNext(())之后，notifier发出了一个采样信号，此时sample操作符才会从源Observable source中选择最新的元素进行发射。因此，在第二次调用notifier.onNext(())之后，sample操作符选择了最新的元素"World"并将其发射出来，控制台输出"World"。

41.takeUntil

当另一个 Observable 发出元素或完成时，停止发出原始 Observable 的元素。

takeUntil 操作符会订阅并发出源 Observable 的元素，直到第二个 Observable 发出元素。

let source = PublishSubject<String>()
let stopper = PublishSubject<String>()source.takeUntil(stopper).subscribe(onNext: { print($0) }).disposed(by: disposeBag)source.onNext("Hello") // Outputs: Hello
source.onNext("World") // Outputs: World
stopper.onNext("Enough")
source.onNext("!!!") // Nothing is printed

在上述示例中，takeUntil 操作符停止了 source 的元素的输出，一旦 stopper 发出元素。

42.skipUntil

skipUntil 操作符与 takeUntil 操作符正好相反，它会忽略源 Observable 的元素，直到第二个 Observable 发出元素。

let source = PublishSubject<String>()
let notifier = PublishSubject<String>()source.skipUntil(notifier).subscribe(onNext: { print($0) }).disposed(by: disposeBag)source.onNext("Hello") // Nothing is printed
notifier.onNext("OK")
source.onNext("World") // Outputs: World

在上述示例中，skipUntil 操作符忽略了 notifier 发出元素之前的所有 source 的元素。

43.takeLast

takeLast 操作符只会发出源 Observable 的最后 n 个元素。

let source = PublishSubject<String>()source.takeLast(2).subscribe(onNext: { print($0) }).disposed(by: disposeBag)source.onNext("1")
source.onNext("2")
source.onNext("3")
source.onCompleted() // Outputs: 2, 3

在上述示例中，takeLast 操作符只输出了最后两个元素。

44.skipLast

skipLast 操作符会跳过源 Observable 的最后 n 个元素。

let source = PublishSubject<String>()source.skipLast(2).subscribe(onNext: { print($0) }).disposed(by: disposeBag)source.onNext("1") // Outputs: 1
source.onNext("2")
source.onNext("3")
source.onCompleted()

在上述示例中，skipLast 操作符跳过了最后两个元素，只输出了第一个元素。

45.buffer

buffer 操作符会定期的从源 Observable 收集元素，并将这些元素作为一个数组发出。

let source = PublishSubject<String>()source.buffer(timeSpan: .seconds(1), count: 2, scheduler: MainScheduler.instance).subscribe(onNext: { print($0) }).disposed(by: disposeBag)source.onNext("1")
source.onNext("2") // Outputs: ["1", "2"]
source.onNext("3")

46.window

`window` 操作符与 `buffer` 类似，但是它会将元素集合在一个 Observable 中，而不是一个数组。

let source = PublishSubject<String>()source.window(timeSpan: .seconds(1), count: 2, scheduler: MainScheduler.instance).flatMap { $0.toArray() }.subscribe(onNext: { print($0) }).disposed(by: disposeBag)source.onNext("1")
source.onNext("2") // Outputs: ["1", "2"]
source.onNext("3")
source.onNext("4") // Outputs: ["3", "4"]

在上述示例中，window 操作符收集了每两个元素，并将它们作为一个数组发出。

47.repeat

重复订阅和发出 Observable 的元素，可以指定重复次数。

Observable.of("Hello").repeatElement(3).subscribe(onNext: { print($0) }).disposed(by: disposeBag) // Outputs: Hello, Hello, Hello

在上述示例中，"Hello" 被重复三次。

48.amb

从多个 Observable 中选择首先发出元素的 Observable，并忽略其它 Observable。

let subject1 = PublishSubject<String>()
let subject2 = PublishSubject<String>()Observable.amb([subject1, subject2]).subscribe(onNext: { print($0) }).disposed(by: disposeBag)subject2.onNext("Hello from subject 2") // Outputs: Hello from subject 2
subject1.onNext("Hello from subject 1") // Nothing is printed

在上述示例中，amb 操作符选择了首先发出元素的 Observable。

49.timeout

timeout 操作符将在指定的时间间隔过去后，如果源 Observable 还没有发出任何元素，就会发出一个错误。

let subject = PublishSubject<String>()subject.timeout(.seconds(2), scheduler: MainScheduler.instance).subscribe(onNext: { print($0) }, onError: { print($0) }).disposed(by: disposeBag)DispatchQueue.main.asyncAfter(deadline: .now() + 3) {subject.onNext("Hello") // Outputs: The operation couldn’t be completed. (RxSwift.RxError error 5.)
}

在上述示例中，由于源 Observable 在 2 秒内没有发出任何元素，因此 timeout 操作符发出了一个错误。

50.debug

debug 操作符将在控制台打印所有源 Observable 的订阅、事件和状态。

let subject = PublishSubject<String>()subject.debug("Observable").subscribe().disposed(by: disposeBag)subject.onNext("Hello") // Outputs: Observable -> subscribed, Observable -> Event next(Hello)
subject.onCompleted() // Outputs: Observable -> Event completed, Observable -> isDisposed

在上述示例中，debug 操作符打印了所有源 Observable 的订阅、事件和状态。
 

四、实际应用

1. 表单验证

传统方法：

在一个注册表单中，我们希望保证用户只有在所有字段（例如用户名、密码、确认密码、电子邮件等）都有效时才能点击提交按钮。在传统的方法中，我们可能需要为每个字段写一个函数，当字段改变时调用。然后在每个函数中重新检查所有字段，并决定是否启用提交按钮。

class SignUpViewController: UIViewController {@IBOutlet weak var usernameField: UITextField!@IBOutlet weak var passwordField: UITextField!@IBOutlet weak var confirmPasswordField: UITextField!@IBOutlet weak var signUpButton: UIButton!// 当文本字段中的文本更改时，调用此函数@IBAction func textChanged(_ sender: UITextField) {let isValid = validateForm()signUpButton.isEnabled = isValid // 根据表单是否有效来启用或禁用注册按钮}// 验证表单// 如果用户名、密码和确认密码都不为空，并且密码与确认密码匹配，则返回 true，否则返回 falsefunc validateForm() -> Bool {guard let username = usernameField.text, !username.isEmpty,let password = passwordField.text, !password.isEmpty,let confirmPassword = confirmPasswordField.text, !confirmPassword.isEmpty,password == confirmPassword else {return false}return true}
}

RxSwift：

使用 RxSwift，我们可以将这些字段看作 Observables，并使用 combineLatest 方法来创建一个新的 Observable，该 Observable 表示表单的有效性。然后我们可以订阅这个 Observable，当它发出新的值时，我们可以启用或禁用提交按钮。这样，我们的代码将更清晰，更易于维护。

import RxSwift
import RxCocoaclass RxSignUpViewController: UIViewController {@IBOutlet weak var usernameField: UITextField!@IBOutlet weak var passwordField: UITextField!@IBOutlet weak var confirmPasswordField: UITextField!@IBOutlet weak var signUpButton: UIButton!let disposeBag = DisposeBag() // 用于存储订阅override func viewDidLoad() {super.viewDidLoad()// 将用户名、密码和确认密码的文本组合成一个 Observable// 然后使用 map 操作符将这些文本转换为一个布尔值，表示表单是否有效// 最后，将这个布尔值绑定到注册按钮的 isEnabled 属性，这样当表单是否有效改变时，按钮的启用状态会自动更新Observable.combineLatest(usernameField.rx.text.orEmpty,passwordField.rx.text.orEmpty,confirmPasswordField.rx.text.orEmpty).map { username, password, confirmPassword inreturn !username.isEmpty && !password.isEmpty && !confirmPassword.isEmpty && password == confirmPassword}.bind(to: signUpButton.rx.isEnabled).disposed(by: disposeBag)}
}

2. 网络请求

传统方法：

在传统的网络请求中，我们可能会使用回调或者 Promise/Future。这可以让我们的代码保持异步，但是当我们需要进行多个连续的网络请求，或者需要取消网络请求时，代码可能会变得复杂和难以维护。

class NetworkRequestViewController: UIViewController {func fetchData() {// 使用 URLSession 发起网络请求// 当请求完成时，我们会得到 data、response 和 error，并打印它们URLSession.shared.dataTask(with: URL(string: "https://example.com")!) { data, response, error inif let error = error {print("Error: \(error)")} else if let data = data {print("Data: \(data)")}}.resume()}
}

RxSwift：

使用 RxSwift，我们可以将网络请求看作 Observables，这样就可以使用 RxSwift 提供的各种操作符（例如 map、filter、concat 等）来处理网络请求。例如，我们可以使用 flatMap 操作符来进行连续的网络请求，我们可以使用 takeUntil 操作符来取消网络请求，等等。

import RxSwift
import RxCocoaclass RxNetworkRequestViewController: UIViewController {let disposeBag = DisposeBag() // 用于存储订阅func fetchData() {let url = URL(string: "https://example.com")!// 使用 URLSession 的 rx 扩展发起网络请求// 当请求完成时，我们会得到 data 或 error，并打印它们URLSession.shared.rx.data(request: URLRequest(url: url)).subscribe(onNext: { data inprint("Data: \(data)")}, onError: { error inprint("Error: \(error)")}).disposed(by: disposeBag)}
}

3. UI 更新

传统方法：

在传统的方法中，我们可能需要在多个地方更新 UI。例如，当数据改变时，我们可能需要在 setter 方法中更新 UI，也可能需要在网络请求的回调中更新 UI。

class UpdateUIViewController: UIViewController {@IBOutlet weak var label: UILabel!// 当数据更新时，我们会更新 label 的文本var data: String? {didSet {label.text = data}}
}

RxSwift：

使用 RxSwift，我们可以将数据看作 Observables，然后我们可以订阅这些 Observables，当它们发出新的值时，我们可以更新 UI。这样，我们的 UI 更新逻辑将更集中，更易于维护。

import RxSwift
import RxCocoaclass RxUpdateUIViewController: UIViewController {@IBOutlet weak var label: UILabel!let disposeBag = DisposeBag() // 用于存储订阅let data = PublishSubject<String>() // 数据源override func viewDidLoad() {super.viewDidLoad()// 将数据绑定到 label 的 text 属性，这样当数据有新的值时，label 的文本会自动更新data.bind(to: label.rx.text).disposed(by: disposeBag)}
}

4.搜索

传统方法：

如果我们想要实现一个搜索框，当用户输入文本时，我们会请求 API 获取搜索结果并更新 UI。不使用 RxSwift 的情况下，我们可能会这样做：

class SearchViewController: UIViewController {@IBOutlet weak var searchBar: UISearchBar!@IBOutlet weak var tableView: UITableView!var searchResults: [String] = [] {didSet {tableView.reloadData()}}override func viewDidLoad() {super.viewDidLoad()searchBar.delegate = self}
}extension SearchViewController: UISearchBarDelegate {func searchBar(_ searchBar: UISearchBar, textDidChange searchText: String) {// 这里我们模拟网络请求，实际使用中会调用实际的 APIDispatchQueue.main.asyncAfter(deadline: .now() + 1.0) { [weak self] inself?.searchResults = ["Result 1", "Result 2", "Result 3"]}}
}

在这个例子中，搜索框的代理方法 searchBar(_:textDidChange:) 被调用时，我们模拟一个网络请求。当网络请求完成时，我们更新 searchResults，这将触发 tableView 的重新加载。

RxSwift：

使用 RxSwift，我们可以将搜索框的文本变化看作一个 Observable，然后我们可以订阅这个 Observable，当它发出新的值时，我们可以请求 API 并更新 UI。这是使用 RxSwift 的版本：

import RxSwift
import RxCocoaclass RxSearchViewController: UIViewController {@IBOutlet weak var searchBar: UISearchBar!@IBOutlet weak var tableView: UITableView!let disposeBag = DisposeBag()override func viewDidLoad() {super.viewDidLoad()searchBar.rx.text.orEmpty.debounce(.milliseconds(300), scheduler: MainScheduler.instance) // 防抖动.distinctUntilChanged() // 仅当新的值和前一个值不相同时才发出.flatMapLatest { query -> Observable<[String]> inreturn self.fetchSearchResults(query) // 获取搜索结果}.bind(to: tableView.rx.items(cellIdentifier: "Cell")) { index, model, cell incell.textLabel?.text = model}.disposed(by: disposeBag)}func fetchSearchResults(_ query: String) -> Observable<[String]> {// 模拟网络请求return Observable.just(["Result 1", "Result 2", "Result 3"])}
}

在这个例子中，我们使用了几个 RxSwift 的操作符。debounce 操作符可以防止我们在用户还在输入时就发送网络请求。distinctUntilChanged 操作符可以保证我们只在搜索框的文本实际改变时才发送网络请求。flatMapLatest 操作符可以保证我们总是获取最新搜索框文本的搜索结果。最后，我们使用 bind(to:) 方法将搜索结果绑定到表视图，这样当搜索结果改变时，表视图会自动更新。
 

bind(to:) 方法是将 Observable 值绑定到特定的对象，例如在上述例子中的 UITableView。这使得当我们的数据源（Observable）发出新的元素时，UITableView 会自动更新，显示新的数据。

在我们的例子中，我们使用了 tableView.rx.items(cellIdentifier: "Cell") 作为绑定的目标。这是 RxSwift 提供的一个方法，它会返回一个 Observer，这个 Observer 会在每次接收到新的元素时更新 UITableView。

当我们将搜索结果绑定到 tableView.rx.items(cellIdentifier: "Cell") 时，我们提供了一个闭包，这个闭包会在每个新的搜索结果到来时被调用。

tableView.rx.items(cellIdentifier: "Cell") { index, model, cell incell.textLabel?.text = model
}

在这个闭包里，我们有三个参数：

• index：当前元素的索引，也就是当前行号。

• model：当前元素的值，也就是搜索结果中的一个值。

• cell：当前的 UITableViewCell，我们需要在这个 cell 上展示数据。

所以，这个闭包的作用就是将搜索结果（model）展示在 UITableViewCell（cell）上。这样，每当我们的搜索结果有新的值时，UITableView 就会自动更新，显示新的搜索结果。

5. 处理用户输入

在这个例子中，我们处理用户在文本字段中的输入。使用 RxSwift，我们可以订阅文本字段的 text 属性，并在用户输入新的文本时自动打印它。

传统方法：

class UserInputViewController: UIViewController {@IBOutlet weak var textField: UITextField!override func viewDidLoad() {super.viewDidLoad()textField.addTarget(self, action: #selector(textFieldDidChange(_:)), for: .editingChanged)}@objc func textFieldDidChange(_ textField: UITextField) {if let text = textField.text {print("User input: \(text)")}}
}

RxSwift：

import RxSwift
import RxCocoaclass RxUserInputViewController: UIViewController {@IBOutlet weak var textField: UITextField!let disposeBag = DisposeBag()override func viewDidLoad() {super.viewDidLoad()textField.rx.text.orEmpty.subscribe(onNext: { text inprint("User input: \(text)")}).disposed(by: disposeBag)}
}

6. 定时操作

在这个例子中，我们创建一个定时器，每秒打印一次 "Timer fired!"。在 RxSwift 中，我们可以使用 interval 操作符来创建一个定时器。

传统方法：

class TimerViewController: UIViewController {var timer: Timer?override func viewDidLoad() {super.viewDidLoad()timer = Timer.scheduledTimer(withTimeInterval: 1.0, repeats: true) { _ inprint("Timer fired!")}}deinit {timer?.invalidate()}
}

RxSwift：

import RxSwift
import RxCocoaclass RxTimerViewController: UIViewController {let disposeBag = DisposeBag()override func viewDidLoad() {super.viewDidLoad()Observable<Int>.interval(RxTimeInterval.seconds(1), scheduler: MainScheduler.instance).subscribe(onNext: { _ inprint("Timer fired!")}).disposed(by: disposeBag)}
}

7. 错误处理

在这个例子中，我们处理网络请求的错误。在 RxSwift 中，错误被视为一种终止序列的事件，我们可以在订阅时处理它。

传统方法：

class ErrorHandlingViewController: UIViewController {func fetchData(completion: @escaping (Data?, Error?) -> Void) {URLSession.shared.dataTask(with: URL(string: "https://example.com")!) { data, _, error incompletion(data, error)}.resume()}func doSomething() {fetchData { data, error inif let error = error {print("Error: \(error)")} else if let data = data {print("Data: \(data)")}}}
}

RxSwift：

import RxSwift
import RxCocoaclass RxErrorHandlingViewController: UIViewController {let disposeBag = DisposeBag()func fetchData() -> Observable<Data> {return URLSession.shared.rx.data(request: URLRequest(url: URL(string: "https://example.com")!))}func doSomething() {fetchData().subscribe(onNext: { data inprint("Data: \(data)")}, onError: { error inprint("Error: \(error)")}).disposed(by: disposeBag)}
}

8.集合操作

在这个示例中，我们在一个集合中进行过滤操作。使用 RxSwift，我们可以将数组变成一个 Observable，并使用 map 操作符进行过滤。

传统方法：

class CollectionViewController: UIViewController {var items: [Int] = []func updateItems() {items = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]print("Updated items: \(items)")}func filterItems() {items = items.filter { $0 % 2 == 0 }print("Filtered items: \(items)")}
}

RxSwift：

import RxSwiftclass RxCollectionViewController: UIViewController {let disposeBag = DisposeBag()let items = BehaviorSubject(value: [Int]())func updateItems() {items.onNext([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])}func filterItems() {items.asObservable().map { $0.filter { $0 % 2 == 0 } }.subscribe(onNext: { items inprint("Filtered items: \(items)")}).disposed(by: disposeBag)}
}

9.多个请求并行

在这个示例中，我们并行完成多个网络请求。使用 RxSwift，我们可以将 URL 数组转换为 Observable，并使用 flatMap 和 toArray 操作符进行并行请求。

传统方法：

class ParallelRequestsViewController: UIViewController {func fetchData(from urls: [URL], completion: @escaping ([Data]) -> Void) {let group = DispatchGroup()var results: [Data] = []for url in urls {group.enter()URLSession.shared.dataTask(with: url) { data, _, _ inif let data = data {results.append(data)}group.leave()}.resume()}group.notify(queue: .main) {completion(results)}}
}

RxSwift：

import RxSwift
import RxCocoaclass RxParallelRequestsViewController: UIViewController {let disposeBag = DisposeBag()func fetchData(from urls: [URL]) -> Observable<[Data]> {return Observable.from(urls).flatMap { url inreturn URLSession.shared.rx.data(request: URLRequest(url: url))}.toArray()}
}

10.交换响应

在这个示例中，我们获取网络请求的响应，并将其转换为我们需要的形式。使用 RxSwift，我们可以使用 map 操作符进行转换。

传统方法：

class TransformResponseViewController: UIViewController {func fetchData(completion: @escaping (String?) -> Void) {URLSession.shared.dataTask(with: URL(string: "https://jsonplaceholder.typicode.com/posts/1")!) { data, _, _ inif let data = data {let json = try? JSONSerialization.jsonObject(with: data, options: []) as? [String: Any]completion(json?["title"] as? String)}}.resume()}
}

RxSwift：

import RxSwift
import RxCocoaclass RxTransformResponseViewController: UIViewController {let disposeBag = DisposeBag()func fetchData() -> Observable<String> {let url = URL(string: "https://jsonplaceholder.typicode.com/posts/1")!return URLSession.shared.rx.json(request: URLRequest(url: url)).map { json inlet dictionary = json as? [String: Any]return dictionary?["title"] as? String ?? ""}}
}

11.事件排队

在这个示例中，我们在事件队列中添加事件，并在2秒后处理事件。在 RxSwift 中，我们可以使用 buffer 操作符来实现这个功能。

传统方法：

class EventQueueViewController: UIViewController {var events: [String] = []var timer: Timer?func enqueue(event: String) {events.append(event)if timer == nil {timer = Timer.scheduledTimer(withTimeInterval: 2.0, repeats: true) { _ inif !self.events.isEmpty {print("Processing event: \(self.events.removeFirst())")} else {self.timer?.invalidate()self.timer = nil}}}}
}

RxSwift：

import RxSwift
import RxCocoaclass RxEventQueueViewController: UIViewController {let disposeBag = DisposeBag()let eventSubject = PublishSubject<String>()override func viewDidLoad() {super.viewDidLoad()eventSubject.buffer(timeSpan: .seconds(2), count: 1, scheduler: MainScheduler.instance).subscribe(onNext: { events inif let event = events.first {print("Processing event: \(event)")}}).disposed(by: disposeBag)}func enqueue(event: String) {eventSubject.onNext(event)}
}

12.组合多个操作

在这个示例中，我们需要组合多个操作：首先获取用户，然后获取该用户的帖子。在 RxSwift 中，我们可以使用 flatMap 和 map 操作符来组合这些操作。

传统方法：

class CombineOperationsViewController: UIViewController {var user: User?var posts: [Post]?func fetchUser(completion: @escaping (User?) -> Void) {// Fetch the user...}func fetchPosts(for user: User, completion: @escaping ([Post]?) -> Void) {// Fetch the posts for the user...}func updateUserAndPosts() {fetchUser { user inself.user = userif let user = user {self.fetchPosts(for: user) { posts inself.posts = posts}}}}
}

RxSwift：

import RxSwift
import RxCocoaclass RxCombineOperationsViewController: UIViewController {let disposeBag = DisposeBag()func fetchUser() -> Observable<User> {// Fetch the user...return Observable.just(User())}func fetchPosts(for user: User) -> Observable<[Post]> {// Fetch the posts for the user...return Observable.just([Post]())}func updateUserAndPosts() {fetchUser().flatMap { user inself.fetchPosts(for: user).map { posts in (user, posts) }}.subscribe(onNext: { user, posts inprint("User: \(user), Posts: \(posts)")}).disposed(by: disposeBag)}
}

13.通知

不论你是在使用传统的 NotificationCenter 还是 RxSwift 的 NotificationCenter 扩展，发送通知的方式都是一样的。

传统方法：

class MyViewController: UIViewController {let notificationName = Notification.Name("MyNotification")override func viewDidLoad() {super.viewDidLoad()NotificationCenter.default.addObserver(self, selector: #selector(self.handleNotification(_:)), name: notificationName, object: nil)}@objc func handleNotification(_ notification: Notification) {// Handle the notificationprint("Notification received!")}deinit {NotificationCenter.default.removeObserver(self)}
}

RxSwift：

class MyViewController: UIViewController {let disposeBag = DisposeBag()let notificationName = Notification.Name("MyNotification")override func viewDidLoad() {super.viewDidLoad()NotificationCenter.default.rx.notification(notificationName).subscribe(onNext: { [weak self] notification in// Handle the notificationprint("Notification received!")}).disposed(by: disposeBag)}
}