[原创][3]探究C#多线程开发细节-“用ConcurrentQueue＜T＞解决多线程的无顺序性的问题“

news2024/9/20 18:53:46

[简介]
常用网名: 猪头三
出生日期: 1981.XX.XX
QQ: 643439947
个人网站: 80x86汇编小站 https://www.x86asm.org
编程生涯: 2001年~至今[共22年]
职业生涯: 20年
开发语言: C/C++、80x86ASM、PHP、Perl、Objective-C、Object Pascal、C#、Python
开发工具: Visual Studio、Delphi、XCode、Eclipse、C++ Builder
技能种类: 逆向驱动磁盘文件
研发领域: Windows应用软件安全/Windows系统内核安全/Windows系统磁盘数据安全/macOS应用软件安全
项目经历: 磁盘性能优化/文件系统数据恢复/文件信息采集/敏感文件监测跟踪/网络安全检测

[序言]
经过上一篇文章([原创][2]探究C#多线程开发细节-“线程的无顺序性“-CSDN博客), 得知在不干预的情况下, 默认运行是无顺序的. 那么这样特性, 对程序的运行来说, 是好还是坏呢？其实无顺序没有好坏之说, 只跟程序功能的业务需求有关系. 当一个业务需求也可以说是功能, 需要多线程的无顺序特性, 那么在写代码的过程中就不要干预它. 如果业务需求对顺序有严格要求, 那么在编写多线程时, 就要适当得干预了.

[到底什么需求和场合需要控制多线程的运行顺序呢?]
这里举例一个最常见的场合: 比如有一个10G的文件, 程序创建了10个线程来读取内容, 每个线程分别依次读取1G内容(0号线程读取的范围是0~1G, 1号线程读取的范围1~2G, 2号线程读取的范围是2~3G, 依次类推), 然后每个线程读取内容完毕之后, 就在程序界面上显示.

[上面的场合, 如果不控制多线程的运行顺序时, 会发生什么现象呢?]
出现的现象就是: 假设当1号线程最先读取完1~2G范围的内容, 该线程就马上在界面显示. 当显示完成之后, 0号线程才完成读取0~1G范围的内容并在界面显示. 这样内容就错乱了.

这里把刚才抽象的描述实例化: 假设一个文件有10个字, "我爱CSDN编程网站". 每1个线程分别控制1个字的读取和显示, 形成如下关系:

0号线程读取并显示 "我"
1号线程读取并显示 "爱"
2号线程读取并显示 "C"
3号线程读取并显示 "S"
4号线程读取并显示 "D"
5号线程读取并显示 "N"
6号线程读取并显示 "编"
7号线程读取并显示 "成"
8号线程读取并显示 "网"
9号线程读取并显示 "站"

如果不加以干预的话, 界面上最终显示出来的内容, 有可能是 "编爱CDSN我网站程" 这样错乱的效果.

[那么如何解决这样的问题呢?]
遇到的这样问题, 就需要一个叫ConcurrentQueue<T>类，它是多线程安全的并提供先进先出(FIFO)的数据结构操作方法. 现在看看如何利用ConcurrentQueue<T>类来强制把无顺序的多线程改变成有顺序的.
1> 在使用for循环创建线程的时候, 同时也把线程的编号有序的存放到ConcurrentQueue<T>.
2> 假如当3号线程提前完成了文件内容的读取, 在返回给界面显示之前, 先判断2号线程是否完成, 如果2号线程没有完成, 那么3号线程就原地等待, 直到2号线程把内容读取完毕并在界面显示完成之后, 它才能在界面显示
步骤2是非常关键的逻辑, 这里仅仅是举例3号线程与2号线程的关系, 依次类推, 其他线程也有这样的等待关系.
3> 如何判断2号线程是否完成内容读取与内容显示?
很简单, 直接从ConcurrentQueue<T>的队列头部获取线程编号即可. 如果队列头部的编号为"3"，就表示"2"号线程已经完成所有工作任务并轮到3号线程开始干活了.

[下面是一套完整的代码]
这套代码是入门级的, 主要是多线程实验演示. 如果哪个朋友一眼看出, 有更好的代码实现, 说明你是高手了. 如果看不出来, 也不要紧. 后期还有更多的讲解并且释放出更优秀的多线程代码让大家学习.

   public partial class Form_Main : Form
   {

       private ConcurrentQueue<int> mpr_cq_ThreadIndex = new ConcurrentQueue<int>();


       public class Thread_Run
       {
           public int mpu_int_ThreadIndex;
           private Action<int> mpr_action_UpdateWaiteInfo;
           private ConcurrentQueue<int> mpr_cq_ThreadIndex;

           public Thread_Run(Action<int> action_param_UpdateWaiteInfo, ref ConcurrentQueue<int> cq_param_ThreadIndex)
           {
               mpr_action_UpdateWaiteInfo = action_param_UpdateWaiteInfo;
               mpr_cq_ThreadIndex = cq_param_ThreadIndex;
           }

           public int mpu_fun_ShowIndex()
           {
               return mpu_int_ThreadIndex;
           }

           public void mpu_pro_StartThread()
           {
               
               Thread class_Thread = new Thread(Thread_Exe);
               class_Thread.Start();
           }

           private void Thread_Exe()
           {
               int int_Dq_ThreadIndex;

               // 核心重点: N号线程等待N-1线程是否完成任务
               while (true)
               {
                   if (mpr_cq_ThreadIndex.TryPeek(out int_Dq_ThreadIndex) && int_Dq_ThreadIndex == mpu_int_ThreadIndex)
                   {
                       //调用委托方法来更新UI
                       mpr_action_UpdateWaiteInfo?.Invoke(mpu_int_ThreadIndex);

                       mpr_cq_ThreadIndex.TryDequeue(out _);

                       break;
                   
                   }

                   Thread.Sleep(5);
               }
           }
       }// End Thread_Run()


       public Form_Main()
       {
           InitializeComponent();
       }


       public void mpu_pro_UpdateWaiteInfo(int int_param_ThreadIndex)
       {

           if (InvokeRequired)
           {
               this.Invoke((MethodInvoker)delegate {

                   lb_WaitInfo.Text += (Environment.NewLine + string.Format("{0} 号线程已经跑到终点.", int_param_ThreadIndex));

               });
           }
       }


       private void Bn_StartThread_Click(object sender, EventArgs e)
       {

           // 启动10个线程
           for (int int_Index = 0; int_Index < 10; int_Index++)
           {

               mpr_cq_ThreadIndex.Enqueue(int_Index);

               Thread_Run class_ThreadRun = new Thread_Run(mpu_pro_UpdateWaiteInfo, ref mpr_cq_ThreadIndex);
               class_ThreadRun.mpu_int_ThreadIndex = int_Index;
               class_ThreadRun.mpu_pro_StartThread();
               
           }

       }

   }

[总结]
如果你们能按照源码把程序运行起来, 并且理解了本文章的内容, 那么是一个很大的进步. 这做这个代码实验的时候, 如遇到问题, 可在下面留言, 我会一一针对性的回复.

[程序截图]