swift - 如何防御性地创建 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 影响观察 block ，即您处理发出的值的地方，例如

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

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

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

司机

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

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

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

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

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