如何在特定队列上启动一系列可观察对象？

Posted 2023-03-10

【中文标题】如何在特定队列上启动一系列可观察对象？

【英文标题】：How do I kick off a chain of observable on a particular queue?

【发布时间】：2019-08-23 09:14:21

【问题描述】：

我的理解是 SubscribeOn 告诉 Rx 序列应该从哪个队列开始，并且 observeOn 切换调度程序。

看来不是这样，我似乎缺乏一些基本的了解。

请考虑以下示例。

override func viewDidLoad() 
        super.viewDidLoad()
        // Do any additional setup after loading the view.
        operation1().observeOn(realmReadScheduler)
            .observeOn(realmWriteScheduler)
            .flatMap(self.operation2)
            .observeOn(realmReadScheduler)
            .flatMap(operation3)
            .observeOn(realmSignalScheduler)
            .flatMap(operation4)
            .observeOn(realmConvertScheduler)
            .flatMap(operation5).subscribeOn(realmConvertScheduler)
            .subscribe(onNext: success in
                print(success)
            )


    

打印出来的：

> Operation 1 com.apple.main-thread Operation 2 com.jci.xaap.realm.write
> Operation 3 com.jci.xaap.realm.read Operation 4
> rxswift.queue.DispatchQoS(qosClass:
> Dispatch.DispatchQoS.QoSClass.background, relativePriority: 0)
> Operation 5 com.jci.xaap.realm.convert

我曾期望 Operation1 在 'realmConvertScheduler' 而不是主线程上执行..

如何确保链中的第一个 observable 在我想要的队列上执行？

【参考方案1】：

我怀疑您的operaton1() 写得不正确。请注意，该函数是在主线程上调用的。 Observable 所做的工作是在subscribeOn 告诉它使用的任何线程上执行的（或者如果不存在subscribeOn，则调用subscribe 的线程。

这是一个例子：

let schedulerA = ConcurrentDispatchQueueScheduler(queue: DispatchQueue(label: "schedulerA"))

func operation1() -> Observable<Void> 
    return Observable.create  observer in
        print("do your work on:", Thread.current)
        observer.onNext(())
        observer.onCompleted()
        return Disposables.create()
    


operation1()
    .observeOn(MainScheduler.instance)
    .subscribeOn(schedulerA)
    .subscribe(onNext: 
        print("subscribe(onNext:) on:", Thread.current)
    )

另一种选择是这样写：

func operation1() -> Observable<Void> 
    return Observable.deferred 
        print("do your work on:", Thread.current)
        return .just(())