在 Task.WhenAll 中执行多个任务时的 C# 线程

如果在单个线程上执行以下操作，会发生什么：

await Task.WhenAll(items.select(x => SomeAsyncMethod(x)))

// Where SomeAsyncMethod is defined like this (writeAsync is pure async io)
async Task SomeAsyncMethod(Item item){
  await myDevice.writeAsync(...).ConfigureAwait(false);
  //do some cpu intensive stuff...
}

并说有 10.000 个项目items。当每个SomeAsyncMethod在等待之后继续，然后在线程池中的线程上执行此操作。所以当许多SomeAsyncsMethods 返回将同时获取线程池中的多个线程，或者仅单个线程执行“执行一些 CPU 密集型操作”SomeAsyncMethod在这种情况下的任何特定时刻？

更新：好的，这是一个示例程序。当我在具有 8 个逻辑核心的 PC 上对此进行测试时，最小线程为 12 或 13，最大线程在 35-40 范围内结束。 因此看起来好像将在逻辑核心上创建最多 4 个线程。创建 10.000 个或 100.000 个文件并不重要 - 使用相同的最大线程数 - 也许这是因为所有任务都排队等待访问文件系统？请注意，该程序将在 c:\tmp\asynctest 中创建大量小文件：

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace ConsoleApplication4 {
    internal class Program {
        private static void Main(string[] args) {
            var myDevice = new MyDevice();
            var ints = new List<int>();
            for (var i = 0; i < 10000; i++) {
                ints.Add(i);
            }
            var task = Task.WhenAll(ints.Select(i => myDevice.WriteTextAsync(i.ToString())));
            task.Wait();
            Console.WriteLine("Max thread count = " + myDevice.MaxThreadCount);
            Console.WriteLine("Min thread count = " + myDevice.MinThreadCount);
            Console.ReadLine();
        }
    }

    public class MyDevice {
        public ConcurrentDictionary<string, string> ThreadIds;
        public int MaxThreadCount;
        public int MinThreadCount = Process.GetCurrentProcess().Threads.Count;
        public async Task WriteTextAsync(string text) {
            var filePath = @"c:\tmp\asynctest\" + text + ".txt";
            var encodedText = Encoding.Unicode.GetBytes(text);
            using (var sourceStream = new FileStream(filePath,
                FileMode.Append, FileAccess.Write, FileShare.None, bufferSize: 4096, useAsync: true)) {
                await sourceStream.WriteAsync(encodedText, 0, encodedText.Length).ConfigureAwait(false);
                MaxThreadCount = Math.Max(MaxThreadCount, Process.GetCurrentProcess().Threads.Count);
                MinThreadCount = Math.Min(MinThreadCount, Process.GetCurrentProcess().Threads.Count);
            }
        }
    }
}

更新 2。现在，如果我启动多个线程，每个线程同时执行大量 aysnc io 任务，那么与更新 1 中的单线程示例相比，看起来总共并没有使用更多线程。在我刚刚运行的测试中，其中每个文件由 4 个线程创建 10.000 个文件，然后最大线程为 41，最小线程为 12 - 因此似乎有一些集中控制用于异步任务延续的线程数量。下面是一个示例，其中 4 个线程各自启动 10.000 个异步操作：

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace ConsoleApplication4 {
    internal class Program {
        private static void Main(string[] args) {
            var myDevice = new MyDevice();
            var ints = new List<int>();
            const int limit = 10000;
            for (var i = 0; i < limit; i++) {
                ints.Add(i);
            }

            List<Task> jobs = new List<Task>();
            for (var j = 0; j < 4*limit; j+=limit) {
                var jobid = j;
                jobs.Add(Task.Run(() => Runjob(ints, myDevice, jobid)));
            }
            Task.WaitAll(jobs.ToArray());

            Console.WriteLine("Max thread count = " + myDevice.MaxThreadCount);
            Console.WriteLine("Min thread count = " + myDevice.MinThreadCount);
            Console.ReadLine();
        }

        private static void Runjob(List<int> ints, MyDevice myDevice, int jobid) {
            Console.WriteLine("Starting job " + jobid);
            var task = Task.WhenAll(ints.Select(i => myDevice.WriteTextAsync((jobid+i).ToString())));
            task.Wait();
            Console.WriteLine("Finished job " + jobid);
        }
    }

    public class MyDevice {
        public int MaxThreadCount;
        public int MinThreadCount = Process.GetCurrentProcess().Threads.Count;
        public async Task WriteTextAsync(string text) {
            var filePath = @"c:\tmp\asynctest\" + text + ".txt";
            var encodedText = Encoding.Unicode.GetBytes(text);
            using (var sourceStream = new FileStream(filePath,
                FileMode.Append, FileAccess.Write, FileShare.None, bufferSize: 4096, useAsync: true)) {
                await sourceStream.WriteAsync(encodedText, 0, encodedText.Length).ConfigureAwait(false);
                MaxThreadCount = Math.Max(MaxThreadCount, Process.GetCurrentProcess().Threads.Count);
                MinThreadCount = Math.Min(MinThreadCount, Process.GetCurrentProcess().Threads.Count);
            }
        }
    }
}

最有可能的情况是，“CPU 密集型工作”将分别发生在随机线程池线程上 - 如果它确实受 CPU 限制，则每个逻辑核心将获得大约 1-2 个线程来完成工作。

关键点是，虽然继续原来的任务（Task.WhenAll）将在 UI 线程上运行（如果有is当然，同步上下文），各个 I/O 操作的延续将发布到线程池上，因为您明确请求忽略同步上下文（ConfigureAwait(false)).

但是，如果 I/O 请求同步完成，则所有内容也有可能在原始线程上运行。在这种情况下，不会进行异步调度，任务也没有机会切换线程。如果需要确保并行化，则必须使用Task.Run明确地。

还应该注意的是，这主要取决于实现，而不是您可以依赖的东西。对于严重异步 I/O 应用程序来说，这也可能是一个糟糕的方法，因为您可能在 I/O 线程池中的线程上运行 CPU 密集型内容 - 扰乱线程池中线程的框架平衡，并防止新的异步响应出现，直到您完成工作。如果您正在做的工作尤其如此isn't纯粹的 CPU 工作——例如，在 Web 服务器之类的东西上，线程池线程上的阻塞可能会非常痛苦。