TPL Dataflow：如何在当前异步操作未完成时启动下一个并保留执行顺序？

考虑以下使用TPL Dataflow的程序，其中ActionBlock来自Dataflow库：

internal static class Program
{
    public static async Task Main(string[] args)
    {
        var actionBlock = new ActionBlock<int>(async i =>
        {
            Console.WriteLine($"Started with {i}");
            await DoSomethingAsync(i);
            Console.WriteLine($"Done with {i}");
        });

        for (int i = 0; i < 5; i++)
        {
            actionBlock.Post(i);
        }

        actionBlock.Complete();
        await actionBlock.Completion;
    }

    private static async Task DoSomethingAsync(int i)
    {
        await Task.Delay(1000);
    }
}

该程序的输出为：

Started with 0
Done with 0
Started with 1
Done with 1
Started with 2
Done with 2
Started with 3
Done with 3
Started with 4
Done with 4

原因是ActionBlock仅在前一个异步任务完成后才会开始处理下一个任务。

如何强制它在前一个任务未完全完成时就开始处理下一个任务？MaxDegreeOfParallelism不可用，因为它会打乱执行顺序。

期望的输出为：

Started with 0
Started with 1
Started with 2
Started with 3
Started with 4
Done with 0
Done with 1
Done with 2
Done with 3
Done with 4

我可以去掉async/await并用ContinueWith替代，但这有两个缺点：

• ActionBlock会认为消息已立即处理完成。此时调用Complete()会直接结束管道，而非等待异步操作完成。
• 我希望添加BoundedCapacity来限制当前未完全完成的消息数量，但由于上述原因，BoundedCapacity无法生效。

解决方案 #

要实现保持启动顺序、异步操作并行执行、完成顺序与启动一致，同时满足Completion等待所有任务完成、BoundedCapacity生效的要求，可以使用TransformBlock + ActionBlock的组合：

代码示例 #

internal static class Program
{
    public static async Task Main(string[] args)
    {
        // 配置TransformBlock：并行处理异步任务，保证输出顺序，设置并发上限
        var transformBlock = new TransformBlock<int, int>(async i =>
        {
            Console.WriteLine($"Started with {i}");
            await DoSomethingAsync(i);
            return i;
        }, new ExecutionDataflowBlockOptions
        {
            MaxDegreeOfParallelism = 5, // 设置并行处理的最大数量
            BoundedCapacity = 5 // 限制未完成的任务数量
        });

        // ActionBlock负责输出完成通知，按顺序接收结果
        var actionBlock = new ActionBlock<int>(i =>
        {
            Console.WriteLine($"Done with {i}");
        });

        // 链接两个块，完成时自动传播完成信号
        transformBlock.LinkTo(actionBlock, new DataflowLinkOptions { PropagateCompletion = true });

        // 发布消息
        for (int i = 0; i < 5; i++)
        {
            await transformBlock.SendAsync(i); // 若BoundedCapacity满，会等待有空闲位置再发送
        }

        // 标记TransformBlock完成
        transformBlock.Complete();
        // 等待整个管道完成
        await actionBlock.Completion;
    }

    private static async Task DoSomethingAsync(int i)
    {
        await Task.Delay(1000);
    }
}

方案说明 #

1. TransformBlock的作用：

   • 负责执行Started输出和异步操作，通过MaxDegreeOfParallelism设置并行处理的最大数量，实现同时启动多个异步任务。
   • 由于TransformBlock默认保证输出顺序与输入顺序一致，即使异步任务并行执行，结果也会按输入顺序传递给后续的ActionBlock，确保Done输出的顺序正确。
   • BoundedCapacity参数可以限制当前正在处理或等待处理的消息数量，避免内存溢出。

2. ActionBlock的作用：

   • 仅负责接收TransformBlock处理完成的结果，输出Done通知，逻辑简单且按顺序执行。

3. 链接与完成处理：

   • 通过LinkTo的PropagateCompletion = true，当TransformBlock标记完成后，会自动将完成信号传递给ActionBlock，无需手动处理。
   • 调用SendAsync而非Post，可以在BoundedCapacity满时等待，避免消息丢失。

该方案既实现了并行启动任务、保持顺序的需求，又解决了ContinueWith方案的两个缺点，同时满足Completion等待所有任务完成、BoundedCapacity生效的要求。

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