异步调用事件并捕获异常，而无需阻止其他任务运行

Question

给定以下代码：

public delegate Task AsyncEventHandler<in TEventArgs>(object sender, TEventArgs eventArgs);

public static Task InvokeAsync(this AsyncEventHandler eventHandler, object sender, EventArgs eventArgs)
{
    if (eventHandler == null) return Task.CompletedTask;

    var delegates = eventHandler.GetInvocationList().Cast<AsyncEventHandler>();
    var tasks = delegates.Select(it => it.Invoke(sender, eventArgs));

    return Task.WhenAll(tasks);
}

我有一个测试功能，其中有缺陷的处理程序应抛出异常，并且WorkingHanlder应该运行 - 目前只有第一个FARKYHANDLER1被调用，而没有其他事件处理程序。

private class NonGenericNotifier
{
    public event AsyncEventHandler SomethingHappened;
    public Task OnSomethingHappening() => SomethingHappened.InvokeAsync(this, EventArgs.Empty);
}

public async Task Multiple_Exceptions_That_Occur_During_Event_Handling_Should_Be_Propagated()
{
    var isHandler1Called = false;
    var isHandler2Called = false;
    var isWorkingHandlerCalled = false;

    var notifier = new NonGenericNotifier();

    Task FaultyHandler1(object sender, EventArgs eventArgs)
    {
        isHandler1Called = true;
        throw new InvalidOperationException();
    }

    Task FaultyHandler2(object sender, EventArgs eventArgs)
    {
        isHandler2Called = true;
        throw new InvalidOperationException();
    }

    Task WorkingHandler(object sender, EventArgs eventArgs)
    {
        isWorkingHandlerCalled = true;
        return Task.CompletedTask;
    }

    notifier.SomethingHappened += FaultyHandler1;
    notifier.SomethingHappened += FaultyHandler2;
    notifier.SomethingHappened += WorkingHandler;

    await Should.ThrowAsync<InvalidOperationException>(async () => await notifier.OnSomethingHappening());
    isHandler1Called.ShouldBe(true);
    isHandler2Called.ShouldBe(true);
    isWorkingHandlerCalled.ShouldBe(true);
}

假设可以提出一个例外，我相信这应该是gentregateException包含每个任务的例外，最重要的是，上面的InvokeAsync方法应在第一个例外情况下保释。

我已经开始在InvoKeAsync扩展方法中创建lt; exception＆gt;，并包装每个it =＆gt; IT.in.invoke（发件人，EventArgs）带有try/Catch构造，并且在CANGE中将异常添加到异常列表中。

但是，我对如何整理此异常列表感到迷失，然后以gencregateException发送。

更新（修复？）

感谢Artur将我指向正确的方向。我将InvokeAsync扩展方法更改为下面的扩展方法，它 works - 不再停止第一个任务。我们从var tasks = delegates.Select（it =＆gt; it.invoke（sender，eventArgs））; 使用代码在这里：

public static Task InvokeAsync(this AsyncEventHandler eventHandler, object sender, EventArgs eventArgs)
{
    if (eventHandler == null) return Task.CompletedTask;

    var delegates = eventHandler.GetInvocationList().Cast<AsyncEventHandler>();
    var tasks = delegates.Select(async it => await it.Invoke(sender, eventArgs));

    return Task.WhenAll(tasks).WithAggregatedExceptions();
}

static Task WithAggregatedExceptions(this Task task)
{
    // using AggregateException.Flatten as a bonus
    return task.ContinueWith(
        continuationFunction: priorTask =>
            priorTask.IsFaulted &&
            priorTask.Exception is AggregateException ex && (ex.InnerExceptions.Count > 1 || ex.InnerException is AggregateException) ? Task.FromException(ex.Flatten()) : priorTask,
        cancellationToken: CancellationToken.None,
        TaskContinuationOptions.ExecuteSynchronously,
        scheduler: TaskScheduler.Default).Unwrap();
}

我的问题是此事件的订阅者，编写我无法控制的同步处理程序 -停止连接连接到同一事件的另一个事件处理程序（同步和异步）。

我也感谢这是task.thenall的设计函数，如果您要混合异步和非Async处理程序...如果有一个理由不带有没有的异步函数中的同步代码等待Task.yield（）就是这样。

问题

我们可以说包裹demeg.select（async it =＆gt;等待it.invoke（sender，evensargs）用异步/等待允许同步方法运行，并且在最差的（？）中，将异步方法（与嵌套异步/等待函数呼叫相同），所以实际上是一个非问题

吗

？赏金寻求有关如何实施的权威指导，一个答案（为讨论做出贡献非常感谢）说避免异步事件，但在其他地方，例如Discord c＃客户端，他们接受了异步事件（带有超时包装器等）。

Answer 1

task.whenall将在重新简介其参数时调用所有处理程序。它将一次调用一个，然后异步等待所有任务完成。

您看到第一个例外停止的原因是因为在重新化过程中抛出了例外。异步（返回任务）功能是正常的，可以在返回的任务上放置任何异常。异步功能直接抛出异常是异常的。

因此，这是有问题的代码：

Task FaultyHandler1(object sender, EventArgs eventArgs)
{
  isHandler1Called = true;
  throw new InvalidOperationException();
}

其中之一是正确的：

async Task FaultyHandler1(object sender, EventArgs eventArgs)
{
  isHandler1Called = true;
  throw new InvalidOperationException();
}

Task FaultyHandler1(object sender, EventArgs eventArgs)
{
  isHandler1Called = true;
  return Task.FromException(new InvalidOperationException());
}

您正在看到奇怪的行为，因为异步处理程序的行为不端（通过抛出同步异常）。

现在，如果您想要允许异步的处理程序，那么您 can can 都可以使用显式尝试/catch >或额外的async/等待：

var tasks = delegates.Select(it => try { return it.Invoke(sender, eventArgs); } catch (Exception ex) { return Task.FromException(ex); });

// async/await is necessary to handle misbehaving asynchronous handlers that throw synchronous exceptions
var tasks = delegates.Select(async it => await it.Invoke(sender, eventArgs));

如果您保留async/等待等待方法，请进行评论，因为这样的编码结构通常被认为是虚假的，并且可以被未来的维护者删除。

withAggregatedExceptions看起来不错，但是如果需要，可以简化它：

static async Task WithAggregatedExceptions(this Task task)
{
  try { await task; }
  catch { throw task.Exception; }
}

我的问题是此事件的订户，编写我无法控制的同步处理程序 - 这将阻止运行在同一事件上的其他事件处理程序（Sync和async）。

好吧，是的。 task.whenall同步重新安排其任务收集，该任务一次调用所有处理程序。

如果要同时允许同步处理程序以及异步处理程序，则您 can can 将调用包装在task.run中：

var tasks = delegates.Select(it => Task.Run(() => it.Invoke(sender, eventArgs)));

我们可以说包装demelect.select（async it =＆gt;等待它。与嵌套异步/等待函数调用相同），所以实际上是一个非问题吗？

的 async /等待异步处理程序的 出现 ，我说它有被删除的危险（除非评论）“允许同步方法运行”；而不是按预期将它们放在返回的任务上。

是否有任何副作用？

并不真地。如果您确实使用task.run方法，则将所有处理程序都在线程池线程上调用并可能同时运行，这可能令人惊讶。

一个答案（非常感谢为讨论做出贡献）说避免异步事件，但是在其他地方，他们接受了异步事件（带有超时包装器等）等其他地方。

我100％同意该答案。

我的思考是：

观察者模式是将观察者通知状态变化的一种方式。观察者模式明显适合OOP中的“事件”：任何数量的观察者都可以订阅状态变更通知。这就是C＃事件的图案：通知订阅者的事物。没有信息可以“回流”的机制。尽管语言允许 c＃具有返回值的事件，但无论如何，它不是天然。异常情况也会发生相同的限制（可以认为是一种返回）：标准处理程序？invoke模式开始断开（在第一个例外情况下停止调用等）。

一旦您有信息“向后流”（包括需要处理异常或需要等待所有处理程序要完成），您将不再处于观察者模式中，并且您不再是在C＃活动的快乐之路。

通常，我发现这种类型的大多数“事件”通常与策略或访客模式有关，而不是观察者。策略和访客都不适合C＃事件，尽管经常（可悲的是）以这种方式实施。我认为这是一个常见的设计错误（对于所有OOP语言）。

Answer 2

我认为，以异步方式使用C＃事件的设计并不强大，并且它总是以一种不受控制的方式行事。有更好的技术可以使事件处理强大。

最好的技术之一是TPL DataFlows（ https://learn.microsoft.com/en-us/dotnet/standard/parallelal-programming/dataflow-task-parellell-library-library ）。该库允许您以非常控制的方式进行编程流处理，它可以帮助您处理任务调度程序等。成功应用此过程后，问题中的所有问题都将被解决。

显然还有其他替代方案，但是我显然会使用C＃事件来弃权。