c# - Parallel.For与ThreadPool和异步/等待

Question

我正在尝试以异步方式处理5000个文件，而又不限制线程池的增长。但是，Parallel.For循环并没有给我一个一致的正确答案（计数很快出现了），而Task.Run是。

在导致这些错误答案的Parallel.For循环中，我在做什么错？

using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

class Program
{
    static volatile int count = 0;
    static volatile int total = 0;
    static void Main(string[] args)
    {
        Parallel.For(0, 5000, new ParallelOptions { MaxDegreeOfParallelism = 10 },
            async (index) =>
            {
                string filePath = $"C:\\temp\\files\\out{index}.txt";
                var bytes = await ReadFileAsync(filePath);
                Interlocked.Add(ref total, bytes.Length);
                Interlocked.Increment(ref count);
            });
        Console.WriteLine(count);
        Console.WriteLine(total);

        count = 0;
        total = 0;
        List<Task> tasks = new List<Task>();
        foreach (int index in Enumerable.Range(0, 5000))
        {
            tasks.Add(Task.Run(async () =>
            {
                string filePath = $"C:\\temp\\files\\out{index}.txt";
                var bytes = await ReadFileAsync(filePath);
                Interlocked.Add(ref total, bytes.Length);
                Interlocked.Increment(ref count);
            }));
        }
        Task.WhenAll(tasks).Wait();
        Console.WriteLine(count);
        Console.WriteLine(total);
    }
    public static async Task<byte[]> ReadFileAsync(string filePath)
    {
        byte[] bytes = new byte[4096];
        using (var sourceStream = new FileStream(filePath,
                FileMode.Open, FileAccess.Read, FileShare.Read,
                bufferSize: 4096, useAsync: true))
        {
            await sourceStream.ReadAsync(bytes, 0, 4096);
        };
        return bytes;
    }
}

Answer 1

Parallel.For不了解async。

因此，Parallel.For的运行不符合您的预期。由于异步lambda生成的任务没有等待，因此所有迭代将在创建任务而不是完成任务的时间内完成。

Parallel.For之后，许多迭代仍将具有尚未完成的待处理任务，因此，您对count和total的添加尚未完成。

Stephen Toub实现了Parallel.ForEach的异步版本。 （ForEachAsync）实现如下：

public static Task ForEachAsync<T>(this IEnumerable<T> source, int dop, Func<T, Task> body)
{
    return Task.WhenAll(
        from partition in Partitioner.Create(source).GetPartitions(dop)
        select Task.Run(async delegate {
            using (partition)
                while (partition.MoveNext())
                    await body(partition.Current);
        }));
}

因此，您可以重写循环：

Enumerable.Range(0, 5000).ForEachAsync(10, async (index)=>{
   //$$$
});