Parallel.ForEach和async-await

Question

我有这样的方法:

public async Task<MyResult> GetResult()
{
    MyResult result = new MyResult();

    foreach(var method in Methods)
    {
        string json = await Process(method);

        result.Prop1 = PopulateProp1(json);
        result.Prop2 = PopulateProp2(json);

    }

    return result;
}

然后我决定使用Parallel.ForEach:

public async Task<MyResult> GetResult()
{
    MyResult result = new MyResult();

    Parallel.ForEach(Methods, async method =>
    {
        string json = await Process(method);    

        result.Prop1 = PopulateProp1(json);
        result.Prop2 = PopulateProp2(json);
    });

    return result;
}

但现在我有一个错误:

在异步操作仍处于挂起状态时完成异步模块或处理程序.

Answer 1

async不适合ForEach.特别是,您的asynclambda正在转换为async void方法.有许多理由要避免async void(正如我在MSDN文章中所描述的那样); 其中之一就是你无法轻易检测到asynclambda 何时完成.ASP.NET将在不完成async void方法的情况下看到您的代码返回,并(适当地)抛出异常.

您可能想要做的是同时处理数据,而不是并行处理.几乎不应该在ASP.NET上使用并行代码.以下是异步并发处理的代码:

public async Task<MyResult> GetResult()
{
  MyResult result = new MyResult();

  var tasks = Methods.Select(method => ProcessAsync(method)).ToArray();
  string[] json = await Task.WhenAll(tasks);

  result.Prop1 = PopulateProp1(json[0]);
  ...

  return result;
}

Answer 2

.NET 6 最终添加了Parallel.ForEachAsync，这是一种安排异步工作的方法，允许您控制并行度：

var urlsToDownload = new [] 
{
    "https://dotnet.microsoft.com",
    "https://www.microsoft.com",
    "https://twitter.com/shahabfar"
};

var client = new HttpClient();

var options = new ParallelOptions { MaxDegreeOfParallelism = 2 };
await Parallel.ForEachAsync(urlsToDownload, options, async (url, token) =>
{
    var targetPath = Path.Combine(Path.GetTempPath(), "http_cache", url);

    var response = await client.GetAsync(url, token);
    // The request will be canceled in case of an error in another URL.

    if (response.IsSuccessStatusCode)
    {
        using var target = File.OpenWrite(targetPath);

        await response.Content.CopyToAsync(target);
    }
});

Answer 3

或者,使用AsyncEnumerator NuGet包,您可以执行以下操作:

using System.Collections.Async;

public async Task<MyResult> GetResult()
{
    MyResult result = new MyResult();

    await Methods.ParallelForEachAsync(async method =>
    {
        string json = await Process(method);    

        result.Prop1 = PopulateProp1(json);
        result.Prop2 = PopulateProp2(json);
    }, maxDegreeOfParallelism: 10);

    return result;
}

哪里ParallelForEachAsync是扩展方法.

Answer 4

啊,好的 我想我知道现在发生了什么. async method =>"异常无效",即"火与忘"(不推荐用于除事件处理程序之外的任何其他内容).这意味着调用者无法知道它何时完成...因此,GetResult在操作仍在运行时返回.虽然我的第一个答案的技术细节是不正确的,但结果在这里是相同的:GetResult在启动的操作ForEach仍在运行时返回.你唯一能做的就是不await打开Process(以便不再使用lambda async)并等待Process完成每次迭代.但是,这将使用至少一个线程池线程来做到这一点,从而稍微强调池 - 可能使用ForEach毫无意义.我不会使用Parallel.ForEach ...