SubscribeOn 和 ObserveOn 有什么区别

Question

我刚刚发现了 SubscribeOn，这让我想知道我是否应该使用它而不是 ObserveOn。 Google 将我带到这里和<一个href="http://www.jaylee.org/post/2011/07/24/Rx-Using-the-ObserveOn-and-SubscribeOn-operators.aspx" rel="noreferrer">此处，但是两者都没有帮助我理解其中的区别：它看起来非常微妙。

（在我的上下文中，我在非 GUI 线程上“出现”了事件，并且在使用事件数据更新控件之前我需要切换到 GUI 线程）。

Answer 1

通过将 SubscribeOn 视为将线程设置为在链中“向上传递”，并将 ObserveOn 视为设置线程在链中“向下传递”，它帮助我理解了这一点。

下面的代码使用您可以使用的命名线程。

Thread.CurrentThread.Name = "Main";

IScheduler thread1 = new NewThreadScheduler(x => new Thread(x) { Name = "Thread1" });
IScheduler thread2 = new NewThreadScheduler(x => new Thread(x) { Name = "Thread2" });

Observable.Create<int>(o =>
{
    Console.WriteLine("Subscribing on " + Thread.CurrentThread.Name);
    o.OnNext(1);
    return Disposable.Create(() => {});
})
.SubscribeOn(thread1)
.ObserveOn(thread2)
.Subscribe(x => Console.WriteLine("Observing '" + x + "' on " + Thread.CurrentThread.Name));

上面的输出是：

Subscribing on Thread1
在 Thread2 上观察 1

同样有趣的是，当您注释掉 SubscribeOn 行时，输出为：

Subscribing on Main
在 Thread2 上观察 1

因为默认情况下，订阅会“传递”正在运行的线程（此处为 Main）。然后ObserveOn“向下传递”Thread2。

如果您注释掉 ObserveOn 行，则输出为：

Subscribing on Thread1
在 Thread1 上观察 1

因为我们“传递”了 Thread1 上的订阅，并且默认情况下，同一线程被“传递”并用于运行观察。

在 GUI 上下文中，为了保持响应，您希望在 GUI 线程上完成最少的工作量，但需要在 GUI 线程上完成订阅（以同步 UI 更新）。所以你想观察GUI线程。

Answer 2

我不久前遇到了类似的问题，并询问了这个问题。我认为那里的回复（包括评论）会回答你的问题。总结一下：

• 如果您想更新 gui 线程上的控件，请使用 ObserveOn。如果您引用System.Reactive.Windows.Forms.dll，您将获得.ObserveOn(form)，这很方便。
• SubscribeOn 控制实际调用订阅的线程。这里解决的问题是，如果从多个不同线程添加事件处理程序，WinForms 和 WPF 将引发异常。

另外， 这篇文章对于弄清楚ObserveOn和SubscribeOn之间的关系非常有帮助。

Answer 3

区别主要在于 subscribeOn 强制整个管道由另一个线程处理，但使用observerOn，只有在observerOn之后定义的管道中的步骤才会在另一个线程中运行，只有在您设置后才会在另一个线程中执行。

    Observable.just(1) 
              .map ---> executed in io thread
              .filter ---> executed in io thread
              .subscribeOn(Scheduers.io)
              .subscribe()

管道的所有步骤都将在另一个线程中执行。

 Observable.just(1) 
              .map ---> executed in Main thread
              .filter ---> executed in Main thread
              .observerOn(Scheduers.io)
              .map ---> executed in New thread
              .filter ---> executed in New thread
              .subscribe()