如何防御性地创建 Rx Observables 并避免竞争条件？

Question

我正在寻找一个奇怪的边缘情况，其中目录中的文件列表没有显示 0...2 个文件的结果，但对于 3...n 个文件工作正常。

原来可观察到的原始序列运行良好。但是我在一个订阅者中使用了PublishSubject 来传递更改的效果。据报道，所有这些都发生在主队列上，但似乎PublishSubject 在它拥有订阅者之前就获得了馈送值。 （由于没有重播，订阅者不会知道。）

所以所有组件的设置（来源——中继订阅者——消费订阅者）似乎都引入了时间作为一个问题。

奇怪的观察：

• 如果我将原来的Observable 变成Driver，一切正常。
• 如果我在原始链中的某处插入.observeOn(MainScheduler.instance)（在中继订户之前），一切正常。尽管我可以通过在映射操作中使用断点看到映射已经发生在 com.apple.main-queue (serial) 上。
• 如果我在来源或订阅者的代码中使用.subscribeOn(MainScheduler.instance)，我仍然会遇到问题。 （可能是因为一开始它只影响中继订阅者，后来为时已晚。）

现在我不明白如何防御性地处理此类问题。

看来PublishSubject 可能不适合这种情况。但是为什么在主队列上观察会改善这种情况呢？

您何时应该（防御性地）指定在主队列上运行的可观察序列在创建/生产时？（同样，这可能是一个实用的修复，但它只是似乎意外地解决了问题。）

或者，换句话说，你应该在什么时候假设消费者的代码没有及时发生，即在订阅设置时？

我无法判断将事件从输入序列中继到PublishSubject 会导致麻烦。这是不可感知的。让我很困惑如何避免这样的错误。

Answer 1

你需要了解observeOn和subscribeOn的区别以及Driver序列的属性。

observeOn 影响观察块，即您处理发射值的位置，例如

someObservable
.subscribe(onNext: { /* this is an example of observation block */ })

subscribeOn 影响订阅块中的调度程序，即 产生值的地方，例如

Observable<Int>.create { /* this is an example of subscription block */ }

驱动程序

驱动程序是一种特殊类型的可观察序列，具有一些属性。

• 驱动程序永远不会失败，即没有 onError 事件
• 驱动程序将.observeOn(MainScheduler.instance) 添加到序列中
• 驱动程序使用.share(replay: 1, scope: .whileConnected) 运算符共享序列

考虑到最后两个，你可以从 observable 做你自己的驱动程序：

someObservable
.observeOn(MainScheduler.instance)
.share(replay: 1, scope: .whileConnected)

这就是你的问题可能想要做的。