c# - 对 Observables 使用 ObserveOn 的必要性？

Question

请看下面的代码:

 var obs = Observable.Start(() => LongRunningMethodToRetrieveData());

 obs.Subscribe(x => MethodThatMustBeOnUIThread(x));

如果我确定这两行代码是在UI线程上执行的，是否需要先调用ObserveOn(SynchronizationContext.Current)？或者类似地，我是否需要检查 MethodThatMustBeOnUIThread 中的 InvokeRequired？

基本上，我是否可以通过这两行代码保证在创建订阅的线程上调用“OnNext”？

谢谢。

编辑:好吧，我在调试器中尝试过，“MethodThatMustBeOnUIThread”确实是从后台线程调用的。为什么是这样？我现在的假设是，观察默认发生在运行异步方法的线程上。

Answer 1

您需要熟悉各种 Rx 方法使用的默认调度程序。

Observable.Generate 之类的方法确实在订阅可观察对象的线程上运行。

另一方面，Observable.Start 方法的目的是在订阅可观察对象时异步调用 lamdba 操作。如果它发生在 UI 线程上，它就不是异步的。所以在这种情况下，它使用 ThreadPool 调度程序。

这可以通过使用 Reflector.NET 看到:

public static IObservable<TSource> Start<TSource>(Func<TSource> function)
{
    if (function == null)
    {
        throw new ArgumentNullException("function");
    }
    return function.ToAsync<TSource>()();
}

public static Func<IObservable<TResult>> ToAsync<TResult>(
    this Func<TResult> function)
{
    if (function == null)
    {
        throw new ArgumentNullException("function");
    }
    return function.ToAsync<TResult>(Scheduler.ThreadPool);
}

因此，如果您希望订阅在 UI 线程上运行，那么在调用 Subscribe 之前，了解所使用的调度程序必须使用 ObserveOn 的形式。

此外，由于您使用的是 Rx，我不会使用 InvokeRequired - 这只是混合了异步编码模型。 Rx 拥有您需要的一切，可以很好地处理线程。