使用线程、线程池和任务在 C# 中进行并行处理

- 日理万妓 2023-10-01 20:14 2阅读 0赞

多线程是在您的进程中同时运行多个操作以最大可能地利用 CPU 功率的概念。一个线程定义了一个执行路径。当进程启动时，它会启动一个称为主线程的线程。如果您的程序有许多复杂的路径，这些路径会执行繁重的工作并执行大量耗时的操作，您可以在其单独的线程中运行每个路径并并行实现结果。例如，如果您的程序处理磁盘上的大文件以提取数据，并且您希望能够并行处理文件而不是一个接一个地处理文件，您可以在单独的线程中执行每个文件处理。

#### 线**程** ####

**C\# 中的线程**代表实际的操作系统线程。每个都有自己的堆栈和内核资源。创建、管理和销毁线程的成本非常高。一旦线程被剥离，它就会消耗它的堆栈内存。CPU 还必须进行时间切片并在线程之间不断切换。

using System;
    using System.Threading;
    
    namespace ThreadSample
    {
        class Program
        {
            static void Main(string[] args)
            {
                Thread myThread = new Thread(new ThreadStart(RunThread));
                myThread.IsBackground = true;
                myThread.Start();
                myThread.Join();
                Console.WriteLine("I am in Main Thread. myThread Exited!");
                Console.ReadLine();
            }
    
            private static void RunThread()
            {
                Console.WriteLine("I am in myThread");
                Thread.Sleep(5000);
            }
        }
    }

Thread 类的构造函数接受两种委托方法。

1.  **ThreadStart** – 没有参数的方法
2.  **ParameterizedThreadStart** – 接受带有对象类型参数的方法

有两种类型的线程：

1.  **前台线程**– 当线程的 IsBackground 属性设置为 false（默认）时。即使父前台线程被中止，前台线程也会继续执行。
2.  **背景线程**– 当 IsBackground 正确设置为 true 时。当取消它的前台线程被中止时，后台线程被中止。

您可以使用 .Sleep() 方法暂停线程的执行。它将放弃分配给线程的时间片并将其交还给 CPU。您还可以看到我使用了 .Join() 方法。可以在Parent线程中使用，等待线程完成工作并退出。请记住，在父线程中调用 Join 将阻塞父线程的执行，直到子线程完成或退出。您可以使用 .Abort() 方法中止线程的执行。

线程一退出通过参数传递的方法就停止执行。为了让线程永远继续工作，您需要在执行方法中编写一个带有中断条件的循环。

除非您有需要更好地控制操作的特定用例，否则建议您使用 ThreadPool 代替跨越线程。

#### 线程池 ####

**ThreadPool**是一个 .NET 类，可让您访问由 CLR 维护的线程池。它让您提交要执行的操作，CLR 负责将其调度到池中的一个线程上。您无法控制执行，调度程序根据池上空闲线程的可用性决定工作何时开始执行。虽然它可以让你从管理线程中解放出来，但如果你想执行长时间运行的操作，线程池不是一个好的选择。您永远无法确定工作是否已完成执行，也无法获得执行结果作为回报。

让我们修改 Thread 示例中编写的示例代码以使用 ThreadPool 代替。ThreadPool 通过 .QueueUserWorkItem 方法接受工作项，该方法接受委托 WaitCallback，您可以在其中传递带参数的方法。它还接受状态作为对象类型的第二个参数，这是可选的。

using System;
    using System.Threading;
    
    namespace ThreadSample
    {
        class Program
        {
            static void Main(string[] args)
            {
                ThreadPool.QueueUserWorkItem(new WaitCallback(RunThread));
                Console.ReadLine();
            }
    
            private static void RunThread(object state)
            {
                Console.WriteLine("I am in ThreadPool Thread");
                Thread.Sleep(5000);
                Console.WriteLine("ThreadPool thread is going to exit");
            }
        }
    }

您会注意到 ThreadPool 线程没有 Join Method。您无法通过任何直接方法确定线程是否已完成执行。一旦你在 ThreadPool 中排队工作项，主线程就会继续执行。如果要等到线程完成执行，则必须使用同步事件编写代码。

**使用线程同步事件**

线程同步事件有两种类型：

1.  **ManualResetEvent** – 这是一个像我们家中的普通门一样工作的事件。您可以使用 .Set() 方法设置（打开）它并使用 .Reset() 方法重置（关闭）它。它将阻塞调用 .WaitOne() 的线程，直到它被设置。设置后，事件对象的状态将处于 Set 状态，直到您使用 .Reset() 方法手动重置它。
2.  **AutoResetEvent** – 它的作用与 ManualResetEvent 相同，只是它的作用类似于自动门。一旦你设置它，它允许通过调用 .WaitOne() 等待的线程通过，然后将自身重置回来。就像自动门允许人通过并再次关闭一样。

在下面的示例中，我们在构建时将 myWaitHandle 初始化为未设置（假）状态。我们将它作为参数传递给 QueueUserWorkItem，它将在 RunThread 方法中作为状态参数可用。然后使用`.Set()`由 ThreadPool 线程执行的 RunThread 中的方法设置它。您的主线程将通过调用 myWaitHandle 上的 .WaitOne() 来等待，直到发出信号。

using System;
    using System.Threading;
    
    namespace ThreadSample
    {
        class Program
        {
            static void Main(string[] args)
            {
                ManualResetEvent myWaitHandle = new ManualResetEvent(false);
    
                ThreadPool.QueueUserWorkItem(new WaitCallback(RunThread), myWaitHandle);
                myWaitHandle.WaitOne();
    
                Console.WriteLine("ThreadPool thread has completed the Work and Set myWaitHandle");
                Console.ReadLine();
            }
    
            private static void RunThread(object state)
            {
                ManualResetEvent waitHandleFromParent = (ManualResetEvent)state;
    
                Console.WriteLine("I am in ThreadPool Thread");
                Thread.Sleep(5000);
                Console.WriteLine("ThreadPool thread is going to exit");
                waitHandleFromParent.Set();
            }
        }
    }

#### 任务 ####

**2010 年引入的任务并行库中的任务**类为您提供了上述两个问题的解决方案。许多人将任务与轻量级线程混淆，但任务不能与线程相提并论。任务只是一组要执行的作业。线程执行调度到 TaskScheduler 的任务。任务不保证并行处理，并根据资源的可用性进行调度。默认情况下，任务由 ThreadPool 上的 TaskScheduler 执行。如果您通过传递 LongRunning 选项将任务指定为长时间运行任务，TaskScheduler 将为该任务衍生新线程。与 Threads 和 ThreadPool 不同，task 也可以返回执行结果。

让我们重新编写上面的 Thread/ThreadPool 使用 Task 实现的示例代码。请记住，在以下示例中，任务**不是异步运行**的，因为我们在主线程中调用 task.Wait()，主线程被阻塞直到任务完成。任务非常适合使用 async/await 进行异步执行，我们将在后面介绍。

using System;
    using System.Threading;
    using System.Threading.Tasks;
    namespace ThreadSample
    {
        class Program
        {
            static void Main(string[] args)
            {
                var task = new Task(RunTask);
                task.Start();
                task.Wait();
    
                Console.WriteLine("Back to main thread. Task completed execution!");
                Console.ReadLine();
            }
    
            private static void RunTask()
            {
                Console.WriteLine("I am in Task");
                Thread.Sleep(5000);
            }
        }
    }

让我们看一个 LongRunning 任务的例子。

using System;
    using System.Threading;
    using System.Threading.Tasks;
    namespace ThreadSample
    {
        class Program
        {
            static void Main(string[] args)
            {
                var task = new Task(RunTask,TaskCreationOptions.LongRunning);
                task.Start();
                task.Wait();
    
                Console.WriteLine("Program will never reach to this line");
            }
    
            private static void RunTask()
            {
                Console.WriteLine("Entered Long Running Task");
    
                while (true)
                {
                    Thread.Sleep(100);
                }
            }
        }
    }

### 线程之间的同步 ###

到目前为止，我们已经看到了如何启动单线程的示例。但是一旦您开始构建更复杂的系统并分拆多个线程来处理将由线程共享的复杂数据，您将需要在线程之间同步对数据的访问。有多种同步方式，每种方式都适用于不同的用例。

#### 联锁 ####

很多时候，您只需要更改线程之间共享的公共变量。例如，您有一个计数器变量，您需要在处理完每个作业后递增该变量。如果您尝试在没有任何同步的情况下增加变量，多个线程将同时访问该变量并增加它创建竞争条件，任何线程可能具有已被其他线程增加的陈旧值。

下面是增加变量 Counter 的程序，这将产生随机结果，因为多个线程在没有同步的情况下访问同一个变量。

using System;
    using System.Threading;
    using System.Threading.Tasks;
    namespace ThreadSample
    {
        class Program
        {
            private static int Counter = 0;
            static void Main(string[] args)
            {
                Task task1 = new Task(RunTask, TaskCreationOptions.LongRunning);
                Task task2 = new Task(RunTask, TaskCreationOptions.LongRunning);
                Task task3 = new Task(RunTask, TaskCreationOptions.LongRunning);
                Task task4 = new Task(RunTask, TaskCreationOptions.LongRunning);
                Task task5 = new Task(RunTask, TaskCreationOptions.LongRunning);
    
    
                task1.Start();
                task2.Start();
                task3.Start();
                task4.Start();
                task5.Start();
    
                Task.WaitAll(task1, task2, task3, task4, task5);
    
                Console.WriteLine("Program will never reach to this line");
            }
    
            private static void RunTask()
            {
                while (true)
                {
                    Counter += 1;
                    Console.Write("Counter value is {0}\r\n", Counter);
                }
            }
        }
    }

Interlocked 提供了对跨线程共享的变量执行原子操作的方法。Interlocked 提供五种基本操作。为简单起见，我将那些没有特定数据类型的数据包括在内。

1.  **Add** – 添加两个整数或 Log 值，并将第一个替换为和，作为原子操作。
2.  **增量**\- 将变量加一
3.  **Decrement** – 将变量减一
4.  **Exchange(v1,v2)** – 交换作为参数提供的变量的值并返回第一个变量的原始值。
5.  **CompareExchange(v1, v2, value)** – 比较两个变量，如果变量相等，则将最后一个参数的值设置为第一个参数的变量。

如果我们使用`Interlocked.Increment`Counter += 1 而不是，您会注意到 now 变量以正确的顺序递增，没有任何随机性。

private static void RunTask()
            {
                while (true)
                {
                    Interlocked.Increment(ref Counter);
                    Console.Write("Counter value is {0}\r\n", Counter);
                }
            }

同样，您可以使用`Interlocked.Add(ref Counter, 10);`将计数器增加 10 而不是 1。在 MSDN 中有一个很好的 Interlocked.Exchange 示例。[您可以在MSDN 文档][MSDN]中阅读有关 Interlocked 的更多信息

#### 监控/锁定 ####

Monitor 提供了通过获取和释放特定对象的锁定来同步对代码区域（通常称为临界区）的访问的机制。例如，如果有 5 个线程，并且如果线程 1 获得对象上的锁，则其他四个线程都不能访问该对象，直到第一个线程释放锁。锁定通过调用对象的 Monitor.Enter 获得，并通过调用 Monitor.Exit 释放。调用 Monitor.Exit 很重要，否则线程将永远获取对象上的锁，并且没有其他线程能够继续前进。最佳做法是在 finally 块中调用 Monitor.Exit，以便即使在出现错误的情况下也能执行。Monitor.Enter 和 Monitor.Exit 有一个简写形式，称为 lock。

lock(variableName){
        //Do Something
    }

和下面的完全一样

Monitor.Enter(variableName);
    try
    {
        //Do Something
    }
    finally
    {
         Monitor.Exit(variableName);
    }

以下示例包含五个线程。1 个线程是 Manager 线程，另外 4 个是 Worker 线程。管理器线程将变量加一并将其加入队列。工作线程将其出列并在控制台上写入。管理器在队列中添加项目时获取 queueLock 的锁，而工作人员在出队时获取 queueLock 的锁以确保并发性。在任何线程获得锁之前，其他线程都无法获得它。因此，经理必须等到工人出队，工人必须等待其他工人和经理。

using System;
    using System.Collections;
    using System.Threading;
    using System.Threading.Tasks;
    namespace ThreadSample
    {
        class Program
        {
            private static Queue queue=new Queue();
            private static object queueLock = new object();
            static void Main(string[] args)
            {
    
                Task manager = new Task(ProduceWork, TaskCreationOptions.LongRunning);
    
                Task worker1 = new Task(DoWork, TaskCreationOptions.LongRunning);
                Task worker2 = new Task(DoWork, TaskCreationOptions.LongRunning);
                Task worker3 = new Task(DoWork, TaskCreationOptions.LongRunning);
                Task worker4 = new Task(DoWork, TaskCreationOptions.LongRunning);
    
    
                manager.Start();
                worker1.Start();
                worker2.Start();
                worker3.Start();
                worker4.Start();
             
                Task.WaitAll(worker1, worker2, worker3, worker4, manager);
    
                Console.WriteLine("Program will never reach to this line");
            }
    
            private static void ProduceWork()
            {
                int counter = 0;
                while (true)
                {
                    Monitor.Enter(queueLock);
                    try
                    {
                        queue.Enqueue(counter++);
                    }
                    finally
                    {
                        Monitor.Exit(queueLock);
                    }
                }
            }
    
            private static void DoWork()
            {
                while (true)
                {
                    lock (queueLock)
                    {
                        var item = queue.Dequeue();
                        Console.Write("Counter value is {0}\r\n", item);
                    }
                }
            }
        }
    }

Monitor 提供了另外两种同步方法。例如，如果您希望经理在达到最大队列大小时停止生产新工作，并且工作人员在队列为空时立即停止工作。我们可以使用 Monitor.Wait 和 Monitor.Pulse 或 Monitor.PulseAll 来同步这两个事件。Monitor.Wait 将导致调用线程等到 queueLock 脉冲。一旦收到脉冲，它将继续进行。PulseAll 将脉冲所有等待该特定锁对象的线程，一旦调用 PulseAll，它们将继续。

以下是针对上述行为修改的 Manager 和 Workers 的示例。

using System;
    using System.Collections;
    using System.Threading;
    using System.Threading.Tasks;
    namespace ThreadSample
    {
        class Program
        {
            private static Queue queue=new Queue();
            private static object queueLock = new object();
            private static int maxSize = 10;
    
            static void Main(string[] args)
            {
    
                Task manager = new Task(ProduceWork, TaskCreationOptions.LongRunning);
    
                Task worker1 = new Task(DoWork, TaskCreationOptions.LongRunning);
                Task worker2 = new Task(DoWork, TaskCreationOptions.LongRunning);
                Task worker3 = new Task(DoWork, TaskCreationOptions.LongRunning);
                Task worker4 = new Task(DoWork, TaskCreationOptions.LongRunning);
    
    
                manager.Start();
                worker1.Start();
                worker2.Start();
                worker3.Start();
                worker4.Start();
             
                Task.WaitAll(worker1, worker2, worker3, worker4, manager);
    
                Console.WriteLine("Program will never reach to this line");
            }
    
            private static void ProduceWork()
            {
                int counter = 0;
                while (true)
                {
                    Monitor.Enter(queueLock);
                    try
                    {
                        if (queue.Count >= maxSize)
                        {
                            Monitor.Wait(queueLock);
                        }
    
                        queue.Enqueue(counter++);
    
                        if (queue.Count == 1)
                        {
                            Monitor.PulseAll(queueLock);
                        }
    
                    }
                    finally
                    {
                        Monitor.Exit(queueLock);
                    }
                }
            }
    
            private static void DoWork()
            {
                while (true)
                {
                    lock (queueLock)
                    {
                        while (queue.Count == 0)
                        {
                            Monitor.Wait(queueLock);
                        }
    
                        var item = queue.Dequeue();
    
                        if (queue.Count == maxSize - 1)
                        {
                            Monitor.PulseAll(queueLock);
                        }
    
                        Console.Write("Counter value is {0}\r\n", item);
                    }
                }
            }
        }
    }

#### 信号 ####

信号量限制了可以同时访问资源的线程数。它不像 Monitor 那样向任何线程授予对对象或区域的独占访问权限。因此，您必须将它与其他同步方法一起使用。

以下代码创建了一个最少 0 个成员和最多 3 个成员的信号量池。这意味着最多允许三个线程访问。默认

using System;
    using System.Collections;
    using System.Threading;
    using System.Threading.Tasks;
    namespace ThreadSample
    {
        class Program
        {
            private static Semaphore semaphore;
            private static int counter;
            static void Main(string[] args)
            {
                semaphore = new Semaphore(0, 3);
    
    
                Task task1 = new Task(RunTask, TaskCreationOptions.LongRunning);
                Task task2 = new Task(RunTask, TaskCreationOptions.LongRunning);
                Task task3 = new Task(RunTask, TaskCreationOptions.LongRunning);
                Task task4 = new Task(RunTask, TaskCreationOptions.LongRunning);
                Task task5 = new Task(RunTask, TaskCreationOptions.LongRunning);
    
    
    
                task1.Start();
                task2.Start();
                task3.Start();
                task4.Start();
                task5.Start();
    
                semaphore.Release();
    
    
                Task.WaitAll(task1, task2, task3, task4, task5);
    
                Console.WriteLine("Program will never reach to this line");
            }
    
            private static void RunTask()
            {
                while (true)
                {
                    semaphore.WaitOne();
    
                    Interlocked.Increment(ref counter);
                    Console.Write("Counter value is {0}\r\n", counter);
    
                    semaphore.Release();
                }
            }
        }
    }

#### 互斥体 ####

互斥锁是一种同步方法，可用于单个进程甚至跨多个进程的同步。与 Monitor 一样，Mutex 还提供对代码块或关键区域的独占访问。一次只有一个线程可以访问代码区域，所有其他线程都必须等待。Mutex 实现 IDisposable 并且必须被释放以清除资源。如果您在类中实现 Mutex，请记住实现 IDisposable 并调用 mutex.Dispose()

using System;
    using System.Collections;
    using System.Threading;
    using System.Threading.Tasks;
    namespace ThreadSample
    {
        class Program
        {
            private static Mutex mutex;
            private static int counter;
            static void Main(string[] args)
            {
                using (mutex = new Mutex())
                {
    
                    Task task1 = new Task(RunTask, TaskCreationOptions.LongRunning);
                    Task task2 = new Task(RunTask, TaskCreationOptions.LongRunning);
                    Task task3 = new Task(RunTask, TaskCreationOptions.LongRunning);
                    Task task4 = new Task(RunTask, TaskCreationOptions.LongRunning);
                    Task task5 = new Task(RunTask, TaskCreationOptions.LongRunning);
    
    
    
                    task1.Start();
                    task2.Start();
                    task3.Start();
                    task4.Start();
                    task5.Start();
    
    
                    Task.WaitAll(task1, task2, task3, task4, task5);
                }
    
                Console.WriteLine("Program will never reach to this line");
            }
    
            private static void RunTask()
            {
                while (true)
                {
                    mutex.WaitOne();
    
                    counter++;
                    Console.Write("Counter value is {0}\r\n", counter);
    
                    mutex.ReleaseMutex();
                }
            }
        }
    }

您可以通过为其提供唯一名称来获取操作系统级别互斥锁。如果 Mutex 被放弃，您将需要捕获 AbandonedMutexException 继续。

using (mutex = new Mutex(false,"My_Mutext_Unique_Name"))
    {
        var successfullyAcquired = false;
        try
        {
            successfullyAcquired = mutex.WaitOne(60000);
        }
        catch (AbandonedMutexException)
        {
           successfullyAcquired = true;
        }
    
        if (!successfullyAcquired)
        {
            return;
        }
    
        try
        {
            //Do some Work
        }
        finally
        {
            mutex.ReleaseMutex();
        }
    }

我试图涵盖使用 C\# 的多线程的所有基础知识。对于详细的研究，我强烈建议阅读每个的 MSDN 文档并深入研究每个同步方法。编码不当的多线程程序可能会造成灾难，并且在编写多线程应用程序之前始终确保您了解自己的概念。

如果我遗漏了一些重要的内容，请告诉我，我很乐意将其合并到这篇文章中。

快乐编码！

[MSDN]: https://docs.microsoft.com/en-us/dotnet/api/system.threading.interlocked?view=netframework-4.8