c# - 如何指定Parallel.ForEach中执行的并行任务数？

Question

这个问题已经有答案了:

Keep running a specific number of tasks (2 个回答)

Have a set of Tasks with only X running at a time (5 个回答)

已关闭 9 年前。

我有大约 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");
        }
    }
}

Answer 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