数据块中的异常处理

Posted 2023-04-12

【中文标题】数据块中的异常处理

【英文标题】：Exception Handling In DataBlocks

【发布时间】：2021-01-28 20:30:50

【问题描述】：

我正在尝试了解 TPL 中的异常处理。

以下代码似乎吞下了异常：

var processor = new ActionBlock<int>((id) => SomeAction(id), new ExecutionDataflowBlockOptions  ... );


async Task SomeAction(int merchantID)

    //Exception producing code
    ...

并且收听TaskScheduler.UnobservedTaskException 事件也不会收到任何东西。

那么，这是否意味着动作块在运行动作时会自己进行 try-catch？

在某处有这方面的官方文档吗？

【参考方案1】：

更新

DataFlow 块的异常处理行为在Exception Handling in TPL DataFlow Networks中解释

**原创

此代码不会吞下异常。如果您使用await processor.Completion 等待该块完成，您将收到异常。如果您在调用Complete() 之前使用循环将消息发送到块，您也需要一种方法来停止循环。一种方法是使用CancellationTokenSource 并在出现异常时发出信号：

void SomeAction(int i,CancellationTokenSource cts)

    try
    
        ...
    
    catch(Exception exc)
    
        //Log the error then
        cts.Cancel();
       //Optionally throw
    

发帖代码不用改那么多，只需要检查是否

var cts=new CancellationTokenSource();
var token=cts.Token;
var dopOptions=new new ExecutionDataflowBlockOptions  
                           MaxDegreeOfParallelism=10,
                           CancellationToken=token
;
var block= new ActioBlock<int>(i=>SomeAction(i,cts),dopOptions);

while(!token.IsCancellationRequested && someCondition)

    block.Post(...);

block.Complete();
await block.Completion;

当动作抛出时，令牌发出信号并且块结束。如果异常被操作重新抛出，它也会被await block.Completion 重新抛出。

如果这看起来令人费解，那是因为这在某种程度上是块的边缘情况。 DataFlow 用于创建管道或块网络。

一般情况

Dataflow 这个名字很重要。 您不必使用相互调用的方法来构建程序，而是使用相互传递消息的处理块。没有接收结果和异常的父方法。块的管道无限期地保持活动状态以接收和处理消息，直到某个外部控制器告诉它停止，例如通过在头块上调用 Complete 或发出信号传递给每个块的 CancellationToken。

块不应允许发生未处理的异常，即使它是独立的 ActionBlock。如您所见，除非您已经调用了Complete() 和await Completion，否则您不会得到异常。

当块内部发生未处理的异常时，块进入故障状态。该状态传播到与PropagateCompletion 选项链接的所有下游块。 上游块不受影响，这意味着它们可以继续工作，将消息存储在其输出缓冲区中，直到进程耗尽内存，或者因为没有收到来自块的响应而死锁。

正确的故障处理

块应该根据应用程序的逻辑捕获异常并决定做什么：

记录并继续处理。这与 Web 应用程序的工作方式没有什么不同 - 请求期间的异常不会导致服务器停机。 明确地向另一个块发送错误消息。这可行，但这种类型的硬编码不是很适合数据流。 使用带有某种错误指示器的消息类型。也许是一个 Success 标志，也许是一个包含消息或错误的 Envelope<TMessage> 对象。 优雅地取消整个流水线，通过发信号通知所有块取消，通过发信号通知CancellationTokenSource 用于产生所有块使用的CancellationTokens。这相当于普通程序中的throw。

#3 是最通用的选项。下游块可以检查信封并忽略或传播失败的消息而不进行处理。本质上，失败的消息会绕过下游块。

另一种选择是使用LinkTo 中的predicate 参数并将失败的消息发送到记录器块，并将成功的消息发送到下一个下游块。在复杂的场景中，这可以用于例如重试某些操作并将结果发送到下游。

这些概念和图像来自 Scott Wlaschin 的 Railway Oriented Programming

【参考方案2】：

TaskScheduler.UnobservedTaskException 事件不是处理故障任务异常的可靠/确定性方法，因为它会延迟到垃圾收集器清理故障任务为止。这可能在错误发生很久之后才发生。

数据流块吞噬的唯一异常类型是OperationCanceledException（AFAIK 出于未记录的原因）。所有其他异常都会导致块转换到故障状态。故障块的Completion 属性（Task）也发生故障（processor.Completion.IsFaulted == true）。您可以将延续附加到 Completion 属性，以便在块失败时接收通知。例如，您可以通过简单地使进程崩溃来确保异常不会被忽视：

processor.Completion.ContinueWith(t =>

    ThreadPool.QueueUserWorkItem(_ => throw t.Exception);
, default, TaskContinuationOptions.OnlyOnFaulted, TaskScheduler.Default);

这是可行的，因为在 ThreadPool 上引发未处理的异常会导致应用程序终止（在引发 AppDomain.CurrentDomain.UnhandledException 事件之后）。

如果您的应用程序有 GUI（WinForms/WPF 等），那么您可以在 UI 线程上抛出异常，从而允许更优雅的错误处理：

var uiContext = SynchronizationContext.Current;
processor.Completion.ContinueWith(t =>

    uiContext.Post(_ => throw t.Exception, null);
, default, TaskContinuationOptions.OnlyOnFaulted, TaskScheduler.Default);

这将在 WinForms 中引发 Application.ThreadException 事件。