C# 执行固定个数任务自行控制进入线程池的线程数量，多任务同时但是并发数据限定

Posted 2020-10-04 一起来学python

思路来源：http://bbs.csdn.NET/topics/390819824，引用该页面某网友提供的方法。

题目：我现在有100个任务，需要多线程去完成，但是要限定同时并发数量不能超过5个。

原理：初始启用5个线程，然后让线程中的过程执行完毕之后，自己去取下一个任务，启动下一个线程。

 

[csharp] view plain copy

print?

1. public class MyTaskList  
2. {  
3.     public List<Action> Tasks = new List<Action>();  
4.   
5.     public void Start()  
6.     {  
7.         for (var i = 0; i < 5; i++)  
8.             StartAsync();  
9.     }  
10.   
11.     public event Action Completed;  
12.   
13.     public void StartAsync()  
14.     {  
15.         lock (Tasks)  
16.         {  
17.             if (Tasks.Count > 0)  
18.             {  
19.                 var t = Tasks[Tasks.Count - 1];  
20.                 Tasks.Remove(t);  
21.                 ThreadPool.QueueUserWorkItem(h =>  
22.                 {  
23.                     t();  
24.                     StartAsync();  
25.                 });  
26.             }  
27.             else if (Completed != null)  
28.                 Completed();  
29.         }  
30.     }  
31. }  

    public class MyTaskList
    {
        public List<Action> Tasks = new List<Action>();
 
        public void Start()
        {
            for (var i = 0; i < 5; i++)
                StartAsync();
        }
 
        public event Action Completed;
 
        public void StartAsync()
        {
            lock (Tasks)
            {
                if (Tasks.Count > 0)
                {
                    var t = Tasks[Tasks.Count - 1];
                    Tasks.Remove(t);
                    ThreadPool.QueueUserWorkItem(h =>
                    {
                        t();
                        StartAsync();
                    });
                }
                else if (Completed != null)
                    Completed();
            }
        }
    }

调用方式：

 

1，自动加入的100个测试任务，每一个运行时间都是不定的、随机的。

2，这里的 StartAsync 方法瞬间就结束了，根本不会阻塞，也不会等待什么 while 循环结束

[csharp] view plain copy

print?

1. var rnd = new Random();  
2. var lst = new MyTaskList();  
3. for (var i = 0; i < 100; i++)  
4. {  
5.     var s = rnd.Next(10);  
6.     var j = i;  
7.     var 测试任务 = new Action(() =>  
8.     {  
9.         Console.WriteLine(string.Format("第{0}个任务（用时{1}秒）已经开始", j, s));  
10.         Thread.Sleep(s * 1000);  
11.         Console.WriteLine(string.Format("第{0}个任务（用时{1}秒）已经结束", j, s));  
12.     });  
13.     lst.Tasks.Add(测试任务);  
14. }  
15. lst.Completed += () => Console.WriteLine("____________________没有更多的任务了！");  
16. lst.Start();  

            var rnd = new Random();
            var lst = new MyTaskList();
            for (var i = 0; i < 100; i++)
            {
                var s = rnd.Next(10);
                var j = i;
                var 测试任务 = new Action(() =>
                {
                    Console.WriteLine(string.Format("第{0}个任务（用时{1}秒）已经开始", j, s));
                    Thread.Sleep(s * 1000);
                    Console.WriteLine(string.Format("第{0}个任务（用时{1}秒）已经结束", j, s));
                });
                lst.Tasks.Add(测试任务);
            }
            lst.Completed += () => Console.WriteLine("____________________没有更多的任务了！");
            lst.Start();

 

自己使用到程序中，自定义最大线程数，然后循环启用最大线程数个线程执行任务，等待有线程完成，退出本次运行方法和线程前，再次调用启用线程方法调用下一个线程，依次循环，直至完成：

 

[csharp] view plain copy

print?

1. AutoResetEvent[] waitEnents = new AutoResetEvent[n_max_thread];  
2. for (int i = 0; i < n_max_thread; i++)  
3. {  
4.      calcState.wait_event = waitEnents[i] = new AutoResetEvent(false);  
5.   
6.      StartAsync(calcState);  
7. }  
8.   
9. AutoResetEvent.WaitAll(waitEnents);  
10.   
11. private static void StartAsync(CalcState calcState)  
12. {  
13.     lock (calcState.locker_ListFormula)  
14.     {  
15.         if (calcState.lstFormula.Count > 0)  
16.         {  
17.             calcState.formulaAttr = calcState.lstFormula.ElementAt(calcState.lstFormula.Count - 1).Value;  
18.             calcState.lstFormula.Remove(calcState.lstFormula.ElementAt(calcState.lstFormula.Count - 1).Key);  
19.             ThreadPool.QueueUserWorkItem(new WaitCallback(ExecAsync), calcState);  
20.         }  
21.         else  
22.         {  
23.             calcState.wait_event.Set();  
24.         }  
25.     }  
26.   
27. }  
28.   
29. private static void ExecAsync(object obj)  
30. {  
31.     CalcState calcState = obj as CalcState;  
32.     startCalcFormula(calcState);  
33.     //递归处理下一个公式  
34.     StartAsync(calcState);  
35. }  
36.   
37. private static void startCalcFormula(CalcState calcState)  
38. {  
39.   
40. }  

        AutoResetEvent[] waitEnents = new AutoResetEvent[n_max_thread];
        for (int i = 0; i < n_max_thread; i++)
        {
             calcState.wait_event = waitEnents[i] = new AutoResetEvent(false);

             StartAsync(calcState);
        }

        AutoResetEvent.WaitAll(waitEnents);

        private static void StartAsync(CalcState calcState)
        {
            lock (calcState.locker_ListFormula)
            {
                if (calcState.lstFormula.Count > 0)
                {
                    calcState.formulaAttr = calcState.lstFormula.ElementAt(calcState.lstFormula.Count - 1).Value;
                    calcState.lstFormula.Remove(calcState.lstFormula.ElementAt(calcState.lstFormula.Count - 1).Key);
                    ThreadPool.QueueUserWorkItem(new WaitCallback(ExecAsync), calcState);
                }
                else
                {
                    calcState.wait_event.Set();
                }
            }

        }

        private static void ExecAsync(object obj)
        {
            CalcState calcState = obj as CalcState;
            startCalcFormula(calcState);
            //递归处理下一个公式
            StartAsync(calcState);
        }

        private static void startCalcFormula(CalcState calcState)
        {

        }