功能描述:
- 支持创建多个线程池,并统一管理
- 支持不同线程池的容量控制,以及最少活动线程的设置
- 支持不同线程池中活动线程的闲时设置,即线程空闲时间到期后即自动被回收
结构设计:
- ThreadWorkerPoolManager: 线程池管理器,用于统一创建,获取,销毁线程池,使用单例模式
- ThreadWorkerPool: 线程池,用于管理指定数量的线程,由ThreadWorkerPoolManager管理,自身无法创建与销毁
- TheadWorkerPoolItem: 线程池项,用于包装线程工作器,协助ThreadWorkerPool更好的管理线程,例如取出,放回,闲时的控制
- TheadWorker: 线程工作器,用于包装系统线程System.Threading.Thread,使其可以重复使用,减少Thrad创建和销毁的性能开销
结构关系图:
详细设计:
ThreadWoker
要点设计:
- 完成一次任务后,System.Threading.Thread不能被系统销毁, 默认情况下new Thread(ThreadStart start).Start(), 当ThreadStart委托的任务完成后,系统将销毁该线程,也就是说创建一个System.Threading.Thread实例只能使用一次;为了使线程能被重复使用,ThreadWoker将使用 while+sleeping 的方式对系统线程进行包装,同时使用AutoResetEvent代替Thread.Sleep(timeout)来达到更佳的控制
- 闲时设计,线程资源是极其宝贵的系统资源,如果线程池中存在大量的空闲线程这是一种浪费,极端情况下将影响系统的稳定性和工作效率;ThreadWorker将使用AutoResetEvent和事件通知的方式来代替在线程池中定期轮询检查的方式,每完成一个任务将重新开始空闲时间的计算,如果ThreadWorker在线程池中被取出,那么ThreadWorker空闲时间将永远不会到期,直到ThreadWorker被返回线程池后才重新开始空闲时间的计算
状态图:
关键代码:
1 private void ThreadWorking() 2 { 3 while (_status != ThreadWorkerStatus.Abort) 4 { 5 //WaitOne 返回false表示等待超时,true接到取消等待的通知 6 //这里利用AutoResetEvent.WaitOne的特性来设计闲时控制,false表示空闲到期,true表示新的任务开始 7 if (!_waitEvent.WaitOne(_idleTime)) 8 { 9 if (!_isCanIdleExpired) //_isCanIdleExpired变量控制是否允许超时,例如被取出后将不能超时 10 continue; 11 12 _status = ThreadWorkerStatus.Abort; 13 _waitEvent.Close(); 14 _waitEvent.Dispose(); 15 if (OnIdleExpired != null) 16 OnIdleExpired(this, null); //空闲到期事件通知 17 return; 18 } 19 else if (_status == ThreadWorkerStatus.Abort) 20 return; 21 22 try 23 { 24 Working(); 25 } 26 catch (Exception ex) 27 { 28 _logger.TraceEvent(TraceEventType.Error, (int)TraceEventType.Error, ex.ToString()); 29 } 30 finally 31 { 32 _status = ThreadWorkerStatus.Idle; 33 if (OnWorkCompleted != null) 34 OnWorkCompleted(this, null); //任务完成事件通知 35 } 36 } 37 }
1 public void Work() 2 { 3 if (_status == ThreadWorkerStatus.Abort) 4 throw new InvalidOperationException("this ThreadWorker was Abort!"); 5 6 if (_status == ThreadWorkerStatus.Working) 7 throw new InvalidOperationException("this ThreadWorker was working, unable to duplicate work!"); 8 9 _status = ThreadWorkerStatus.Working; 10 _waitEvent.Set(); //通知线程有个新的工作要开始 11 }
ThreadWorkerPoolItem
要点设计:
- 链接ThreadWorker和线程池,线程池通过ThreadWorkerPoolItem控制ThreadWorker在线程池的取出,放回,销毁
- 通过订阅ThreadWorker的空闲到期事件OnIdleExpired,来完成线程池对线程的移除
- 通过订阅ThreadWorker的任务完成事件OnWorkCompleted,来完成线程返回线程池的操作
- 提供剩余空闲时间查询,来为线程池提供更优线程取出方案
完整代码:
1 public sealed class ThreadWorkerPoolItem 2 { 3 private ThreadWorkerPoolItemStatus _status; 4 private readonly ThreadWorkerBase _threadWorker; 5 private readonly ThreadWorkerPoolBase _threadWorkerPool; 6 private readonly int _idleTime; 7 private DateTime _startIdleTime; 8 9 internal ThreadWorkerPoolItem(ThreadWorkerPoolBase pool, ThreadWorkerBase threadWorker, ThreadWorkerPoolSettings poolSettings) 10 { 11 _threadWorkerPool = pool; 12 _threadWorker = threadWorker; 13 _threadWorker.OnIdleExpired += _threadWorker_OnIdleExpired; 14 _threadWorker.OnWorkCompleted += _threadWorker_OnWorkCompleted; 15 _threadWorker.Start(); 16 _status = ThreadWorkerPoolItemStatus.Idle; 17 _idleTime = poolSettings.IdleTime; 18 } 19 20 void _threadWorker_OnWorkCompleted(object sender, EventArgs args) 21 { 22 _threadWorkerPool.Return(this); 23 } 24 25 void _threadWorker_OnIdleExpired(object sender, EventArgs args) 26 { 27 _threadWorkerPool.Remove(this); 28 } 29 30 internal ThreadWorkerPoolItemStatus Status 31 { 32 get 33 { 34 if (_threadWorker.Status == ThreadWorkerStatus.Abort || _status == ThreadWorkerPoolItemStatus.Abort) 35 return ThreadWorkerPoolItemStatus.Abort; 36 37 return _status; 38 } 39 } 40 41 internal int SurplusIdleTime 42 { 43 get 44 { 45 if (_status == ThreadWorkerPoolItemStatus.Take || _idleTime == -1) 46 return -1; 47 48 int idledTime = (int)(_startIdleTime - DateTime.Now).TotalMilliseconds; 49 if (idledTime >= _idleTime) 50 return 0; 51 52 return idledTime; 53 } 54 } 55 56 internal void SetTake() 57 { 58 _threadWorker.IsCanIdleExpried = false; 59 _status = ThreadWorkerPoolItemStatus.Take; 60 } 61 62 internal void SetIdle() 63 { 64 _startIdleTime = DateTime.Now; 65 _status = ThreadWorkerPoolItemStatus.Idle; 66 _threadWorker.IsCanIdleExpried = true; 67 } 68 69 internal ThreadWorkerBase ThreadWorker 70 { 71 get { return _threadWorker; } 72 } 73 }
ThreadWorkerPool
要点设计:
- 使用Lock配合ThreadWorkerPoolItem的状态来确保多线程下,每次取出的都是空闲的ThreadWorker
- 取出的超时设计,由于线程池有容量控制,高并发下必然导致线程池满负荷,提供超时设置,有利于使用者自行控制满负荷情况下的处理;ThreadWorkerPool将使用while+sleeping的方式,同时使用AutoResetEvent代替Thread.Sleep(timeout)来达到更佳的控制,当一个线程被放回线程池时,另一等待获取者立即获取,而无需等待下一次轮询的到来
关键代码:
1 protected bool TryTake(int timeout, out ThreadWorkerBase threadWorker) 2 { 3 threadWorker = null; 4 lock (_takeLocker) 5 { 6 ThreadWorkerPoolItem worker = null; 7 DateTime startWaitTime; 8 while (!_isDestoryed) 9 { 10 worker = _threadWorkerList.Where(e => e.Status == Core.ThreadWorkerPoolItemStatus.Idle).OrderByDescending(e => e.SurplusIdleTime).FirstOrDefault(); 11 if (worker == null) 12 { 13 if (_threadWorkerList.Count < _settings.MaxThreadWorkerCount) 14 { 15 worker = this.CreatePoolItem(_threadWorkerList.Count + 1, _settings.IdleTime); 16 worker.SetTake(); 17 _threadWorkerList.Add(worker); 18 threadWorker = worker.ThreadWorker; 19 return true; 20 } 21 22 startWaitTime = DateTime.Now; 23 if (!_takeWaitEvent.WaitOne(timeout)) 24 { 25 threadWorker = null; 26 return false; 27 } 28 29 if (timeout != -1) 30 { 31 timeout = timeout - (int)(DateTime.Now - startWaitTime).TotalMilliseconds; 32 if (timeout <= 0) 33 { 34 threadWorker = null; 35 return false; 36 } 37 } 38 continue; 39 } 40 41 threadWorker = worker.ThreadWorker; 42 worker.SetTake(); 43 return true; 44 } 45 46 threadWorker = null; 47 return false; 48 } 49 }
1 internal void Return(ThreadWorkerPoolItem item) 2 { 3 item.SetIdle(); 4 _takeWaitEvent.Set(); 5 }
ThreadWorkerPoolManager使用单例模式管理,代码过于简单这里就不贴了......
有兴趣的同学可以点击这里进行下载源码查看:Nutshell.ThreadWorkerPool.zip
时间: 2024-10-08 08:32:40