Parallel.For vs ThreadPool and async/await

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.Forasync不知道。

因此，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)=>{
//$$$
});