如何指定在 Parallel.ForEach 中执行的并行任务数？ [复制]答案

【问题标题】：How to specify the number of parallel tasks executed in Parallel.ForEach? [duplicate]如何指定在 Parallel.ForEach 中执行的并行任务数？ [复制]
【发布时间】：2014-02-26 08:40:19
【问题描述】：

我有大约 500 个任务，每个任务需要大约 5 秒，大部分时间都浪费在等待远程资源回复上。我想定义自己应该产生的线程数（经过一些测试）并在这些线程上运行任务。当一个任务完成时，我想在可用的线程上生成另一个任务。

我发现System.Threading.Tasks最容易实现我想要的，但我认为不可能指定应该并行执行的任务数量。对于我的机器，它总是在 8 左右（四核 cpu）。是否有可能以某种方式告诉应该并行执行多少个任务？如果不是，实现我想要的最简单的方法是什么？（我尝试使用线程，但代码要复杂得多）。我尝试增加MaxDegreeOfParallelism参数，但它只限制了最大数量，所以这里没有运气......

这是我目前拥有的代码：

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

namespace ConsoleApplication1
{
    class Program
    {
        private static List<string> _list = new List<string>();
        private static int _toProcess = 0;

        static void Main(string[] args)
        {   
            for (int i = 0; i < 1000; ++i)
            {
                _list.Add("parameter" + i);
            }

            var w = new Worker();
            var w2 = new StringAnalyzer();

            Parallel.ForEach(_list, new ParallelOptions() { MaxDegreeOfParallelism = 32 }, item =>
            {
                ++_toProcess;
                string data = w.DoWork(item);
                w2.AnalyzeProcessedString(data);
            });

            Console.WriteLine("Finished");           
            Console.ReadKey();
        }

        static void Done(Task<string> t)
        {            
            Console.WriteLine(t.Result);
            --_toProcess;
        }
    }

    class Worker
    {
        public string DoWork(string par)
        {
            // It's a long running but not CPU heavy task (downloading stuff from the internet)
            System.Threading.Thread.Sleep(5000);            
            return par + " processed";
        }
    }

    class StringAnalyzer
    {
        public void AnalyzeProcessedString(string data)
        {
            // Rather short, not CPU heavy
            System.Threading.Thread.Sleep(1000);
            Console.WriteLine(data + " and analyzed");
        }
    }
}

【问题讨论】：

使用比处理器内核更多的任务通常会减慢它的速度。这就是它受到限制的原因。我猜它限制在 8 个，因为你有 8 个逻辑核心的四核 + 超线程。
恕我直言，您遇到了设计问题。通过创建更多线程来解决这个问题是个坏主意。您应该使用 I/O 完成端口。
@MatthewWatson “大部分时间都浪费在等待远程资源回复上”，所以我认为使用比内核数更多的线程不会有问题在这里（只要远程服务器对此感到满意:)）
@MichalB。创建花费大部分时间等待的任务是一个很大的禁忌。你不应该那样做。需要不同的设计——正如 Fyodor 所说，IO Completion Ports 是一个可能的解决方案。但是创建一个完整的线程负载确实是一个非常糟糕的主意。
@MichalB。告诉你不要做错事很有用，不是吗？它至少会引导你远离糟糕的解决方案。

标签： c# .net parallel-processing parallel.foreach

【解决方案1】：

假设您在获取资源时可以使用 native 异步方法，例如 HttpClient.GetStringAsync，

int numTasks = 20;
SemaphoreSlim semaphore = new SemaphoreSlim(numTasks);
HttpClient client = new HttpClient();

List<string> result = new List<string>();
foreach(var url in urls)
{
    semaphore.Wait();

    client.GetStringAsync(url)
          .ContinueWith(t => {
              lock (result) result.Add(t.Result);
              semaphore.Release();
          });
}

for (int i = 0; i < numTasks; i++) semaphore.Wait();

由于GetStringAsync 在内部使用 IO 完成端口（与大多数其他异步 IO 方法一样）而不是创建新线程，因此这可能是您所追求的解决方案。

另见http://blog.stephencleary.com/2013/11/there-is-no-thread.html

【讨论】：

在内部我使用的是 HttpWebRequest，但我现在看到它也有异步方法（BeginGetRequestStream 和 BeginGetResponse）。我又错过了一些东西:-)。谢谢！
@MichalB。您也可以使用await request.GetResponseAsync();，这样会更容易。下面是一个来自 Microsoft 的有用示例：msdn.microsoft.com/en-us/library/hh300224.aspx 这些使用 IO 完成端口作为它们的实现（但它没有向您公开，这就是为什么您在 C# 文档中看不到关于完成端口的太多明确内容的原因）。另请注意，“重叠 IO”使用完成端口。
@MatthewWatson 我的回答是，await 会破坏同时运行numTasks 的目的。查看代码，如果任务（下载）少于numTasks 进行中，我不会等待/等待。
@MatthewWatson 查看答案中的链接。没有线程...
啊，是的，好点子。但我认为微软的例子确实展示了一个很好的通用方法。

【解决方案2】：

正如 LB 所提到的，.NET Framework 具有执行 I/O 操作（对数据库、Web 服务等的请求）using IOCP internally 的方法，它们可以通过它们的名称来识别 - 它以 Async结尾> 按照惯例。因此，您可以使用它们来构建可以同时处理多个请求的强大的可扩展应用程序。

编辑：我已经用现代最佳实践完全重写了代码示例，因此它变得更具可读性、更短且易于使用。

对于 .NET 4.5，我们可以使用 async-await 方法：

class Program
{
    static void Main(string[] args)
    {
        var task = Worker.DoWorkAsync();
        task.Wait(); //stop and wait until our async method completed

        foreach (var item in task.Result)
        {
            Console.WriteLine(item);
        }

        Console.ReadLine();
    }
}

static class Worker
{
    public async static Task<IEnumerable<string>> DoWorkAsync()
    {
        List<string> results = new List<string>();

        for (int i = 0; i < 10; i++)
        {
            var request = (HttpWebRequest)WebRequest.Create("http://microsoft.com");
            using (var response = await request.GetResponseAsync())
            {
                results.Add(response.ContentType);
            }
        }

        return results;
    }
}

Here 是关于使用 async-await 进行异步编程的精彩 MSDN 教程。

【讨论】：