通过异步示例了解 C# 中的并行编程答案

【问题标题】：Understand parallel programming in C# with async examples通过异步示例了解 C# 中的并行编程
【发布时间】：2019-10-05 20:22:10
【问题描述】：

我正在尝试理解并行编程，我希望我的 async 方法可以在多个线程上运行。我写了一些东西，但它并没有像我想象的那样工作。

代码

public static async Task Main(string[] args)
{
    var listAfterParallel =  RunParallel(); // Running this function to return tasks
    await Task.WhenAll(listAfterParallel); // I want the program exceution to stop until all tasks are returned or tasks are completed
    Console.WriteLine("After Parallel Loop"); // But currently when I run program, after parallel loop command is printed first
    Console.ReadLine();
}

public static async Task<ConcurrentBag<string>> RunParallel()
{
     var client = new System.Net.Http.HttpClient();
     client.DefaultRequestHeaders.Add("Accept", "application/json");
     client.BaseAddress = new Uri("https://jsonplaceholder.typicode.com");
     var list = new List<int>();
     var listResults = new ConcurrentBag<string>();
     for (int i = 1; i < 5; i++)
     {
       list.Add(i);
     }
     // Parallel for each branch to run await commands on multiple threads. 
     Parallel.ForEach(list, new ParallelOptions() { MaxDegreeOfParallelism = 2 }, async (index) =>
     {
         var response = await client.GetAsync("posts/" + index);
         var contents = await response.Content.ReadAsStringAsync();
         listResults.Add(contents);
         Console.WriteLine(contents);
     });
     return listResults;
}

我希望在打印“并行循环之后”之前完成 RunParallel 函数。另外我希望我的 get posts 方法可以在多个线程上运行。

任何帮助将不胜感激！

【问题讨论】：

并行与异步不同。一个是在多个线程上运行东西，另一个是在等待某事（通常是 IO，但可能是另一个线程完成某些工作）时不阻塞线程。如果您希望 IO 并行发生，您只需要收集任务并摆脱 Parallel.ForEach
async/await 不用于并行性，它们有助于异步操作。 Parallel.ForEach 用于数据并行（在本地处理 100K/1M 项），绝对不用于异步工作。事实上，它不能等待任何异步操作。此代码将同时触发所有请求，并且永远不会收到结果
在任何情况下，异步操作已经在另一个线程上运行，或者在完成之前不会打扰它们的创建者。您可以使用例如var results = Task.WhenAll(Enumerable.Range(1,5).Select(i=>client.GetStringAsync($"posts/{i}"))); 来启动所有 5 个任务并等待它们的结果而不会阻塞
@PanagiotisKanavos 这会在多个线程中运行吗？
@PanagiotisKanavos 你会用什么来代替 Parallel.ForEach？我们可以将 ActionBlock 用于数据并行和异步工作吗？

标签： c# parallel-processing task-parallel-library parallel.foreach

【解决方案1】：

这里发生的情况是，您永远不会等待 Parallel.ForEach 块完成 - 您只是返回最终将泵入的袋子。原因是因为Parallel.ForEach 需要Action 代表，所以您创建了一个返回void 而不是Task 的lambda。虽然async void 方法是有效的，但它们通常会在新线程上继续工作并在await 一个任务时立即返回给调用者，因此Parallel.ForEach 方法认为处理程序已经完成，即使它被踢了剩下的工作放到一个单独的线程中。

这里改为使用同步方式；

Parallel.ForEach(list, new ParallelOptions() { MaxDegreeOfParallelism = 2 }, index => 
{
    var response = client.GetAsync("posts/" + index).Result;

    var contents = response.Content.ReadAsStringAsync().Result;
    listResults.Add(contents);
    Console.WriteLine(contents);
});

如果你绝对必须在里面使用await，请将其包裹在Task.Run(...).GetAwaiter().GetResult()中；

Parallel.ForEach(list, new ParallelOptions() { MaxDegreeOfParallelism = 2 }, index => Task.Run(async () =>
{
    var response = await client.GetAsync("posts/" + index);

    var contents = await response.Content.ReadAsStringAsync();
    listResults.Add(contents);
    Console.WriteLine(contents);
}).GetAwaiter().GetResult();

然而，在这种情况下，Task.run 通常会转到一个新线程，因此我们颠覆了 Parallel.ForEach 的大部分控制；最好一直使用async；

var tasks = list.Select(async (index) => {
        var response = await client.GetAsync("posts/" + index);

        var contents = await response.Content.ReadAsStringAsync();
        listResults.Add(contents);
        Console.WriteLine(contents);
    });
await Task.WhenAll(tasks);

由于Select 需要一个Func<T, TResult>，它会将没有return 的async lambda 解释为async Task 方法而不是async void，从而给我们一些我们可以明确地await

【讨论】：

我喜欢你的方法，但你能确认 list.select 函数是否会在多个线程上运行并且会并行运行吗？我还想传递一个可以使用多少个内核的选项
在您使用任务时并行运行是由同步上下文控制的。我将更新以显示使用 Select 和 AsParallel 的解决方案来制作相同的控件
另外，我们可以为异步方法使用动作块吗，因为我还想传递它可以使用的核心/线程数
如果你想使用 Async 方法但定义最大并行化，你需要弄乱你的本地 TaskScheduler，由于 TaskScheduler 是抽象的，这是一个更大范围的问题

【解决方案2】：

看看这个：There Is No Thread

当您发出多个并发 Web 请求时，不是您的 CPU 正在做艰苦的工作。它是为您的请求提供服务的 Web 服务器的 CPU。您的 CPU 在此期间什么也不做。它不是处于特殊的“等待状态”之类的。盒子里正在工作的硬件是你的网卡，它将数据写入你的 RAM。当收到响应时，您的 CPU 将收到有关到达数据的通知，因此它可以对它们进行处理。

当您在自己的盒子里有繁重的工作要做时，您需要并行性，而不是当您希望外部世界完成繁重的工作时。从你的 CPU 的角度来看，甚至你的硬盘也是外部世界的一部分。因此，适用于 Web 请求的所有内容也适用于针对文件系统和数据库的请求。这些工作负载称为I/O bound，以区别于所谓的 CPU 绑定工作负载。

对于 I/O 绑定工作负载，.NET 平台提供的工具是异步Task。整个库中有多个 API 返回 Task 对象。为了实现并发，您通常启动多个任务，然后使用Task.WhenAll await 它们。还有更高级的工具，例如TPL Dataflow library，它是在Tasks 之上构建的。它提供缓冲、批处理、配置最大并发度等功能。

【讨论】：