swift - 连接两个都具有 subscribeOn 的可观察序列。如何确保我的可观察对象在线程上运行？

Question

在强制执行 Observable.create 的某部分时代码在特定线程(即后台线程)中运行，我担心使用 subscribeOn运算符可能不起作用，因为有时我可能会将此可观察序列链接到另一个在主线程上运行的可观察序列(使用 observeOn )。

例子

情况是我在主线程上运行了一个 Observable 序列(即一个警告框，要求用户输入是否执行网络调用)。

确保这个 Observable.create 会更好吗？代码通过类似以下内容在后台线程中运行:

Observable<String>.empty()
   .observeOn(ConcurrentMainScheduler(queue: background.queue))
   .concat(myObservableNetworkCall)

为什么不直接使用 subscribeOn ？

问题是如果我使用了subscribeOn (第二个)和之前的可观察对象(警报 Controller )设置为使用 subscribeOn 在后台线程上运行(第一个)，然后是第二个 subscribeOn运算符将不起作用，因为第一次调用更接近源可观察对象:

If you specify multiple subscribeOn() operators, the one closes to the source (the left-most), will be the one used.

• Thomas Nield on RxJava's subscribeOn and observeOn operators (February 2016)

这可能是 RxJava 的行为，但我不确定 Swift 的行为。 Reactivex.io 只是说我们不应该调用 subscribeOn多次。

我倾向于将操作包装到 Observable<Void> 中s 并且它们需要在不同的线程上运行...这就是为什么我要问如何确保 Observable 代码在我指定的线程中运行。 subscribeOn行不通，因为我可以连接可观察对象。

我希望它应该运行的线程被封装在我的 Observable 定义中，而不是在链的更高层。

最佳做法是执行以下操作:

1. 从使用我希望使用的数据类型的 Observable.empty 开始。
2. 使用observeOn强制我希望它在其中运行的线程。
3. 将它与我想要使用的实际 Observable 连接起来。

编辑

1. 我已阅读 subscribeOn和 observeOn关于 reactx.io 的文档。

2. 我熟悉如何使用 subscribeOn 在线程之间切换和 observeOn .

3. 我特别关心的是使用 subscribeOn 的复杂性在连接或组合可观察序列时。

4. 问题是，可观察对象需要专门在一个线程上运行，而且它们不知道它们将在何处以及与谁连接。因为我确切地知道它们应该在哪个线程上运行，所以我更愿意将调度程序定义封装在可观察对象的定义中，而不是在链接序列时。

Answer 1

在函数声明中最好不要指定在哪个线程上调用该函数。例如:

func myObservableNetworkCall() -> Observable<String> {
    return Observable<String>.create { observer in
        // your network code here

        return Disposables.create {
            // Your dispose
        }
    }
}

func otherObservableNetworkCall(s: String) -> Observable<String> {
    return Observable<String>.create { observer in
        // your network code here

        return Disposables.create {
            // Your dispose
        }
    }
}

然后在调度器之间切换:

myObservableNetworkCall()
    .observeOn(ConcurrentMainScheduler(queue: background.queue)) // .background thread, network request, mapping, etc...
    .flatMap { string in
        otherObservableNetworkCall(s: string)
    }
    .observeOn(MainScheduler.instance) // switch to MainScheduler, UI updates
    .subscribe(onNext:{ string in
        // do something 
    })