c# - 了解 Parallel.Invoke，线程的创建和重用

Question

我试图了解 Parallel.Invoke 如何创建和重用线程。我运行了以下示例代码(来自 MSDN，https://msdn.microsoft.com/en-us/library/dd642243(v=vs.110).aspx):

using System;
using System.Threading;
using System.Threading.Tasks;

class ThreadLocalDemo
{
        static void Main()
        {
            // Thread-Local variable that yields a name for a thread
            ThreadLocal<string> ThreadName = new ThreadLocal<string>(() =>
            {
                return "Thread" + Thread.CurrentThread.ManagedThreadId;
            });

            // Action that prints out ThreadName for the current thread
            Action action = () =>
            {
                // If ThreadName.IsValueCreated is true, it means that we are not the
                // first action to run on this thread.
                bool repeat = ThreadName.IsValueCreated;

                Console.WriteLine("ThreadName = {0} {1}", ThreadName.Value, repeat ? "(repeat)" : "");
            };

            // Launch eight of them. On 4 cores or less, you should see some repeat ThreadNames
            Parallel.Invoke(action, action, action, action, action, action, action, action);

            // Dispose when you are done
            ThreadName.Dispose();
        }
}

据我了解，Parallel.Invoke 尝试在此处创建 8 个线程 - 每个操作一个。所以它创建第一个线程，运行第一个action，然后给线程一个ThreadName。然后它创建下一个线程(获得不同的 ThreadName)等等。

如果它不能创建新线程，它将重用之前创建的线程之一。在这种情况下，repeat 的值将为 true，我们可以在控制台输出中看到这一点。

到这里为止都是正确的吗？

倒数第二个注释(“启动其中的八个。在 4 个或更少的内核上，您应该看到一些重复的线程名称”)暗示 Invoke 创建的线程对应于可用的 cpu 线程处理器:在 4 个内核上，我们有 8 个 cpu 线程，至少有一个忙(运行操作系统和其他东西)，所以 Invoke 只能使用 7 个不同的线程，所以我们必须至少有一个 “重复”。

我对这条评论的解释是否正确？

我在配备 Intel® Core™ i7-2860QM 处理器(即 4 个内核，8 个 cpu 线程)的 PC 上运行了这段代码。我希望至少得到一个 "repeat"，但我没有。当我将 Invoke 更改为执行 10 个而不是 8 个操作时，我得到了这个输出:

ThreadName = Thread6
ThreadName = Thread8
ThreadName = Thread6 (repeat)
ThreadName = Thread5
ThreadName = Thread3
ThreadName = Thread1
ThreadName = Thread10
ThreadName = Thread7
ThreadName = Thread4
ThreadName = Thread9

所以我在控制台应用程序中至少有 9 个不同的线程。这与我的处理器只有 8 个线程的事实相矛盾。

所以我想我上面的一些推理是错误的。 Parallel.Invoke 的工作方式是否与我上面描述的不同？如果是，如何？

Answer 1

如果您将少于 10 个项目传递给 Parallel.Invoke，并且您没有在选项中指定 MaxDegreeOfParallelism(所以 - 您的情况)，它只会运行它们所有在线程池调度程序上并行使用以下代码:

var actions = new [] { action, action, action, action, action, action, action, action };
var tasks = new Task[actions.Length];
for (int index = 1; index < tasks.Length; ++index)
    tasks[index] = Task.Factory.StartNew(actions[index]);
tasks[0] = new Task(actions[0]);
tasks[0].RunSynchronously();
Task.WaitAll(tasks);

所以只是一个普通的 Task.Factory.StartNew。如果您将查看线程池中的最大线程数

int th, io;
ThreadPool.GetMaxThreads(out th, out io);
Console.WriteLine(th);

您会看到一些大数字，例如 32767。因此，执行 Parallel.Invoke 的线程数(在您的情况下)根本不限于 cpu 核心数。即使在 1 核 cpu 上，它也可能并行运行 8 个线程。

您可能会想，为什么有些线程会被重用？因为在线程池线程上完成工作时 - 该线程返回到池中并准备好接受新工作。您的示例中的操作基本上根本不起作用并且完成得非常快。因此，有时通过 Task.Factory.StartNew 启动的第一个线程已经完成了您的操作，并在所有后续线程启动之前返回到池中。因此该线程被重用。

顺便说一下，您可以在您的示例中看到 (repeat) 有 8 个 Action ，如果您足够努力，甚至可以在 8 核(16 个逻辑核)处理器上看到 7 个。

更新以回答您的评论。线程池调度程序不需要立即创建新线程。线程池中有最小和最大线程数。如何查看最大值我已经在上面展示过。查看最小数量:

int th, io;
ThreadPool.GetMinThreads(out th, out io);

这个数字通常等于核心数(例如 8)。现在，当您请求在线程池线程上执行新操作，并且线程池中的线程数小于最小值时 - 将立即创建新线程。但是，如果可用线程数大于最小值 - 在创建新线程之前会引入一定的延迟(不幸的是，我不记得确切的时间，大约 500 毫秒)。

我非常怀疑您在评论中添加的语句能否在 2-3 秒内执行。对我来说，它最多执行 0.3 秒。因此，当线程池创建前 8 个线程时，在创建第 9 个线程之前有 500 毫秒的延迟。在此延迟期间，前 8 个线程中的一些(或全部)线程已完成其工作并可用于新工作，因此无需创建新线程并且可以重复使用它们。

为了验证这一点，引入更大的延迟:

static void Main()
{
    // Thread-Local variable that yields a name for a thread
    ThreadLocal<string> ThreadName = new ThreadLocal<string>(() =>
    {
        return "Thread" + Thread.CurrentThread.ManagedThreadId;
    });

    // Action that prints out ThreadName for the current thread
    Action action = () =>
    {
        // If ThreadName.IsValueCreated is true, it means that we are not the
        // first action to run on this thread.
        bool repeat = ThreadName.IsValueCreated;            
        Console.WriteLine("ThreadName = {0} {1}", ThreadName.Value, repeat ? "(repeat)" : "");
        Thread.Sleep(1000000);
    };
    int th, io;
    ThreadPool.GetMinThreads(out th, out io);
    Console.WriteLine("cpu:" + Environment.ProcessorCount);
    Console.WriteLine(th);        
    Parallel.Invoke(Enumerable.Repeat(action, 100).ToArray());        

    // Dispose when you are done
    ThreadName.Dispose();
    Console.ReadKey();
}

你会看到现在线程池每次都必须创建新线程(比内核多得多)，因为它不能重用以前繁忙的线程。

您还可以增加线程池中的最小线程数，如下所示:

int th, io;
ThreadPool.GetMinThreads(out th, out io);
ThreadPool.SetMinThreads(100, io);

这将消除延迟(直到创建 100 个线程)，在上面的示例中您会注意到这一点。