Rx Throttle(...).ObserveOn(调度程序)和Throttle(...,调度程序)之间的区别

Question

我有以下代码:

IDisposable subscription = myObservable.Throttle(TimeSpan.FromMilliseconds(50), RxApp.MainThreadScheduler)
                                       .Subscribe(_ => UpdateUi());

正如所料,UpdateUi()将始终在主线程上执行.当我将代码更改为

IDisposable subscription = myObservable.Throttle(TimeSpan.FromMilliseconds(50))
                                       .ObserveOn(RxApp.MainThreadScheduler)
                                       .Subscribe(_ => UpdateUi());

UpdateUI() 将在后台线程中执行.

为什么不Throttle(...).ObserveOn(scheduler)等同Throttle(..., scheduler)？

Answer 1

在您给出的代码的两个示例中，始终UpdateUi 会在 指定的调度程序上调用RxApp.MainThreadScheduler。我可以肯定地说这一点，因为它ObserveOn是一个装饰器，可确保OnNext在指定的调度程序上调用订阅者的处理程序。请参阅此处进行深入分析。

这么说来，这有点令人费解。要么RxApp.MainThreadScheduler没有引用正确的调度程序调度程序，要么UpdateUi正在从调度程序线程转换。前者并不是史无前例的 - 请参阅https://github.com/reactiveui/ReactiveUI/issues/768，其他人也遇到过这种情况。我不知道那种情况下出了什么问题。也许 @PaulBetts 可以参与进来，或者您可以在https://github.com/reactiveui/上提出问题。不管怎样，我会在这里仔细检查你的假设，因为我希望这是一个经过充分测试的领域。你有完整的重现吗？

至于你的具体问题，Throttle(...).ObserveOn(scheduler)和之间的区别Throttle(..., scheduler)之间的区别如下：

在第一种情况下，当Throttle指定没有调度程序时，它将使用默认的平台调度程序来引入运行其计时器所需的并发性 - 在 WPF 上，这将使用线程池线程。因此，所有限制都将在后台线程上完成，并且由于以下原因ObserveOn仅发布的事件将传递给指定调度程序上的订阅者。

在这种情况下Throttle，限制是在该调度程序上完成的 - 抑制的事件和释放的事件都将在该调度程序上进行管理，并且订阅者也将在同一调度程序上被调用。

所以无论哪种方式，都会UpdateUi被调用RxApp.MainThreadScheduler.

在大多数情况下，最好在调度程序上限制 ui 事件，因为如果只有一小部分事件要通过限制，那么在后台线程上运行单独的计时器并为上下文切换付费通常成本更高。

因此，为了检查您是否遇到了问题RxApp.MainThreadScheduler，我会尝试指定调度程序或SynchronizationContext通过其他方式明确指定。如何执行此操作将取决于您所在的平台 -ObserveOnDispatcher()希望可用，或者使用合适的ObserveOn重载。Rx如果导入了正确的库，则可以选择控件、同步上下文和调度程序。

Answer 2

经过一番调查后,我认为这是由于运行时使用的Rx版本比我预期的不同(我为第三方应用程序开发了一个插件).

我不知道为什么,但似乎默认RxApp.MainThreadScheduler无法正确初始化.默认实例是WaitForDispatcherScheduler(源).此类中的所有函数都依赖于attemptToCreateScheduler:

    IScheduler attemptToCreateScheduler()
    {
        if (_innerScheduler != null) return _innerScheduler;
        try {
            _innerScheduler = _schedulerFactory();
            return _innerScheduler;
        } catch (Exception) {
            // NB: Dispatcher's not ready yet. Keep using CurrentThread
            return CurrentThreadScheduler.Instance;
        }
    }

在我的情况下似乎发生的是_schedulerFactory()投掷,导致CurrentThreadScheduler.Instance返回.

通过手动初始化RxApp.MainThreadSchedulerto new SynchronizationContextScheduler(SynchronizationContext.Current)行为是预期的.