反应式扩展吞下线程池线程上调用的 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 不会捕获观察者抛出的异常。这是一个非常重要的设计原则，之前已经详细讨论过，尽管出于某种原因，它仅作为 §6.4 的脚注包含在 Rx Design Guidelines 中。

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

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

但更重要的是，它还确保 OnNext 处理程序抛出的任何异常都未处理。这样可以更轻松地调试程序并保护用户数据。

话虽如此，当 OnNext 在池线程上执行时，您可能会观察到不同行为的原因仅仅是您的调试经验的结果。尝试启用first-chance exceptions。

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

Answer 2

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