【C#】并行编程实战：并行编程简介网站首页 学无止境

【C#】并行编程实战：并行编程简介

魔术师Dix 2024-10-18 00:01:02

简介【C#】并行编程实战：并行编程简介

本章内容为多线程编程入门知识，旨在介绍多线程的特点，以及提供了C#部分基础的多线程API使用。

1、进程与线程

这一小节包含大量概念和基础知识，虽然建议阅读但确实比较枯燥。

可以直接跳到后面的实际应用的章节。

进程

狭义定义：正在运行的程序示例。

就操作系统而言，进程是内存中的地址空间。进程为程序提供了安全性，在同一系统上分配给某一运行中程序的数据不会被另一程序意外访问。进程提供隔离，程序可以彼此独立，并在操作系统底层可以独立启动和停止。

进程就是程序的实体，是线程（Tread）存在和运行的地方。

进程是线程的容器。

多任务

多任务处理（Multitasking）：计算机系统一次运行多个进程（应用程序）的能力。与该系统中的内核数呈正比。一般单核处理器一次只能处理一个任务，双核两个，四核四个；但如果向其中添加CPU调度的概念，则CPU可以根据调度算法进行调度或切换进程，从而一次运行更多的应用程序。

超线程

超线程（Hyber-Threading，HT）技术由英特尔公司开发的专有技术，可改善在x86处理器上的计算并行化。支持超线程（HT）的单处理器芯片使用两个虚拟（逻辑）内核运行，并且能够一次执行两个任务。

Flynn分类法

根据指令流和数据流的数量将计算机结构进行分类。

单指令单数据（SISD：Single Instruciont Single Data )：一个控制单元一个指令流，系统一次只能运行一条指令，没有任何并行处理。适用于所有单核处理机。
单指令多数据（SIMD：Single Instruction Multiple Data）：一个指令流多个数据流，相同的指令流并行应用于多个数据流。（假设我们有多种算法，但不知道哪个更快，这种模型能方便测试出来：他为每个算法提供相同的输入，并在多个处理器上并行）
多指令单数据（MISD：Multiple Instruction Single Data）：多个指令对一个数据流进行操作。该模型通常运行于需要高容错的计算机（如航天飞机）中。
多指令多数据（MISD：Multiple Instruction Multiple Data）：多个指令流多个数据流，可以实现真正的并行性，每个处理器都可以在不同数据流上运行不同指令。如今大多数计算机都是采用这种体系结构。

线程

线程是进程内部的执行单元。

一个程序可能包含一个或多个线程，以提高性能。

前台线程（Foregroud Thread）：直接影响应用程序的生命周期。只要有一个前台线程，应用程序就会一直运行。
后台线程（Backgroud Thread）：对应用程序的生命周期没有影响。当应用程序退出时，所有的后台线程都会被杀死。

应用程序可以包含任意数量的前台线程或后台线程。

线程单元状态

线程单元状态（Apartment State）：是线程内组件对象模型（Component Object Model，COM）对象所驻留的区域。

ApartmentState是一个枚举变量，线程可以属于以下状态中的一个：

namespace System.Threading
{
    public enum ApartmentState
    {
        STA = 0,//单线程单元（Single-Thread Apartment）：只能通过单线程访问底层COM对象。
        MTA = 1,//多线程单元（Multi-Thread Apartment）：一次可以通过多个线程访问底层COM对象。
        Unknown = 2
    }
}

线程单元状态的要点：

进程可以具有多个线程，无论是前台还是后台。
每个线程都有一个单元状态（STA、MTA）。
每个单元都有一个并发模型（即单线程或多线程）。
可以通过编程改变线程状态。
一个应用进程可能具有多个STA，但最多只有一个MTA。
STA应用程序示例：Windows应用程序。
MTA应用程序示例：Web应用程序。
COM对象是在单元中被创建的。
一个COM对象只能驻留在一个线程单元中，并且单元是不能共享的。

并发与并行

并发：逻辑上同时发生。
并行：物理上同时发生。
并发不一定不行，并行一定并发。
并行一般只有在多核计算机上，多个核心同时处理并发任务才符合。即便有超线程技术，在严格意义的物理上依旧不算并行。

2、Thread

下面开始直接用代码进行实践！

随意写了两个简单的测试代码：

        /// <summary>
        /// 测试方法1：不带参数
        /// </summary>
        public static void LoopAddNumber()
        {
            int length = 100;
            LoopAddNumberWithParameter(length);
        }

        /// <summary>
        /// 测试方法2：带参数
        /// </summary>
        /// <param name="param">因为多线程API的缘故，这里只支持传入 object 类型</param>
        public static void LoopAddNumberWithParameter(object param)
        {
            //将参数转换成Int32
            int length = Convert.ToInt32(param);
            Debug.Log($"LoopAddNumber Start : {length}");
            int result = 0;
            for (int i = 0; i < length; i++)
            {
                result += 5;
            }
            Debug.Log($"LoopAddNumber Finish : {result}");
        }

首先，作为对照，我们直接在Unity调用 LoopAddNumber（不进行多线程操作）：

        public void RunTestFuncion_1()
        {
            Debug.Log("RunTestFuncion_1 Start !");
            TestFunction.LoopAddNumber();
            Debug.Log("RunTestFuncion_1 End !");
        }

这个结果很显然了：

主线程运行就是正常地按照代码顺序依次执行：

2.1、使用 Thread 进行多线程

下面使用无参数线程来运行上述程序：

        public void RunTestFuncion_2()
        {
            Debug.Log("RunTestFuncion_2 Start !");

            //使用无参数线程
            Thread thread = new Thread(new ThreadStart(TestFunction.LoopAddNumber));
            thread.Start();

            Debug.Log("RunTestFuncion_2 End !");
        }

这次结果有所不同：

可以看到先是执行了 Start，然后是 End ，最后才是 LoopAddNumber 完成。

此时，LoopAddNumber 的方法就已经在其他线程里执行了。比较令我意外的是，Unity的 Debug居然能直接在子线程使用。因为我印象中凡是 UnityEngine 的类只能在主线程调用，否则会报错。可能 Debug 是一个比较特殊的类吧（要看是否可以多线程，要看Unity原生方法里是否有 ThreadAndSerializationSafe 宏）。

2.2、使用带参数的多线程

直接看代码，变化不大：

        public void RunTestFuncion_3()
        {
            Debug.Log("RunTestFuncion_3 Start !");

            //使用带参数线程
            Thread thread = new Thread(new ParameterizedThreadStart(TestFunction.LoopAddNumberWithParameter));
            thread.Start(20230214);

            Debug.Log("RunTestFuncion_3 End !");
        }

运行结果如下：

和无参数的运行效果是一样的，但是可以看到我们已经将参数成功传递进去了。

3、ThreadPool

就内存和CPU而言，创建线程是一项昂贵的操作，有时甚至会降低应用程序的性能。创建多少线程是由硬件水平决定的，在某个设备上的最佳性能数量在另一个系统上可能反而更糟。

要找到最佳线程数，除了程序员自己调试外，还可以将其留给共用语言运行时（Common Language Runtime ，CLR）。CLR有一种算法，可以根据任何时间点的CPU负载确定最佳线程数量。CLR会维护一个线程池（Thread Pool）。每个进程都有其自己的线程池。线程池的最大数量由可用物理资源的数量决定。

以下是不同框架在 ThreadPool 内可以创建的最佳线程数：

.NET Framework 2.0 ：每个核心 25 个线程。
.NET Framework 3.5 ：每个核心 250 个线程。
32bit 环境的 .NET Framework 4.0 ：每个核心 1023 个线程。
64bit 环境的 .NET Framework 4.0 及更高版本：每个核心 32768 个线程。

3.1、使用 ThreadPool 进行多线程

这里我们先在TestFuntion中补充一个测试代码：

        /// <summary>
        /// 测试方法3：打印线程池信息
        /// </summary>
        /// <param name="state"></param>
        public static void DebugThreadPoolInfo(object state)
        {
            Debug.LogWarning("state : "+state);

            int workerThreads=-1;//工作线程数
            int completionPortThreads=-1;//完成端口线程数

            //剩余空闲线程数
            ThreadPool.GetAvailableThreads(out workerThreads, out completionPortThreads);
            Debug.LogWarning($"GetAvailableThreads | workerThreads : {workerThreads} completionPortThreads : {completionPortThreads}");

            //检索线程池在新请求预测中维护的空闲线程数，最少保留的线程数
            ThreadPool.GetMinThreads(out workerThreads, out completionPortThreads);
            Debug.LogWarning($"GetMinThreads | workerThreads : {workerThreads} completionPortThreads : {completionPortThreads}");

            //最多可用线程数，所有大于此数目的请求将保持排队状态，直到线程池线程变为可用
            ThreadPool.GetMaxThreads(out workerThreads, out completionPortThreads);
            Debug.LogWarning($"GetMaxThreads | workerThreads : {workerThreads} completionPortThreads : {completionPortThreads}");
        }

为方便查看结果，这里我把打印改为了LogWarning。

之后我们使用线程池来运行：

        public void RunTestFuncion_4()
        {
            Debug.Log("RunTestFuncion_4 Start !");

            //使用线程池
            ThreadPool.QueueUserWorkItem(new WaitCallback(TestFunction.LoopAddNumberWithParameter), 20230215);
            ThreadPool.QueueUserWorkItem(new WaitCallback(TestFunction.DebugThreadPoolInfo));

            Debug.Log("RunTestFuncion_4 End !");
        }

结果如下：

3.2、使用 ThreadPool 的优缺点

优点：

线程可以用于释放主线程
可以通过CLR以最佳方式创建和维护线程

缺点：

随着线程的增多，代码变得难以维护和调试。
程序员需要在Worker方法中进行异常处理，因为任何未处理的的异常都可能导致程序崩溃。
进度报告、取消和完成逻辑需要从头开始编写。

以下情况需要避免使用 ThreadPool：

需要前台线程。
需要为线程设置显示优先级。
长时间运行或阻塞的任务。（由于 ThreaPool 中每个进程可用的线程数有限，因此池中有大量阻塞的线程将阻止启动新任务）
需要STA线程时。（ThreadPool 线程默认为 MTA）
需要通过一个独特身份线程专用于任务时。（ThreadPool 线程无法命名）

4、BackgroundWorker

BackgroundWorker 是 .NET 提供的一种结构，用于从 ThreadPool 中创建更多可用管理的线程。除了可以通知操作结果，还支持进度报告和取消。

4.1、使用 BackgroundWorker 进行多线程

首先我们在TestFuntion中写如下三个方法：

        /// <summary>
        /// 测试方法：长时间循环打印
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        public static void LongTimeWaitting(object sender, DoWorkEventArgs e)
        {
            var worker = sender as BackgroundWorker;
            int length = Convert.ToInt32(e.Argument);
            Debug.Log($"LongTimeWaitting Start  => {length}");
            for (int i = 0; i < length; i++)
            {
                //等待1s
                Thread.Sleep(1000);
                float progress = (i + 1) / (float)length;
                int reportProgress = (int)(progress * 100);//转换成百分比
                Debug.Log($"LongTimeWaitting:  {i + 1}/{length} | 进度 {reportProgress} %");
                worker.ReportProgress(reportProgress);
            }
        }

        /// <summary>
        /// 测试方法：获取到线程进度报告
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        public static void OnGetProgressReporter(object sender, ProgressChangedEventArgs e)
        {
            //var worker = sender as BackgroundWorker;
            Debug.Log($"报告进度 {e.ProgressPercentage} %");
        }

        /// <summary>
        /// 测试方法：线程任务完成
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        public static void OnWorkerCompelete(object sender, RunWorkerCompletedEventArgs e)
        {
            //var worker = sender as BackgroundWorker;

            if (e.Error == null)
                Debug.Log($"{sender} 完成，且无错误");
            else
                Debug.LogError($"{sender} 异常：{e.Error.Message}");
        }

这里我想通过一个方法实现启动和取消，即第一触发时启动，第二次触发时取消：

        public void RunTestFuncion_5()
        {
            //运行完成之后直接释放；
            if (backgroundWorker != null && !backgroundWorker.IsBusy)
            {
                backgroundWorker.Dispose();
                backgroundWorker = null;
            }

            //这里做一个分支：如果没有线程则创建一个，如果已有则手动停止此线程
            if (backgroundWorker == null)
            {
                Debug.Log("RunTestFuncion_5 Start !");

                backgroundWorker = new BackgroundWorker();
                backgroundWorker.WorkerReportsProgress = true;//是否有进度报告
                backgroundWorker.WorkerSupportsCancellation = true;//是否支持取消线程
                backgroundWorker.DoWork += TestFunction.LongTimeWaitting;//工作方法
                backgroundWorker.ProgressChanged += TestFunction.OnGetProgressReporter;//进度报告回调
                backgroundWorker.RunWorkerCompleted += TestFunction.OnWorkerCompelete;//线程结束回调
                backgroundWorker.RunWorkerAsync(100);//带参数的调用

                Debug.Log("RunTestFuncion_5 End !");
            }
            else
            {
                backgroundWorker.CancelAsync();
                backgroundWorker.Dispose();
                backgroundWorker = null;
                Debug.Log("取消工作线程！");
            }
        }

之后我们开始运行这一段代码，得到如下输出：

这里看没什么问题哈，已经按照我们的设想进行一个长耗时后台线程并且能进行输出了，完成时也能得到回调。但是，不对劲的地方出现了：取消的方法失效了！

backgroundWorker.CancelAsync() 并不能取消线程，他还是会正常运行，即便调用了 Dispose 也不行！而如果要正确取消，需要我们自己在方法里判定取消时机：

        public static void LongTimeWaitting(object sender, DoWorkEventArgs e)
        {
            var worker = sender as BackgroundWorker;
            int length = Convert.ToInt32(e.Argument);
            Debug.Log($"LongTimeWaitting Start  => {length}");
            for (int i = 0; i < length; i++)
            {
                //查看是否有取消
                if (worker.CancellationPending)
                {
                    Debug.Log($"已经手动取消线程！");
                    return;
                }

                //等待1s
                Thread.Sleep(1000);
                float progress = (i + 1) / (float)length;
                int reportProgress = (int)(progress * 100);//转换成百分比
                Debug.Log($"LongTimeWaitting:  {i + 1}/{length} | 进度 {reportProgress} %");
                worker.ReportProgress(reportProgress);//百分比进度
            }
        }

也就是说，需要 LongTimeWaitting 方法自己轮询 worker.CancellationPending 的值，并手动取消线程才能生效。而 CancelAsync 方法实际上只是把 CancellationPending 设置为 true 而已，本身没有任何功能！

总之，需要注意的点有以下几个：