RxSwift: SubscribeOn 与 ObserverO
摘要
使用 subscribeOn
我们用 subscribeOn 来决定数据序列的构建函数在哪个 Scheduler 上运行。以上例子中，由于获取 Data 需要花很长的时间，所以用 subscribeOn 切换到 后台 Scheduler 来获取 Data。这样可以避免主线程被阻塞。
使用 observeOn
我们用 observeOn 来决定在哪个 Scheduler 监听这个数据序列。以上例子中，通过使用 observeOn 方法切换到主线程来监听并且处理结果。
一个比较典型的例子就是，在后台发起网络请求，然后解析数据，最后在主线程刷新页面。你就可以先用 subscribeOn 切到后台去发送请求并解析数据，最后用 observeOn 切换到主线程更新页面
结构
在讨论 SubscribeOn 与 ObserverOn 的区别之前，先来讨论一下 Observable 从产生事件到响应事件的结构，可以分为三部分：
- 产生事件：
Observable<Int>.create(...) - 操作事件：
map(),filter(),zip()等操作符的功能 - 响应事件：
subscribe(onNext: ..., onCompleted: ..., onError: ...)
Observable<Int>.create { observer in
1 ..... 产生事件
}
.map { element -> String in
2 ...... 操作事件
}
.subscribe(onNext: { element in
3 ...... 响应事件
}).disposed(by: disposeBag)
1. 默认情况
在没有使用 SubscribeOn 和 ObserverOn 明确指定在哪个队列中调度时，subscribe()在哪个队列被调用，「产生事件」和「响应事件」的方法就在哪个队列中执行
在「子线程」中调用 subscribe()
func invokeSubscribeInGlobalQueue() {
let observableInt = Observable<Int>.create { observer in
print("产生事件 -> \(self.currentQueueName() ?? "queue")")
observer.onNext(1)
return Disposables.create()
}
.map { element -> Int in
print("操作事件1 -> \(self.currentQueueName() ?? "queue")")
return element + 1
}
.map { element -> Int in
print("操作事件2 -> \(self.currentQueueName() ?? "queue")")
return element + 1
}
DispatchQueue.global().async {
print("调用 subscribe() 的队列: \(self.currentQueueName() ?? "queue")")
observableInt.subscribe(onNext: { element in
print("响应事件 -> \(self.currentQueueName() ?? "queue"), element -> \(element)\n\n")
}).disposed(by: self.disposeBag)
}
}
输出结果：
调用 subscribe() 的队列: com.apple.root.default-qos
产生事件 -> com.apple.root.default-qos
操作事件1 -> com.apple.root.default-qos
操作事件2 -> com.apple.root.default-qos
响应事件 -> com.apple.root.default-qos, element -> 3
在「主线程」中调用 subscribe()
func invokeSubscribeInMainQueue() {
let observableInt = Observable<Int>.create { observer in
print("产生事件 -> \(self.currentQueueName() ?? "queue")")
observer.onNext(1)
return Disposables.create()
}
.map { element -> Int in
print("操作事件1 -> \(self.currentQueueName() ?? "queue")")
return element + 1
}
.map { element -> Int in
print("操作事件2 -> \(self.currentQueueName() ?? "queue")")
return element + 1
}
print("调用 subscribe() 的队列: \(self.currentQueueName() ?? "queue")")
observableInt.subscribe(onNext: { element in
print("响应事件 -> \(self.currentQueueName() ?? "queue"), element -> \(element)\n\n")
}).disposed(by: self.disposeBag)
}
输出结果：
调用 subscribe() 的队列: com.apple.main-thread
产生事件 -> com.apple.main-thread
操作事件1 -> com.apple.main-thread
操作事件2 -> com.apple.main-thread
响应事件 -> com.apple.main-thread, element -> 3
2. 只使用 SubscribeOn 指定「产生事件」执行的队列
SubscribeOn 是向上和向下作用的，只使用subscribeOn指定执行的队列之后，「产生事件」、「操作事件」、「响应事件」都将在指定的队列中执行。
func subscribeOn() {
Observable<Int>.create { observer in
print("产生事件 -> \(self.currentQueueName() ?? "queue")")
observer.onNext(1)
return Disposables.create()
}
.map { element -> Int in
print("操作事件1 -> \(self.currentQueueName() ?? "queue")")
return element + 1
}
.subscribeOn(SerialDispatchQueueScheduler(qos: .background))
.map { element -> Int in
print("操作事件2 -> \(self.currentQueueName() ?? "queue")")
return element + 1
}
.subscribe(onNext: { element in
print("响应事件 -> \(self.currentQueueName() ?? "queue"), element -> \(element)\n\n")
}).disposed(by: self.disposeBag)
}
输出结果：
产生事件 -> rx.global_dispatch_queue.serial
操作事件1 -> rx.global_dispatch_queue.serial
操作事件2 -> rx.global_dispatch_queue.serial
响应事件 -> rx.global_dispatch_queue.serial, element -> 3
当出现多个 subscribeOn的时候，只有第一个能起到作用。
func subscribeOn() {
Observable<Int>.create { observer in
print("产生事件 -> \(self.currentQueueName() ?? "queue")")
observer.onNext(1)
return Disposables.create()
}
.map { element -> Int in
print("操作事件1 -> \(self.currentQueueName() ?? "queue")")
return element + 1
}
.subscribeOn(SerialDispatchQueueScheduler(qos: .background))
.map { element -> Int in
print("操作事件2 -> \(self.currentQueueName() ?? "queue")")
return element + 1
}
.subscribeOn(MainScheduler.instance)
.subscribe(onNext: { element in
print("响应事件 -> \(self.currentQueueName() ?? "queue"), element -> \(element)\n\n")
}).disposed(by: self.disposeBag)
}
输出结果：
产生事件 -> rx.global_dispatch_queue.serial
操作事件1 -> rx.global_dispatch_queue.serial
操作事件2 -> rx.global_dispatch_queue.serial
响应事件 -> rx.global_dispatch_queue.serial, element -> 3
3. 只使用 ObserverOn 指定「响应事件」执行的队列
observarOn是向下作用的, observarOn 可以指定其后面的「操作事件」和「响应事件」执行的队列，可以使用多个observarOn来改变不同的「操作事件」执行的队列
func observerOn() {
Observable<Int>.create { observer in
print("产生事件 -> \(self.currentQueueName() ?? "queue")")
observer.onNext(1)
return Disposables.create()
}
.map { element -> Int in
print("操作事件1 -> \(self.currentQueueName() ?? "queue")")
return element + 1
}
.observeOn(SerialDispatchQueueScheduler(qos: .background))
.map { element -> Int in
print("操作事件2 -> \(self.currentQueueName() ?? "queue")")
return element + 1
}
.observeOn(MainScheduler.instance)
.map { element -> Int in
print("操作事件3 -> \(self.currentQueueName() ?? "queue")")
return element + 1
}
.observeOn(SerialDispatchQueueScheduler(qos: .background))
.map { element -> Int in
print("操作事件4 -> \(self.currentQueueName() ?? "queue")")
return element + 1
}
.observeOn(MainScheduler.instance)
.observeOn(SerialDispatchQueueScheduler(qos: .background))
.subscribe(onNext: { element in
print("响应事件 -> \(self.currentQueueName() ?? "queue"), element -> \(element) \n\n")
}).disposed(by: disposeBag)
}
输出结果：
产生事件 -> com.apple.main-thread
操作事件1 -> com.apple.main-thread
操作事件2 -> rx.global_dispatch_queue.serial
操作事件3 -> com.apple.main-thread
操作事件4 -> rx.global_dispatch_queue.serial
响应事件 -> rx.global_dispatch_queue.serial, element -> 5
4. 同时使用 SubscribeOn 和 ObserverOn
同时使用 SubscribeOn 和 ObserverOn时。「产生事件」和 ObserveOn 之前的 「操作事件」将会在 SubscribeOn 指定的队列中执行。ObserveOn 之后的 「操作事件」和 「响应事件」将会在 ObserveOn 指定的队列中执行。
SubscribeOn和ObserveOn在代码中出现的顺序不会影响上述的逻辑
func subscribeOnAndObserveOn() {
Observable<Int>.create { observer in
print("产生事件 -> \(self.currentQueueName() ?? "queue")")
observer.onNext(1)
return Disposables.create()
}
.map { element -> Int in
print("操作事件1 -> \(self.currentQueueName() ?? "queue")")
return element + 1
}
.observeOn(MainScheduler.instance)
.subscribeOn(SerialDispatchQueueScheduler(qos: .background))
.map { element -> Int in
print("操作事件2 -> \(self.currentQueueName() ?? "queue")")
return element + 1
}
.subscribe(onNext: { element in
print("响应事件 -> \(self.currentQueueName() ?? "queue"), element -> \(element)\n\n")
}).disposed(by: self.disposeBag)
}
输出结果：
产生事件 -> rx.global_dispatch_queue.serial
操作事件1 -> rx.global_dispatch_queue.serial
操作事件2 -> com.apple.main-thread
响应事件 -> com.apple.main-thread, element -> 3
