RxSwift: SubscribeOn 与 ObserverO

摘要

---

使用 subscribeOn

我们用 subscribeOn 来决定数据序列的构建函数在哪个 Scheduler 上运行。以上例子中，由于获取 Data 需要花很长的时间，所以用 subscribeOn 切换到 后台 Scheduler 来获取 Data。这样可以避免主线程被阻塞。

使用 observeOn

我们用 observeOn 来决定在哪个 Scheduler 监听这个数据序列。以上例子中，通过使用 observeOn 方法切换到主线程来监听并且处理结果。

一个比较典型的例子就是，在后台发起网络请求，然后解析数据，最后在主线程刷新页面。你就可以先用 subscribeOn 切到后台去发送请求并解析数据，最后用 observeOn 切换到主线程更新页面

结构

---

在讨论 SubscribeOn 与 ObserverOn 的区别之前，先来讨论一下 Observable 从产生事件到响应事件的结构，可以分为三部分：

1. 产生事件： Observable<Int>.create(...)
2. 操作事件： map(),filter(), zip() 等操作符的功能
3. 响应事件：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