TPL Dataflow级联连接的数据流块应如何正确实现错误处理？

问题核心原因 #

PropagateCompletion = true 仅支持上游到下游的单向完成信号传播：只有BufferBlock主动调用Complete()后，它的正常完成/故障状态才会同步给下游ActionBlock；反过来下游ActionBlock先抛出异常进入故障状态时，故障信号不会自动反向同步给上游BufferBlock。此时BufferBlock会持续保持可接收状态，一旦内部容量打满，后续的SendAsync调用就会进入永久等待，这就是你遇到阻塞问题的根本原因。

现有方案的合理性 #

你目前通过监听末端块Completion状态、用CancellationTokenSource终止发送流程的实现，属于TPL Dataflow官方推荐的标准错误处理模式，完全符合设计规范。

更优的优化方案 #

1. 简化ContinueWith逻辑，减少无效调度 #

原有的ContinueWith可以增加参数过滤，只在ActionBlock故障时触发取消，同时指定默认调度器避免占用UI同步上下文资源：

actionBlock.Completion.ContinueWith(
    _ => cts.Cancel(), 
    CancellationToken.None,
    TaskContinuationOptions.OnlyOnFaulted,
    TaskScheduler.Default);

2. 简化发送循环逻辑 #

SendAsync方法传入取消令牌后，令牌触发时会直接抛出OperationCanceledException，无需在循环内额外判断令牌状态，代码可以简化为：

using (var cts = new CancellationTokenSource())
{
    // 监听故障取消
    actionBlock.Completion.ContinueWith(
        _ => cts.Cancel(), 
        CancellationToken.None,
        TaskContinuationOptions.OnlyOnFaulted,
        TaskScheduler.Default);

    try
    {
        for (var i = 0; i < 10000; i++)
        {
            if (!await bufferBlock.SendAsync(i, cts.Token))
            {
                break;
            }
        }
    }
    catch (OperationCanceledException)
    {
        // 预期的取消异常，无需额外处理
    }
    finally
    {
        bufferBlock.Complete();
    }

    // 等待链路完成，会抛出原始故障异常
    await actionBlock.Completion;
}

3. 业务场景适配方案 #

如果你的场景允许单个任务失败不影响整体链路运行，可以选择在ActionBlock内部捕获异常做降级处理，不需要终止整个链路：

var actionBlock = new ActionBlock<int[]>(async tasks =>
{
    foreach (var task in tasks)
    {
        try
        {
            await Task.Delay(1);
            if (task > 30)
            {
                throw new InvalidOperationException();
            }
            Console.WriteLine("{0} Completed", task);
        }
        catch (Exception ex)
        {
            // 自定义错误处理：记录日志、写入死信队列等
            Console.WriteLine("任务{0}处理失败：{1}", task, ex.Message);
        }
    }
}, new ExecutionDataflowBlockOptions
{
    BoundedCapacity = 200,
    MaxDegreeOfParallelism = 4
});

这种方式适合数据独立性高、允许部分数据处理失败的场景，整体链路稳定性更高。

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