响应式扩展从线程池线程上调用的OnNext()中吞噬异常

Question

我在.Net 4.5中使用Rx 2。当以下代码运行时，它只是以静默方式退出，而不执行OnCompleted委托或显示任何错误。如果我在ToObservable中使用Scheduler.CurrentThread，它至少会抛出错误并终止程序，此时不执行OnCompleted是有意义的。但是，当它在主线程之外的线程中执行时，这种行为似乎是不合理和不可接受的。我错过什么了吗？

static void Main()
{
     Enumerable.Range(0, 1)
                           .ToObservable(Scheduler.Default)
                           .Subscribe(o => { throw new Exception("blah"); }, () => Console.WriteLine("completed"));
     Thread.Sleep(2000);
 }

编辑：是的，当作为控制台应用程序运行时，无论在哪个线程上执行观察，它都会抛出错误。

然而，当我在NUnit中以如下方式运行此代码作为测试时，它在2秒后（线程睡眠时间）静默退出，没有任何错误或消息（期望"已完成"）。那么，这真的是NUnit造成的问题吗？

[TestFixture]
class Program
{
    [Test]
    public void Test()
    {
        Enumerable.Range(0, 1)
                .ToObservable(Scheduler.Default)
                .Subscribe(
                    o => { throw new Exception("blah"); }, 
                    () => Console.WriteLine("completed"));
        Thread.Sleep(2000);
    }
}

Answer 1

Rx不会捕获观察者抛出的异常。这是一个非常重要的设计原则，之前已经详细讨论过，尽管出于某种原因，它仅作为Rx设计指南中§6.4的脚注。

注意：不保护对Subscribe、Dispose和OnNext方法的调用。这些调用处于monad的边缘。从这些地方调用OnError方法将导致意外行为。

从本质上讲，本指南确保，从观察者的角度来看，OnError将仅由源自可观察对象本身的异常调用，包括对直接参与计算的用户代码的任何调用（而不仅仅是观察结果）。如果不是这种情况，则观察者可能无法区分传递给OnError的异常是其OnNext处理程序中的错误，还是可观察对象中的错误。

但更重要的是，它还确保OnNext处理程序引发的任何异常都不会得到处理。这样可以更容易地调试程序并保护用户数据。

也就是说，当OnNext在池线程上执行时，您可能会观察到不同的行为，这只是您调试经验的结果。尝试启用首次机会异常。

此外，我还将通过将Thread.Sleep更改为Console.ReadKey()来避免竞争条件。

Answer 2

Subscribe块中抛出的异常具有未定义的行为。如果你正在做一些可以抛出的事情，你需要将其封装在Select或SelectMany中（或者只是将代码封装在try-catch中）。