C# 简单实现线程池

NET 6 环境开发 实现 线程数量，任务队列，非核心线程，及核心线程活跃时间的管理。

1. namespace CustomThreadPool;
2. /// <summary>
3. /// 线程池类
4. /// </summary>
5. public class ThreadPoolExecutor
6. {
7.     /// <summary>
8.     /// 核心线程的任务队列
9.     /// </summary>
10.     private readonly Queue<WorkTask> tasks = new Queue<WorkTask>();
11.     /// <summary>
12.     /// 最大核心线程数
13.     /// </summary>
14.     private int coreThreadCount;
15.     /// <summary>
16.     /// 最大非核心线程数
17.     /// </summary>
18.     private int noneCoreThreadCount;
19.     /// <summary>
20.     /// 当前运行的核心线程的数量
21.     /// </summary>
22.     private int runCoreThreadCount;
23.     /// <summary>
24.     /// 当前运行的非核心线程的数量
25.     /// </summary>
26.     private int runNoneCoreThreadCount;
27.     /// <summary>
28.     /// 核心线程队列的最大数
29.     /// </summary>
30.     private int maxQueueCount;
31.     /// <summary>
32.     /// 当核心线程空闲时最大活跃时间
33.     /// </summary>
34.     private int keepAliveTimeout;
35.     /// <summary>
36.     /// 设置是否为后台线程
37.     /// </summary>
38.     private bool isBackground;
39.     private ThreadPoolExecutor() { }
40.     /// <summary>
41.     ///
42.     /// </summary>
43.     /// <param name="CoreThreadCount">核心线程数</param>
44.     /// <param name="TotalThreadCount">总线程数</param>
45.     /// <param name="IsBackground">是否为后台线程</param>
46.     /// <param name="QueueCount">核心队列的最大数</param>
47.     /// <param name="KeepAliveTimeout">当核心线程空闲时最大活跃时间</param>
48.     /// <exception cref="ArgumentOutOfRangeException"></exception>
49.     /// <exception cref="ArgumentException"></exception>
50.     public ThreadPoolExecutor(int CoreThreadCount = 5, int TotalThreadCount = 10, bool IsBackground = true, int QueueCount = 200, int KeepAliveTimeout = 0)
51.     {
52.         if (CoreThreadCount < 1) throw new ArgumentOutOfRangeException(nameof(CoreThreadCount), CoreThreadCount, null);
53.         if (TotalThreadCount < CoreThreadCount) throw new ArgumentException($"{nameof(TotalThreadCount)}:{TotalThreadCount} must be greater than {nameof(CoreThreadCount)}:{CoreThreadCount}");
54.         if (QueueCount < 0) throw new ArgumentOutOfRangeException(nameof(QueueCount), QueueCount, null);
55.         if (KeepAliveTimeout < 0) throw new ArgumentOutOfRangeException(nameof(KeepAliveTimeout), KeepAliveTimeout, null);
56.         coreThreadCount = CoreThreadCount;
57.         noneCoreThreadCount = TotalThreadCount - CoreThreadCount;
58.         keepAliveTimeout = KeepAliveTimeout;
59.         maxQueueCount = QueueCount;
60.         isBackground = IsBackground;
61.     }
62.     /// <summary>
63.     /// 执行任务
64.     /// </summary>
65.     /// <param name="task">一个自定义任务</param>
66.     /// <exception cref="ArgumentNullException">任务为null时，抛出该错误</exception>
67.     /// <exception cref="NotSupportedException">当核心任务队列已满且非核心线程最大数为0时抛出该错误</exception>
68.     public void QueueTask(WorkTask task)
69.     {
70.         if (task == null) throw new ArgumentNullException(nameof(task));
71.         lock (tasks)
72.         {
73.             tasks.Enqueue(task);
74.             if (tasks.Count <= maxQueueCount)
75.             {
76.                 if (runCoreThreadCount < coreThreadCount)
77.                 {
78.                     ++runCoreThreadCount;
79.                     Run(true);
80.                 }
81.             }
82.             else
83.             {
84.                 if (noneCoreThreadCount > 0 && runNoneCoreThreadCount < noneCoreThreadCount)
85.                 {
86.                     ++runNoneCoreThreadCount;
87.                     Run(false);
88.                 }
89.             }
90.         }
91.     }
93.     private void Run(bool isCore)
94.     {
95.         Tuple<int, bool> state = new(keepAliveTimeout, isCore);
96.         Thread thread = new(t => Excute(t))
97.         {
98.             Name = Guid.NewGuid().ToString("D"),
99.             IsBackground = isBackground
100.         };
101.         thread.Start(state);
102.     }
104.     private void Excute(object? state)
105.     {
106.         if (state == null) return;
107.         var parameter = (Tuple<int, bool>)state;
108.         bool first = true;
109.         DateTime firstTime = DateTime.Now;
110.         while (true)
111.         {
112.             WorkTask? item = null;
113.             lock (tasks)
114.             {
115.                 if (tasks.Count > 0)
116.                 {
117.                     first = true;
118.                     item = tasks.Dequeue();
119.                 }
120.                 else
121.                 {
122.                     if (parameter.Item2)
123.                     {
125.                         if (first)
126.                         {
127.                             firstTime = DateTime.Now;
128.                             first = false;
129.                         }
130.                         if ((DateTime.Now - firstTime).TotalMilliseconds > parameter.Item1)
131.                         {
132.                             --runCoreThreadCount;
133.                             break;
134.                         }
135.                     }
136.                     else
137.                     {
138.                         --runNoneCoreThreadCount;
139.                         break;
140.                     }
141.                 }
142.             }
143.             item?.Runsynchronous();
144.         }
145.     }
146. }

复制代码

1. namespace CustomThreadPool;
2. /// <summary>
3. /// 包装的任务类
4. /// </summary>
5. public class WorkTask
6. {
7.     public static WorkTaskFactory Factory { get; private set; } = new WorkTaskFactory();
8.     /// <summary>
9.     /// 任务运行结束时触发该事件
10.     /// </summary>
11.     public event Action<WorkTask>? TaskCompleted;
12.     /// <summary>
13.     /// 任务ID
14.     /// </summary>
15.     private static int _id = 0;
16.     /// <summary>
17.     /// 委托给任务不带执行参数的代码
18.     /// </summary>
19.     private readonly Action? action;
20.     /// <summary>
21.     /// 委托给任务执行的带输入参数代码
22.     /// </summary>
23.     private readonly Action<object?>? actionWithParamter;
24.     /// <summary>
25.     /// 线程间的同步事件
26.     /// </summary>
27.     public AutoResetEvent WaitHandle { get; protected set; } = new AutoResetEvent(false);
28.     /// <summary>
29.     /// 执行代码的参数
30.     /// </summary>
31.     public object? State { get; protected set; }
32.     /// <summary>
33.     /// 接收任务抛出的异常
34.     /// </summary>
35.     public WorkTaskException? Exception { get; protected set; }
36.     /// <summary>
37.     /// 任务是否完成标志
38.     /// </summary>
39.     public bool IsCompleted { get; protected set; } = false;
40.     /// <summary>
41.     /// 任务知否有异常
42.     /// </summary>
43.     public bool IsFaulted { get; protected set; } = false;
44.     /// <summary>
45.     /// 任务状态
46.     /// </summary>
47.     public WorkTaskStatus Status { get; protected set; } = WorkTaskStatus.Created;
48.     public int Id { get { return Interlocked.Increment(ref _id); } }
49.     protected WorkTask() { }
51.     protected void OnTaskCompleted(WorkTask sender)
52.     {
53.         TaskCompleted?.Invoke(sender);
54.     }
55.     public WorkTask(Action action)
56.     {
57.         this.action = action ?? throw new ArgumentNullException(nameof(action));
58.     }
59.     public WorkTask(Action<object?> action, object state)
60.     {
61.         actionWithParamter = action ?? throw new ArgumentNullException(nameof(action));
62.         this.State = state;
63.     }
64.     /// <summary>
65.     /// 任务的同步方法
66.     /// </summary>
67.     public virtual void Runsynchronous()
68.     {
69.         if (Status != WorkTaskStatus.Created) return;
70.         Status = WorkTaskStatus.Running;
71.         try
72.         {
73.             action?.Invoke();
74.             actionWithParamter?.Invoke(State);
75.         }
76.         catch (Exception ex)
77.         {
78.             Exception = new WorkTaskException(ex.Message, ex);
79.             IsFaulted = true;
80.         }
81.         finally
82.         {
83.             OnTaskCompleted(this);
84.             WaitHandle.Set();
85.             IsCompleted = true;
86.             Status = WorkTaskStatus.RanToCompleted;
87.         }
88.     }
89.     /// <summary>
90.     /// 通过调用线程执行的方法
91.     /// </summary>
92.     public void Start()
93.     {
94.         Factory.ThreadPoolExcutor?.QueueTask(this);
95.     }
96.     /// <summary>
97.     /// 通过调用线程执行的方法
98.     /// </summary>
99.     /// <param name="executor">线程池管理类</param>
100.     public void Start(ThreadPoolExecutor executor)
101.     {
102.         executor.QueueTask(this);
103.     }
104.     /// <summary>
105.     /// 执行一组任务并等待所有任务完成。
106.     /// </summary>
107.     /// <param name="tasks">一组任务</param>
108.     /// <returns>所有任务是否都接收到完成的信号。</returns>
109.     public static bool WaitAll(WorkTask[] tasks)
110.     {
111.         var result = true;
112.         foreach (var task in tasks)
113.         {
114.             result = result && task.WaitHandle.WaitOne();
115.         }
116.         return result;
117.     }
119.     /// <summary>
120.     /// 执行一组任务并等待任意一个任务完成。
121.     /// </summary>
122.     /// <param name="tasks">一组任务</param>
123.     /// <returns>返回已完成任务的索引</returns>
124.     public static int WaitAny(WorkTask[] tasks)
125.     {
126.         var index = new Random().Next(0, tasks.Length - 1);
127.         tasks[index].WaitHandle.WaitOne();
128.         return index;
129.     }
130. }
131. /// <summary>
132. /// 具有返回类型的任务
133. /// </summary>
134. /// <typeparam name="TResult"></typeparam>
135. public class WorkTask<TResult> : WorkTask
136. {
137.     private readonly Func<TResult>? func;
138.     private readonly Func<object?, TResult>? funcWithParameter;
140.     protected TResult? _result = default(TResult);
141.     public TResult? Result
142.     {
143.         get
144.         {
145.             if (!isSetSignal)
146.                 WaitHandle.WaitOne();
147.             return _result;
148.         }
149.     }
150.     public WorkTask(Func<TResult> func)
151.     {
152.         this.func = func ?? throw new ArgumentNullException(nameof(func));
153.     }
154.     public WorkTask(Func<object?, TResult> func, object? state)
155.     {
156.         this.funcWithParameter = func ?? throw new ArgumentNullException(nameof(func));
157.         this.State = state;
158.     }
160.     private bool isSetSignal = false;
161.     public override void Runsynchronous()
162.     {
163.         if (Status != WorkTaskStatus.Created) return;
164.         Status = WorkTaskStatus.Running;
165.         try
166.         {
167.             if (func != null) _result = func();
168.             if (funcWithParameter != null) _result = funcWithParameter(State);
169.         }
170.         catch (Exception ex)
171.         {
172.             Exception = new WorkTaskException(ex.Message, ex);
173.             IsFaulted = true;
174.         }
175.         finally
176.         {
177.             OnTaskCompleted(this);
178.             isSetSignal = WaitHandle.Set();
179.             Status = WorkTaskStatus.RanToCompleted;
180.             IsCompleted = true;
182.         }
183.     }
184. }
185. public class WorkTaskException : Exception
186. {
187.     public WorkTaskException()
188.     {
189.     }
190.     public WorkTaskException(string Message)
191.         : base(Message)
192.     {
193.     }
195.     public WorkTaskException(string Message, Exception InnerException)
196.         : base(Message, InnerException)
197.     {
198.     }
199. }
201. public enum WorkTaskStatus
202. {
203.     /// <summary>
204.     /// 已创建
205.     /// </summary>
206.     Created = 0,
207.     /// <summary>
208.     /// 正在运行
209.     /// </summary>
210.     Running = 1,
211.     /// <summary>
212.     /// 已完成
213.     /// </summary>
214.     RanToCompleted = 2,
215. }

复制代码

1. namespace CustomThreadPool;
3. public class WorkTaskFactory
4. {
5.     public ThreadPoolExecutor? ThreadPoolExcutor { get; private set; }
6.     public WorkTaskFactory(ThreadPoolExecutor excutor)
7.     {
8.         ThreadPoolExcutor = excutor;
9.     }
11.     public WorkTaskFactory()
12.     : this(new ThreadPoolExecutor(5, 10))
13.     {
14.     }
16.     public WorkTask StartNew(Action action, ThreadPoolExecutor? executor = null)
17.     {
18.         WorkTask task = new WorkTask(action);
19.         ThreadPoolExcutor = executor ?? ThreadPoolExcutor;
20.         ThreadPoolExcutor?.QueueTask(task);
21.         return task;
22.     }
24.     public WorkTask<TResult> StartNew<TResult>(Func<object?, TResult> func, object? state, ThreadPoolExecutor? executor = null)
25.     {
26.         WorkTask<TResult> task = new WorkTask<TResult>(func, state);
27.         ThreadPoolExcutor = executor ?? ThreadPoolExcutor;
28.         ThreadPoolExcutor?.QueueTask(task);
29.         return task;
30.     }
31. }

复制代码

1. namespace CustomThreadPool;
2. using System.Threading;
3. using System.Text;
4. using System;
5. using System.Diagnostics;
6. using System.Reflection.Emit;
8. class Program
9. {
10.     static void Main(string[] args)
11.     {
12.         int count = 5;
13.         ThreadPoolExecutor poolExcutor = new(5, 6, QueueCount: 5, KeepAliveTimeout: 2000);
14.         WorkTask<int?>[] workTasks = new WorkTask<int?>[count];
15.         for (int i = 0; i < count; i++) workTasks[i] = WorkTask.Factory.StartNew(t => Action(t), state: i, executor: poolExcutor);
17.         WorkTask<int> task = WorkTask.Factory.StartNew(t =>
18.         {
19.             Thread.Sleep(100);
20.             Console.WriteLine("start thread");
21.             return 100;
22.         }, state: null, executor: poolExcutor);
24.         Console.WriteLine("start main");
25.         WorkTask.WaitAll(workTasks);
26.         Console.WriteLine(task.Result);
27.         Console.WriteLine(workTasks.Sum(t => t.Result));
28.     }
29.     private static int? Action(object? t)
30.     {
31.         Thread.Sleep(2000);
32.         Console.WriteLine($"Task Id:{Environment.CurrentManagedThreadId},Parameter:{t}");
33.         return t == null ? default(int?) : (int)t + 1;
34.     }
35. }

复制代码

调用结果

 

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！