c# - 吞噬长时间运行的事件聚合器可观察到的异常

Question

我正在使用一个简单的 Subject<object>在 Web 应用程序中实现事件聚合器模式，如下所示:

public class EventAggregator
{
    private readonly ISubject<object, object> _subject = Subject.Synchronize(new Subject<object>());
    public IObservable<T> GetEvent<T>()
    {
        return _subject.AsObservable().OfType<T>();
    }

    public void Publish<TEvent>(TEvent sampleEvent)
    {
        _subject.OnNext(sampleEvent);
    }
}

当运算符(operator)或订阅者抛出异常时，我想记录它然后忽略它并继续发布事件 - 本质上我希望流在某些意外行为的情况下“继续”，因为事件聚合器有一个单例应用程序上下文中的生命周期。

在this question给出的答案是创建一个延迟的可观察对象并使用 Retry() ，但我不想在这里观察到一个冷的。

我想到的解决方案是使用 Catch以及订阅者方法中的 try-catch，我将其封装到一个扩展中:

public static IDisposable SubscribeSwallowExceptions<T>(this IObservable<T> source, Action<T> subscription, Action<Exception> log)
{
    source = source.Catch((Exception e) => { log(e); return source; });

    return source.Subscribe(x =>
    {
        try { subscription(x); }
        catch (Exception e) { log(e); }
    });
}

我知道“包罗万象”的异常处理通常不受欢迎，但在这种情况下，我不确定我还有哪些其他选择，因为我希望即使抛出异常也能保留订阅。我不知道会发生什么类型的异常，因为我还不知道在处理流时会完成什么工作。

这是一种可以接受的处理潜在异常的方法吗？您能预见到任何可能使我对这种方法感到困惑的问题吗？

Answer 1

这是一个有问题的解决方案。一旦订阅者抛出异常，您必须真正假设他们此时已经死在水中(您可以明智地做的很少)并且不再调用它们。使用您的方法，“变坏”的订阅者将继续收到事件，并可能无限期地抛出异常。

此外，您不能真正依赖使用您的扩展方法的订阅者。我会重写你的 GetEvent<T>方法如下(显然用一些日志记录替换我的 Console.WriteLine)。这种方法将仅因不良订阅者而处理订阅，并让其他所有人保持运行。

public IObservable<T> GetEvent<T>()
{       
    return Observable.Create<T>(o =>
    {
        var source = _subject.OfType<T>();
        var m = new SingleAssignmentDisposable();
        m.Disposable = source.Subscribe(
            x => {
                try {
                    o.OnNext(x);                        
                }
                catch(Exception e) {
                    Console.WriteLine(e);
                    m.Dispose();
                }
            },
            e => {
                try {
                    o.OnError(e);                       
                }
                catch(Exception ex) {
                    Console.WriteLine(ex);
                }
                finally {
                    m.Dispose();
                }
            },
            () => {
                try {
                    o.OnCompleted();                        
                }
                catch(Exception e) {
                    Console.WriteLine(e);
                }
                finally {
                    m.Dispose();
                }
            }               
        );

        return m;
    });
}