专业网站建设品牌，十四年专业建站经验，服务6000+客户--广州京杭网络
免费热线：400-683-0016      微信咨询  |  联系我们

C#实现控制线程池最大数并发线程

当前位置：网站建设 > 技术支持
资料来源：网络整理       时间：2023/2/14 0:43:29       共计：3610 浏览

1. 实验目的：

      使用线程池的时候，有时候需要考虑服务器的最大线程数目和程序最快执行所有业务逻辑的取舍。
并非逻辑线程越多也好，而且新的逻辑线程必须会在线程池的等待队列中等待 ，直到线程池中工作的线程执行完毕，
才会有系统线程取出等待队列中的逻辑线程，进行CPU运算。

2.  解决问题：

     <a>如果不考虑服务器实际可支持的最大并行线程个数，程序不停往线程池申请新的逻辑线程，这个时候我们可以发现CPU的使用率会不断飙升，并且内存、网络带宽占用也会随着逻辑线程在CPU队列中堆积，而不断增大。

      如果我们想在主程序有200个http网络通讯需要执行，如何每次循环用10个线程并发处理10个网络http通讯回话，下一次循环只有在上一次循环的10个线程都执行完毕后才会执行下一次循环，并且主程序监听和等待200个http网络通讯都在CPU线程池中执行完毕后，才会退出主程序。

 3.  实现逻辑：

      我们通过两个AutoResetEvent和线程监听器Monitor，分别实现：

       <a>wait_sync:   任务线程的 并发执行，每次循环只处理最大10个线程分别对网络做http通讯回话。并且当前循环的10个线程都执行完毕后，才会进行下一次循环处理。
       wait_main: 主程序线程的监听和等待，只有所有任务线程都执行完毕后，主程序线程才会退出程序。
       <c> list_Thread: 负责记录每次循环，CPU实际分配的系统线程的个数。和Monitor配合使用，Monitor.Enter(list_Thread)=占用共享线程资源的占用锁，Monitor.Exit(list_Thread)释放共享线程资源的占用锁。  
       <d> n_total_thread: 配合wait_main使用，记录全部逻辑线程，已经执行完毕的当前总个数，用来判断主线程是否还需要继续等待，还是可以结束主程序运行。

4. 主要代码：

    <a> 线程池控制代码，如下： 

		
    
			
			
  1. 
				/// <summary>
			
    2. 

			
  2. 
				/// 多线程调用WCF
			
    3. 

			
  3. 
				/// </summary>
			
    4. 

			
  4. 
				/// <param name="select">调用WCF的方式，1=Restful，2=Tcp</param>
			
    5. 

			
  5. 
				/// <param name="num"></param>
			
    6. 

			
  6. 
				static void DoTest_MultiThread(string select, long num)
			
    7. 

			
  7. 
				{
			
    8. 

			
  8. 
				 int n_max_thread = 10; // 设置并行最大为10个线程
			
    9. 

			
  9. 
				 int n_total_thread = 0; // 用来控制：主程序的结束执行，当所有任务线程执行完毕
			
    10. 

			
  10. 
				 
			
    11. 

			
  11. 
				 ILog log_add = new LogHelper("Add_Thread");
			
    12. 

			
  12. 
				 ILog log_del = new LogHelper("Del_Thread");
			
    13. 

			
  13. 
				 ILog log_wait = new LogHelper("Wait_Thread");
			
    14. 

			
  14. 
				 ILog log_set = new LogHelper("Set_Thread");
			
    15. 

			
  15. 
				 ILog log_for = new LogHelper("For_Thread");
			
    16. 

			
  16. 
				 
			
    17. 

			
  17. 
				 Console.Title = string.Format("调用WCF的方式 => {0}, 调用次数=> {1}"
			
    18. 

			
  18. 
				 , select == "1" ? "Restful" : "Socket"
			
    19. 

			
  19. 
				 , num);
			
    20. 

			
  20. 
				 
			
    21. 

			
  21. 
				 List<int> list_Thread = new List<int>();
			
    22. 

			
  22. 
				 
			
    23. 

			
  23. 
				 System.Threading.AutoResetEvent wait_sync = new System.Threading.AutoResetEvent(false); // 用来控制：并发最大个数线程=n_max_thread
			
    24. 

			
  24. 
				 System.Threading.AutoResetEvent wait_main = new System.Threading.AutoResetEvent(false); // 用来控制：主程序的结束执行，当所有任务线程执行完毕
			
    25. 

			
  25. 
				 
			
    26. 

			
  26. 
				 DateTime date_step = DateTime.Now;
			
    27. 

			
  27. 
				 for (long i = 0; i < num; i++)
			
    28. 

			
  28. 
				 {
			
    29. 

			
  29. 
				 Num_Query_Static++;
			
    30. 

			
  30. 
				 if (i >0 && (i+1-1) % n_max_thread == 0) // -1 表示第max个线程尚未开始
			
    31. 

			
  31. 
				 {
			
    32. 

			
  32. 
				 //log_wait.Info(string.Format("thread n= {0},for i= {1}", dic_Thread.Count, i + 1));
			
    33. 

			
  33. 
				 wait_sync.WaitOne(); // 每次并发10个线程，等待处理完毕后，在发送下一次并发线程
			
    34. 

			
  34. 
				 }
			
    35. 

			
  35. 
				 log_for.Info(string.Format("thread n= {0},for i= {1}", list_Thread.Count, i + 1));
			
    36. 

			
  36. 
				 
			
    37. 

			
  37. 
				 System.Threading.ThreadPool.QueueUserWorkItem
			
    38. 

			
  38. 
				 ((data) =>
			
    39. 

			
  39. 
				 {
			
    40. 

			
  40. 
				 int id = System.Threading.Thread.CurrentThread.ManagedThreadId;
			
    41. 

			
  41. 
				 System.Threading.Monitor.Enter(list_Thread);
			
    42. 

			
  42. 
				 list_Thread.Add(id);
			
    43. 

			
  43. 
				 System.Threading.Monitor.Exit(list_Thread);
			
    44. 

			
  44. 
				 
			
    45. 

			
  45. 
				 log_add.Info(string.Format("id={0}, count={1}", id, list_Thread.Count)); // 日志
			
    46. 

			
  46. 
				 
			
    47. 

			
  47. 
				 if (select == "1") // Restful方式调用
			
    48. 

			
  48. 
				 {
			
    49. 

			
  49. 
				 Query_Htty();
			
    50. 

			
  50. 
				 }
			
    51. 

			
  51. 
				 else
			
    52. 

			
  52. 
				 {
			
    53. 

			
  53. 
				 Query_Socket();
			
    54. 

			
  54. 
				 }
			
    55. 

			
  55. 
				 
			
    56. 

			
  56. 
				 n_total_thread += 1;
			
    57. 

			
  57. 
				 if (list_Thread.Count == (n_max_thread) || n_total_thread == num)
			
    58. 

			
  58. 
				 {
			
    59. 

			
  59. 
				 list_Thread.Clear();
			
    60. 

			
  60. 
				 //log_set.Info(string.Format("thread n= {0},for i= {1}", dic_Thread.Count, i + 1));
			
    61. 

			
  61. 
				 //wait_sync.Set(); 
			
    62. 

			
  62. 
				 if (n_total_thread != num)
			
    63. 

			
  63. 
				 {
			
    64. 

			
  64. 
				 wait_sync.Set(); // 任务线程，继续执行
			
    65. 

			
  65. 
				 }
			
    66. 

			
  66. 
				 else
			
    67. 

			
  67. 
				 {
			
    68. 

			
  68. 
				 wait_main.Set(); // 主程序线程，继续执行
			
    69. 

			
  69. 
				 }
			
    70. 

			
  70. 
				 }
			
    71. 

			
  71. 
				 }, list_Thread);
			
    72. 

			
  72. 
				 }
			
    73. 

			
  73. 
				 
			
    74. 

			
  74. 
				 wait_main.WaitOne();
			
    75. 

			
  75. 
				 
			
    76. 

			
  76. 
				 Console.WriteLine(string.Format("总测试{0}次，总耗时{1}, 平均耗时{2}"
			
    77. 

			
  77. 
				 , num
			
    78. 

			
  78. 
				 , (DateTime.Now - date_step).ToString()
			
    79. 

			
  79. 
				 , (DateTime.Now - date_step).TotalMilliseconds / num));
			
    80. 

			
  80. 
				 
			
    81. 

			
  81. 
				 Query_Thread();
			
    82. 

			
  82. 
				}
			
    83.

WCF后台服务代码 

		
    
			
			
  1. 
				private static ILog log = new LogHelper("SeqService"); // 日志
			
    2. 

			
  2. 
				private static Dictionary<int, DateTime> dic_thread = new Dictionary<int, DateTime>(); // 线程列表
			
    3. 

			
  3. 
				 
			
    4. 

			
  4. 
				private static long Num = 0; // 线程个数
			
    5. 

			
  5. 
				private static object lock_Num = 0; // 共享数据-锁
			
    6. 

			
  6. 
				 
			
    7. 

			
  7. 
				/// <summary>
			
    8. 

			
  8. 
				/// 在线申请流水号
			
    9. 

			
  9. 
				/// </summary>
			
    10. 

			
  10. 
				/// <returns></returns>
			
    11. 

			
  11. 
				[WebGet(UriTemplate = "GetSeqNum/Json", ResponseFormat = WebMessageFormat.Json)]
			
    12. 

			
  12. 
				public string GetSeqNumber()
			
    13. 

			
  13. 
				{ 
			
    14. 

			
  14. 
				 lock (lock_Num)
			
    15. 

			
  15. 
				 {
			
    16. 

			
  16. 
				 Num++;
			
    17. 

			
  17. 
				 int id_thread = System.Threading.Thread.CurrentThread.ManagedThreadId;
			
    18. 

			
  18. 
				 DateTime now = DateTime.Now;
			
    19. 

			
  19. 
				 if (!dic_thread.TryGetValue(id_thread, out now))
			
    20. 

			
  20. 
				 {
			
    21. 

			
  21. 
				 dic_thread.Add(id_thread, DateTime.Now);
			
    22. 

			
  22. 
				 }
			
    23. 

			
  23. 
				 
			
    24. 

			
  24. 
				 }
			
    25. 

			
  25. 
				 string ret = DateTime.Now.ToString("yyyyMMdd") + Num.ToString(new string('0', 9)); 
			
    26. 

			
  26. 
				 
			
    27. 

			
  27. 
				 log.Info(string.Format("{0}, Thread={1}/{2}", ret, System.Threading.Thread.CurrentThread.ManagedThreadId, dic_thread.Count));
			
    28. 

			
  28. 
				 return ret;
			
    29. 

			
  29. 
				}
			
    30. 

			
  30. 
				 
			
    31. 

			
  31. 
				 
			
    32.

5.  实验结果

1. 10000个WCF网络http请求，CPU分成每次10个（10可以按需求调整）线程并发执行，并且主程序在所有请求都执行完毕后，才退出主程序。

1. 前端日志：LogFile\Add_Thread\Info

2. WCF日志：LogFile\SeqService\Info

本文标题: C#实现控制线程池最大数并发线程 本文地址: http://www.cppcns.com/ruanjian/csharp/158346.html
版权说明：
本网站凡注明“广州京杭 原创”的皆为本站原创文章，如需转载请注明出处！
本网转载皆注明出处，遵循行业规范，如发现作品内容版权或其它问题的，请与我们联系处理！
欢迎扫描右侧微信二维码与我们联系。
·上一条：详解C# Tuple VS ValueTuple（元组类 VS 值元组） | ·下一条：C#使用互斥量(Mutex)实现多进程并发操作时多进程间线程同步操作(进程同步)的简单示例代码及使用方法

Copyright © 广州京杭网络科技有限公司 2005-2024 版权所有    粤ICP备16019765号 

广州京杭网络科技有限公司 版权所有