c# - Reactive Extensions 使 .net 进程崩溃

Question

我开始在我的项目中使用响应式(Reactive)扩展并发现以下问题。我的类(class)中有一个主题，我需要所有客户端观察者在不同的线程中处理 OnNext 消息。

private Subject<IEnumerable<int>> _multiResponse =
    new Subject<IEnumerable<int>>();

public IObservable<IEnumerable<int>>> MultiResponse
{
     get
     {
         return _response.ObserveOn(TaskPoolScheduler.Default);
     }
}

如果客户端在 OnNext 事件中抛出异常，则由于 Task 中未观察到的异常，所有进程都会崩溃。堆栈跟踪:

Scheduler.cs 中的 ProcessResponses(IEnumerable 响应):第 149 行
在 System.Reactive.AnonymousSafeObserver'1.OnNext(T 值)
在 System.Reactive.ScheduledObserver'1.Dispatch(ICancelable 取消)
在 System.Reactive.Concurrency.Scheduler.b_32(Action'1 a, ICancelable c)
在 System.Reactive.Concurrency.TaskPoolScheduler.<>c_DisplayClass7'1.b__6()
在 System.Threading.Tasks.Task.Execute()

这是预期的行为还是 RX 中的错误以及如何处理？

Answer 1

是的，现在注意……这是关于未观察到的异常。

我认为你异步做事的事实应该无关紧要。此同步代码抛出 Exception ，正如我所期望的:

Observable.Return(1).Subscribe(
    _ => { throw new Exception(); });

但是，我希望找到以下代码会抛出 Exception (从而解释了你所看到的)但起初我很惊讶地发现它没有!

Observable.Return(1).ObserveOn(TaskPoolScheduler.Default).Subscribe(
    _ => { throw new Exception(); });

(编辑 - 见下文 - 这是 .NET 4.5 行为)

当然切换到 ThreadPoolScheduler给出预期 Exception :

Observable.Return(1).ObserveOn(ThreadPoolScheduler.Instance).Subscribe(
    _ => { throw new Exception(); });

编辑:

对 - 所以我不再偷懒并开始阅读源代码。事实证明，行为取决于您是否使用 .NET 4.5。

在 .NET 4.5 上，任务池的默认行为是不会让未观察到的异常导致进程中断。在 .NET 4.0 上确实如此。因此，Rx 遵循平台默认设置，并按照 .NET 4.0 上的其他调度程序运行，但在 .NET 4.5 上默默吞下异常。

因此，假设您使用的是 .NET 4.0，您将看到预期的结果 - 您有一个未观察到的异常。

另请注意，由于 .NET 4.5 是就地升级，因此这是一个重大更改。重要的是机器上安装了Framework， 不是 在项目属性中选择了什么框架。

相关的源代码在这里 - 值得一读，评论很有启发性:

http://rx.codeplex.com/SourceControl/latest#Rx.NET/Source/System.Reactive.PlatformServices/Reactive/Concurrency/TaskPoolScheduler.cs

以下是 MSDN 上关于此行为更改的一些评论(请参阅备注)，其中还展示了如何在 .NET 4.5 安装上强加 .NET 4.0 行为:

http://msdn.microsoft.com/en-us/library/jj160346(v=vs.110).aspx

我的建议是不要依赖这种行为，而是保护您的 OnNext 处理程序 带有异常处理程序。明确地说，我的意思是订阅者不应该抛出。 Rx 设计指南明确指出不应保护对 OnNext、OnError 和 OnCompleted 的调用。 (请参阅 http://go.microsoft.com/fwlink/?LinkID=205219 处的第 6.4 节)