.NET中使用Rx出现非预期行为 观察者异常致源调用崩溃

该现象完全符合Rx.NET的官方设计约定。
Rx的核心执行契约明确：OnNext消息默认沿订阅链同步传播，观察者侧未被捕获的异常会直接沿调用栈向上回溯，最终抛到最初触发OnNext调用的源头代码。当前实现中，WorkService.DoWork调用observableWork.OnNext时，会同步执行所有订阅者注册的处理逻辑，异步任务抛出的未捕获异常自然会沿调用栈传回DoWork方法，触发崩溃。
额外说明：当前使用的Concat()操作符本身带有终止语义——只要序列中任意一个内部流抛出未捕获异常，整个订阅会直接终止，后续传入的工作项再也不会被处理。

• 使用了无参Subscribe()重载，该重载没有注册异常处理回调，序列中出现的未捕获异常会被直接抛出到触发消息的调用栈
• 原生Subject没有做执行隔离，所有订阅者的逻辑默认直接运行在触发OnNext的同步上下文中，没有做异常边界切割
• 异步处理逻辑内部没有做异常兜底，异常直接抛出到了Rx序列管道中

核心原则是在观察者侧完成所有异常的捕获和处理，绝对不能让异常泄漏到Rx管道层面，更不能穿透回消息触发方。

保留顺序排队语义的修复（匹配当前Concat的行为） #

如果业务要求工作项严格按触发顺序排队执行异步处理，只需要在异步逻辑内部加异常捕获，同时给订阅加兜底异常回调即可，参考代码：

this.reactiveService.WorkBeingDone()
    .Select(workPayload => Observable.FromAsync(async () =>
    {
        try
        {
            // 编写对应workPayload的异步处理逻辑
        }
        catch (Exception ex)
        {
            // 自行处理异常：记录日志、执行降级逻辑、存入死信队列均可
            // 严禁异常抛出到try块外部
        }
    }))
    .Concat()
    .Subscribe(
        onNext: _ => { },
        onError: ex =>
        {
            // 极端场景兜底：如果走到这里说明整个序列已经因异常终止
            // 在这里添加日志、重建订阅的逻辑即可
        }
    );

其他场景可选优化 #

• 不需要严格顺序执行时，可以将Concat()替换为带并发度参数的Merge(maxConcurrency)，让多个异步任务并行执行，同样要在内部逻辑中捕获所有异常
• 希望彻底隔离触发方和观察者的执行上下文时，可以在管道中追加.ObserveOn(TaskPoolScheduler.Default)，将观察者的逻辑调度到线程池执行，从执行层面切断异常沿调用栈回传的路径。注意该操作不能替代异常捕获，漏处理的异常依然会终止订阅
• 希望异常发生后序列不终止，可以在内部Observable层追加.Catch(Observable.Empty<Unit>())，将异常吞噬后返回空序列，保证主管道持续运行

生产环境禁止直接使用无参Subscribe()重载，所有订阅必须显式传入异常处理回调，避免出现未处理异常导致进程崩溃、业务逻辑中断的问题。

内容的提问来源于stack exchange，提问作者Julien Debache