如何正确并行运行多个异步任务？答案

【问题标题】：How to properly run multiple async tasks in parallel?如何正确并行运行多个异步任务？
【发布时间】：2012-06-03 19:19:41
【问题描述】：

如果您需要并行运行多个异步 I/O 任务，但需要确保同时运行的 I/O 进程不超过 X 个，该怎么办？而前后 I/O 处理任务不应该有这样的限制。

这是一个场景——假设有 1000 个任务；它们中的每一个都接受一个文本字符串作为输入参数；转换该文本（预 I/O 处理），然后将该转换后的文本写入文件。目标是使预处理逻辑利用 100% 的 CPU/内核和 I/O 部分任务以最大 10 度的并行度运行（一次最多同时打开 10 个用于写入文件）。

您能否提供一个示例代码如何使用 C# / .NET 4.5 进行操作？

http://blogs.msdn.com/b/csharpfaq/archive/2012/01/23/using-async-for-file-access-alan-berman.aspx

【问题讨论】：

Rx 2.0 可能非常适合这个（一次将第二阶段限制为 10 个），但我对它还不够熟悉，无法肯定地说。 :-/
这能回答你的问题吗？ Nesting await in Parallel.ForEach

标签： c# asynchronous task-parallel-library async-ctp async-await

【解决方案1】：

我将创建一种扩展方法，在该方法中可以设置最大并行度。 SemaphoreSlim 将成为这里的救星。

    /// <summary>
    /// Concurrently Executes async actions for each item of <see cref="IEnumerable<typeparamref name="T"/>
    /// </summary>
    /// <typeparam name="T">Type of IEnumerable</typeparam>
    /// <param name="enumerable">instance of <see cref="IEnumerable<typeparamref name="T"/>"/></param>
    /// <param name="action">an async <see cref="Action" /> to execute</param>
    /// <param name="maxDegreeOfParallelism">Optional, An integer that represents the maximum degree of parallelism,
    /// Must be grater than 0</param>
    /// <returns>A Task representing an async operation</returns>
    /// <exception cref="ArgumentOutOfRangeException">If the maxActionsToRunInParallel is less than 1</exception>
    public static async Task ForEachAsyncConcurrent<T>(
        this IEnumerable<T> enumerable,
        Func<T, Task> action,
        int? maxDegreeOfParallelism = null)
    {
        if (maxDegreeOfParallelism.HasValue)
        {
            using (var semaphoreSlim = new SemaphoreSlim(
                maxDegreeOfParallelism.Value, maxDegreeOfParallelism.Value))
            {
                var tasksWithThrottler = new List<Task>();

                foreach (var item in enumerable)
                {
                    // Increment the number of currently running tasks and wait if they are more than limit.
                    await semaphoreSlim.WaitAsync();

                    tasksWithThrottler.Add(Task.Run(async () =>
                    {
                        await action(item).ContinueWith(res =>
                        {
                            // action is completed, so decrement the number of currently running tasks
                            semaphoreSlim.Release();
                        });
                    }));
                }

                // Wait for all tasks to complete.
                await Task.WhenAll(tasksWithThrottler.ToArray());
            }
        }
        else
        {
            await Task.WhenAll(enumerable.Select(item => action(item)));
        }
    }

示例用法：

await enumerable.ForEachAsyncConcurrent(
    async item =>
    {
        await SomeAsyncMethod(item);
    },
    5);

【讨论】：

嗨，我在一个线程中使用它。我试图通过 Abort 函数停止线程，但 ForEachAsyncConcurrent 任务仍在运行。你有解决这个问题的办法吗？
@TienNguyen 我会说添加 cancelationToken 作为 ForEachAsyncConcurrent 方法的参数，并在你停止线程时取消它。
你能用cancelationToken更新你的示例代码吗？非常感谢！

【解决方案2】：

我认为为此使用 TPL Dataflow 是一个好主意：您可以创建具有无限并行性的前处理和后处理块、具有有限并行性的文件写入块并将它们链接在一起。比如：

var unboundedParallelismOptions =
    new ExecutionDataflowBlockOptions
    {
        MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded
    };

var preProcessBlock = new TransformBlock<string, string>(
    s => PreProcess(s), unboundedParallelismOptions);

var writeToFileBlock = new TransformBlock<string, string>(
    async s =>
            {
                await WriteToFile(s);
                return s;
            },
    new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 10 });

var postProcessBlock = new ActionBlock<string>(
    s => PostProcess(s), unboundedParallelismOptions);

var propagateCompletionOptions =
    new DataflowLinkOptions { PropagateCompletion = true };

preProcessBlock.LinkTo(writeToFileBlock, propagateCompletionOptions);
writeToFileBlock.LinkTo(postProcessBlock, propagateCompletionOptions);

// use something like await preProcessBlock.SendAsync("text") here

preProcessBlock.Complete();
await postProcessBlock.Completion;

WriteToFile() 可能如下所示：

private static async Task WriteToFile(string s)
{
    using (var writer = new StreamWriter(GetFileName()))
        await writer.WriteAsync(s);
}

【讨论】：

这里的PreProcess 和PostProcess 方法是什么？
@shashwat 他们做任何需要的事情。最初的问题是关于“前后 I/O 处理任务”，所以我使用方法来表示。

【解决方案3】：

听起来您应该考虑使用 Djikstra Semaphore 来控制对任务开始的访问。

但是，这听起来像是典型的队列/固定数量的消费者类型的问题，这可能是一种更合适的结构方式。

【讨论】：