[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 RxSwift1.Observables 可观察序列
RxSwift 的核心组成部分是 Observables。一个 Observable 发出事件，这些事件可以被观察者（Observers）捕获并做出响应。
这是如何创建一个 Observable 的例子：
let myObservable = Observable<String>.create { observer in
observer.onNext("Hello, RxSwift!")
observer.onCompleted()
return Disposables.create()
}2.Observers 观察者
Observer 是接收并处理 Observable 发出的事件的实体。你可以使用 subscribe 方法来创建一个 Observer 并订阅一个 Observable：
let myObserver = myObservable.subscribe { event in
print(event)
}这将会打印 "Hello, RxSwift!" 到控制台。
3.Disposing 和 Dispose Bags
当你订阅一个 Observable，subscribe 方法会返回一个 Disposable 对象。当你不再需要 Observable 时，你应该释放它，以避免内存泄露。你可以通过调用 dispose 方法或者将其添加到一个 DisposeBag 来实现释放。
let disposeBag = DisposeBag()
myObservable.subscribe { event in
print(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 in
print(string)
})
.disposed(by: disposeBag)
subject.onNext("Hello again, RxSwift!")
subscriptionOne.dispose()
let subscriptionTwo = subject
.subscribe(onNext: { string in
print(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 in
print(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 in
Observable.of(value, value * 2) // 将每个元素转换为两个元素的 Observable
}
.subscribe(onNext: { value in
print(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 in
print(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 in
print(value) // 输出：2, 4
})
.disposed(by: disposeBag)在示例中，`filter` 操作符只发出偶数元素。
5.scan
对 Observable 发出的每个元素应用一个聚合函数，返回一个逐步累积的结果序列。
let numbers = Observable.of(1, 2, 3, 4, 5)
numbers
.scan(0) { accumulated, value in
accumulated + value
}
.subscribe(onNext: { value in
print(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 in
accumulated + value
}
.subscribe(onNext: { value in
print(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 in
print(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 in
print(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 in
print(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 in
print(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 in
print(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 in
print(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 in
print(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 in
print(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 in
print("\(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 in
print("\(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 in
print(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 in
print(value) // 输出：0, 1, 2
})
.disposed(by: disposeBag)在示例中，amb操作符选择首先发出元素的 Observable。
20.catchError
捕获 Observable 发出的错误，并返回一个新的 Observable 或执行错误处理逻辑。
enum CustomError: Error {
case error
}
let observable = Observable<Int>.create { observer in
observer.onNext(1)
observer.onNext(2)
observer.onError(CustomError.error)
return Disposables.create()
}
observable
.catchError { error in
return Observable.just(3) // 捕获错误并返回新的 Observable 发出默认值 3
}
.subscribe(onNext: { value in
print(value) // 输出结果: 1, 2, 3
})
.disposed(by: disposeBag)在示例中，Observable 发出 1 和 2，然后遇到错误，使用 catchError 操作符捕获错误并返回一个新的 Observable，新的 Observable 发出默认值 3。
21.retry
在遇到错误时重新订阅 Observable，可以指定最大重试次数。
var count = 0
let observable = Observable<Int>.create { observer in
if 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 in
print(value) // 输出结果: 1
}, onError: { error in
print(error) // 不会产生错误
})
.disposed(by: disposeBag)在示例中，Observable 遇到错误时使用 retry 操作符重新订阅 Observable，最多重试 2 次，所以总共尝试 3 次，最终成功发出值 1。
22.repeatElement
重复发出同一个元素的 Observable。
Observable.repeatElement(1)
.take(3) // 只取前 3 个元素
.subscribe(onNext: { value in
print(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 in
print(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 in
print(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 in
print(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 in
print(value) // 不会输出结果
}, onError: { error in
print(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 in
print(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 in
print(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 in
print(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 in
return Observable.of(value * 2, value * 3) // 将每个元素乘以 2 和 3，并按顺序连接输出
}
.subscribe(onNext: { value in
print(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，否则返回 false
func 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 RxCocoa
class 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 in
return !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 in
if 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 RxCocoa
class 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 in
print("Data: \(data)")
}, onError: { error in
print("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 RxCocoa
class 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) {
// 这里我们模拟网络请求，实际使用中会调用实际的 API
DispatchQueue.main.asyncAfter(deadline: .now() + 1.0) { [weak self] in
self?.searchResults = ["Result 1", "Result 2", "Result 3"]
}
}
}在这个例子中，搜索框的代理方法 searchBar(_:textDidChange:) 被调用时，我们模拟一个网络请求。当网络请求完成时，我们更新 searchResults，这将触发 tableView 的重新加载。
RxSwift：
使用 RxSwift，我们可以将搜索框的文本变化看作一个 Observable，然后我们可以订阅这个 Observable，当它发出新的值时，我们可以请求 API 并更新 UI。这是使用 RxSwift 的版本：
import RxSwift
import RxCocoa
class 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]> in
return self.fetchSearchResults(query) // 获取搜索结果
}
.bind(to: tableView.rx.items(cellIdentifier: "Cell")) { index, model, cell in
cell.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 in
cell.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 RxCocoa
class RxUserInputViewController: UIViewController {
@IBOutlet weak var textField: UITextField!
let disposeBag = DisposeBag()
override func viewDidLoad() {
super.viewDidLoad()
textField.rx.text.orEmpty
.subscribe(onNext: { text in
print("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) { _ in
print("Timer fired!")
}
}
deinit {
timer?.invalidate()
}
}RxSwift：
import RxSwift
import RxCocoa
class RxTimerViewController: UIViewController {
let disposeBag = DisposeBag()
override func viewDidLoad() {
super.viewDidLoad()
Observable<Int>.interval(RxTimeInterval.seconds(1), scheduler: MainScheduler.instance)
.subscribe(onNext: { _ in
print("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 in
completion(data, error)
}.resume()
}
func doSomething() {
fetchData { data, error in
if let error = error {
print("Error: \(error)")
} else if let data = data {
print("Data: \(data)")
}
}
}
}RxSwift：
import RxSwift
import RxCocoa
class 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 in
print("Data: \(data)")
}, onError: { error in
print("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 RxSwift
class 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 in
print("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, _, _ in
if let data = data {
results.append(data)
}
group.leave()
}.resume()
}
group.notify(queue: .main) {
completion(results)
}
}
}RxSwift：
import RxSwift
import RxCocoa
class RxParallelRequestsViewController: UIViewController {
let disposeBag = DisposeBag()
func fetchData(from urls: [URL]) -> Observable<[Data]> {
return Observable.from(urls)
.flatMap { url in
return 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, _, _ in
if let data = data {
let json = try? JSONSerialization.jsonObject(with: data, options: []) as? [String: Any]
completion(json?["title"] as? String)
}
}.resume()
}
}RxSwift：
import RxSwift
import RxCocoa
class 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 in
let 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) { _ in
if !self.events.isEmpty {
print("Processing event: \(self.events.removeFirst())")
} else {
self.timer?.invalidate()
self.timer = nil
}
}
}
}
}RxSwift：
import RxSwift
import RxCocoa
class 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 in
if 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 in
self.user = user
if let user = user {
self.fetchPosts(for: user) { posts in
self.posts = posts
}
}
}
}
}RxSwift：
import RxSwift
import RxCocoa
class 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 in
self.fetchPosts(for: user)
.map { posts in (user, posts) }
}
.subscribe(onNext: { user, posts in
print("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 notification
print("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 notification
print("Notification received!")
})
.disposed(by: disposeBag)
}
}