下游块中的 TPL 数据流和异常处理

我有以下伪代码：

var queue = new BufferBlock<int>(new DataflowBlockOptions { BoundedCapacity = 5 });
var a = new ActionBlock<int>(async item =>
    {
        await Task.Delay(500);
        Trace.TraceInformation(
            $"Target 1: | Type: {typeof(int).Name} | Thread: {Thread.CurrentThread.ManagedThreadId} | Message: {item}");
        // handling some logic but it throws
        if (item >= 5) throw new Exception("Something bad happened");

    }, new ExecutionDataflowBlockOptions { BoundedCapacity = 1, MaxDegreeOfParallelism = 1 });

queue.LinkTo(a, new DataflowLinkOptions { PropagateCompletion = true });

var targets = new List<ITargetBlock<int>> {queue};

var broadcaster = new ActionBlock<int>(
    async item =>
    {
        var processingTasks = targets.Select(async t =>
        {
            try
            {
                // This is condition is always false
                // t (bufferblock) has no exceptions. Exception is raised in downstream action block where it sends to
                if (!await t.SendAsync(item))
                    await t.Completion;
            }
            catch (Exception e)
            {
                Trace.TraceInformation("Handled exception : " + e.Message);
            }
        });

        try
        {
            // Neither here the exception is rethrowed
            await Task.WhenAll(processingTasks);
        }
        catch (Exception e)
        {
            Trace.TraceInformation("Handled exception WhenAll : " + e.Message);
        }
    });

for (var i = 1; i <= 10; i++)
{
    broadcaster.Post(i);
}

管道是这样配置的ActionBlock<int> => BufferBlock<int> => ActionBlock<int>.

最后ActionBlock<int>抛出异常，但它不会重新抛出到我想要处理它的源代码块。

如何重写此代码以便正确处理异常？

您可以找到该主题的官方指南here https://blogs.msdn.microsoft.com/pfxteam/2011/11/09/exception-handling-in-tpl-dataflow-networks/。总体解决方案是订阅所有区块Completion https://msdn.microsoft.com/en-us/library/system.threading.tasks.dataflow.idataflowblock.completion.aspx任务包括检查其状态，并在需要时替换有故障的块（也应该存储块的所有引用）。请参阅整篇文章以获取更多信息。

网络的行为Faulted blocks

1. 保留消息
   为了避免消息损坏，故障块应该清除其消息队列并移动到Faulted状态 尽快地。有一个场景不符合 这条规则：源块保存目标保留的消息。 如果遇到内部异常的块有一条消息： 被目标保留，保留消息必须不 掉落，并且该块不应移动到Faulted状态 直到消息被释放或消费。

2. 悬挂网络
   ...

   • 保留对网络中所有块的引用并使用Task.WaitAll https://msdn.microsoft.com/en-us/library/system.threading.tasks.task.waitall.aspx or Task.WhenAll https://msdn.microsoft.com/en-us/library/system.threading.tasks.task.whenall.aspx等待他们（同步或 异步）。如果一个块出现故障，它的Completion https://msdn.microsoft.com/en-us/library/system.threading.tasks.dataflow.idataflowblock.completion.aspx任务将 完成于Faulted state.
   • Use DataflowLinkOptions https://msdn.microsoft.com/en-us/library/system.threading.tasks.dataflow.dataflowlinkoptions.aspx with PropagateCompletion == true构建线性网络时。这将从中传播块完成 源到目标。在这种情况下，等待网络就足够了 叶块。