原文网址: http://www.cnblogs.com/csdev
Networkcomms 是一款C# 语言编写的TCP/UDP通信框架 作者是英国人 以前是收费的 目前作者已经开源 许可是:Apache License v2
开源地址是:https://github.com/MarcFletcher/NetworkComms.Net
自定义线程池,用于并发处理通信框架收到的消息
/// <summary> /// A compact priority based thread pool used by NetworkComms.Net to run packet handlers /// 支持优先级处理的自定义线程池 /// </summary> public class CommsThreadPool { /// <summary> /// A sync object to make things thread safe /// 同步锁 /// </summary> object SyncRoot = new object(); /// <summary> /// Dictionary of threads, index is ThreadId /// 线程字典 索引是线程ID /// </summary> Dictionary<int, Thread> threadDict = new Dictionary<int, Thread>(); /// <summary> /// Dictionary of thread worker info, index is ThreadId /// 线程工作者字典 索引是线程ID /// </summary> Dictionary<int, WorkerInfo> workerInfoDict = new Dictionary<int, WorkerInfo>(); /// <summary> /// The minimum timespan between thread wait sleep join updates /// 线程 等待 睡眠 加入 更新 的一个最小时间段 /// </summary> TimeSpan ThreadWaitSleepJoinCountUpdateInterval = , , , , ); /// <summary> /// A quick lookup of the number of current threads which are idle and require jobs /// 正在等待任务的线程数量 /// </summary> ; /// <summary> /// Priority queue used to order call backs /// 用于处理回调委托的优先级队列 /// </summary> PriorityQueue<WaitCallBackWrapper> jobQueue = new PriorityQueue<WaitCallBackWrapper>(); /// <summary> /// Set to true to ensure correct shutdown of worker threads. /// 是否关闭 /// </summary> bool shutdown = false; /// <summary> /// The timespan after which an idle thread will close /// 线程空闲多长时间将会关闭 /// </summary> TimeSpan ThreadIdleTimeoutClose { get; set; } /// <summary> /// The maximum number of threads to create in the pool /// 最大的线程数 /// </summary> public int MaxTotalThreadsCount { get; private set; } /// <summary> /// The maximum number of active threads in the pool. This can be less than MaxTotalThreadsCount, taking account of waiting threads. /// 线程池中活动线程的最大数量 /// </summary> public int MaxActiveThreadsCount { get; private set; } /// <summary> /// The minimum number of idle threads to maintain in the pool /// 最小线程数 /// </summary> public int MinThreadsCount { get; private set; } /// <summary> /// The most recent count of pool threads which are waiting for IO /// 正在等待的线程数 /// </summary> public int CurrentNumWaitSleepJoinThreadsCache { get; private set; } /// <summary> /// The dateTime associated with the most recent count of pool threads which are waiting for IO /// /// </summary> public DateTime LastThreadWaitSleepJoinCountCacheUpdate { get; private set; } /// <summary> /// The total number of threads currently in the thread pool /// 当前的线程数 /// </summary> public int CurrentNumTotalThreads { get { lock (SyncRoot) return threadDict.Count; } } /// <summary> /// The total number of idle threads currently in the thread pool /// 当前空闲的线程数 /// </summary> public int CurrentNumIdleThreads { get { lock (SyncRoot) return requireJobThreadsCount; } } /// <summary> /// The total number of items currently waiting to be collected by a thread /// 队列数量 /// </summary> public int QueueCount { get { return jobQueue.Count; } } /// <summary> /// Create a new NetworkComms.Net thread pool /// 创建一个新的自定义线程池 /// </summary> /// <param name="minThreadsCount">最小线程数 Minimum number of idle threads to maintain in the pool</param> /// <param name="maxActiveThreadsCount">最大活动线程数 The maximum number of active (i.e. not waiting for IO) threads</param> /// <param name="maxTotalThreadsCount">最大线程数 Maximum number of threads to create in the pool</param> /// <param name="threadIdleTimeoutClose">空闲多长时间线程被关闭 Timespan after which an idle thread will close</param> public CommsThreadPool(int minThreadsCount, int maxActiveThreadsCount, int maxTotalThreadsCount, TimeSpan threadIdleTimeoutClose) { MinThreadsCount = minThreadsCount; MaxTotalThreadsCount = maxTotalThreadsCount; MaxActiveThreadsCount = maxActiveThreadsCount; ThreadIdleTimeoutClose = threadIdleTimeoutClose; } /// <summary> /// Prevent any additional threads from starting. Returns immediately. /// 关闭 系统不会再创建新的线程 /// </summary> public void BeginShutdown() { lock (SyncRoot) shutdown = true; } /// <summary> /// Prevent any additional threads from starting and return once all existing workers have completed. /// 不再创建新的线程 当前工作都完成后就返回 /// </summary> /// <param name="threadShutdownTimeoutMS"></param> ) { List<Thread> allWorkerThreads = new List<Thread>(); lock (SyncRoot) { foreach (var thread in threadDict) { workerInfoDict[thread.Key].ThreadSignal.Set(); allWorkerThreads.Add(thread.Value); } } //Wait for all threads to finish 等待所有线程完成 foreach (Thread thread in allWorkerThreads) { try { if (!thread.Join(threadShutdownTimeoutMS)) thread.Abort(); } catch (Exception ex) { LogTools.LogException(ex, "ManagedThreadPoolShutdownError"); } } lock (SyncRoot) { jobQueue.Clear(); shutdown = false; } } /// <summary> /// Enqueue a callback to the thread pool. /// 给线程池添加要处理的任务 /// </summary> /// <param name="priority">优先级 The priority with which to enqueue the provided callback</param> /// <param name="callback">回调方法(用于处理进入的数据) The callback to execute</param> /// <param name="state">通信框架解析好的数据 The state parameter to pass to the callback when executed</param> /// <returns>Returns the managed threadId running the callback if one was available, otherwise -1</returns> public int EnqueueItem(QueueItemPriority priority, WaitCallback callback, object state) { ; lock (SyncRoot) { UpdateThreadWaitSleepJoinCountCache(); , threadDict.Count - CurrentNumWaitSleepJoinThreadsCache - requireJobThreadsCount); //int numActiveThreads = Math.Max(0,threadDict.Count - CurrentNumWaitSleepJoinThreadsCache); && numInJobActiveThreadsCount < MaxActiveThreadsCount && threadDict.Count < MaxTotalThreadsCount) { //Launch a new thread 启动一个新线程 Thread newThread = new Thread(ThreadWorker); newThread.Name = "ManagedThreadPool_" + newThread.ManagedThreadId.ToString(); newThread.IsBackground = true; WorkerInfo info = new WorkerInfo(newThread.ManagedThreadId, new WaitCallBackWrapper(callback, state)); chosenThreadId = newThread.ManagedThreadId; threadDict.Add(newThread.ManagedThreadId, newThread); workerInfoDict.Add(newThread.ManagedThreadId, info); newThread.Start(info); } && numInJobActiveThreadsCount < MaxActiveThreadsCount) { jobQueue.TryAdd(new KeyValuePair<QueueItemPriority, WaitCallBackWrapper>(priority, new WaitCallBackWrapper(callback, state))); ; foreach (var info in workerInfoDict) { //Trigger the first idle thread 触发第一个空闲线程 checkCount++; if (info.Value.ThreadIdle) { info.Value.ClearThreadIdle(); requireJobThreadsCount--; info.Value.ThreadSignal.Set(); chosenThreadId = info.Value.ThreadId; break; } if (checkCount == workerInfoDict.Count) throw new Exception("IdleThreads count is " + requireJobThreadsCount.ToString() + " but unable to locate thread marked as idle."); } } else if (!shutdown) { //If there are no idle threads and we can't start any new ones we just have to enqueue the item //如果没有空闲线程 我们不能启动新线程 只是把任务添加到队列中 jobQueue.TryAdd(new KeyValuePair<QueueItemPriority, WaitCallBackWrapper>(priority, new WaitCallBackWrapper(callback, state))); } } return chosenThreadId; } /// <summary> /// The worker object for the thread pool /// 线程池的工作者方法 /// </summary> /// <param name="state"></param> private void ThreadWorker(object state) { WorkerInfo threadInfo = (WorkerInfo)state; do { //While there are jobs in the queue process the jobs //处理队列中的任务 while (true) { if (threadInfo.CurrentCallBackWrapper == null) { KeyValuePair<QueueItemPriority, WaitCallBackWrapper> packetQueueItem; lock (SyncRoot) { UpdateThreadWaitSleepJoinCountCache(); , threadDict.Count - CurrentNumWaitSleepJoinThreadsCache - requireJobThreadsCount); if (shutdown || threadDict.Count > MaxTotalThreadsCount) //If we have too many active threads { //If shutdown was true then we may need to set thread to idle //如果 shutdown设置为True 我们需要设置线程为空闲 ) requireJobThreadsCount--; threadInfo.ClearThreadIdle(); threadDict.Remove(threadInfo.ThreadId); workerInfoDict.Remove(threadInfo.ThreadId); UpdateThreadWaitSleepJoinCountCache(); return; } else if (numInJobActiveThreadsCount > MaxActiveThreadsCount) //If we have too many active threads 活动的线程太多 { //We wont close here to prevent thread creation/destruction thrashing. //We will instead act as if there is no work and wait to potentially be timed out if (!threadInfo.ThreadIdle) { threadInfo.SetThreadIdle(); requireJobThreadsCount++; } break; } else { //Try to get a job 获取一个新任务 if (!jobQueue.TryTake(out packetQueueItem)) //We fail to get a new job 没有获取到新任务 { //If we failed to get a job we switch to idle and wait to be triggered //如果没有获取到任务 转为空闲状态并等待被触发 if (!threadInfo.ThreadIdle) { threadInfo.SetThreadIdle(); requireJobThreadsCount++; } break; } else { ) requireJobThreadsCount--; threadInfo.UpdateCurrentCallBackWrapper(packetQueueItem.Value); threadInfo.ClearThreadIdle(); } } } } //Perform the waitcallBack //执行回调方法 try { threadInfo.SetInsideCallBack(); threadInfo.CurrentCallBackWrapper.WaitCallBack(threadInfo.CurrentCallBackWrapper.State); } catch (Exception ex) { LogTools.LogException(ex, "ManagedThreadPoolCallBackError", "An unhandled exception was caught while processing a callback. Make sure to catch errors in callbacks to prevent this error file being produced."); } finally { threadInfo.ClearInsideCallBack(); } threadInfo.UpdateLastActiveTime(); threadInfo.ClearCallBackWrapper(); } //As soon as the queue is empty we wait until perhaps close time //当队列为空时 就是当前没有可以处理的任务 线程进行等待 直到超过某个时间 #if NET2 , false)) #else )) #endif { //While we are waiting we check to see if we need to close //线程如果等待了太长时间 又没有新任务来 关闭线程 if (DateTime.Now - threadInfo.LastActiveTime > ThreadIdleTimeoutClose) { lock (SyncRoot) { //We have timed out but we don't go below the minimum //如果当前线程池中的线程数太少,接近最小线程数,则不再关闭线程 if (threadDict.Count > MinThreadsCount) { ) requireJobThreadsCount--; threadInfo.ClearThreadIdle(); threadDict.Remove(threadInfo.ThreadId); workerInfoDict.Remove(threadInfo.ThreadId); UpdateThreadWaitSleepJoinCountCache(); return; } } } } //We only leave via one of our possible breaks } while (true); } /// <summary> /// Returns the total number of threads in the pool which are waiting for IO /// 返回在等待IO库总的线程数 /// </summary> private void UpdateThreadWaitSleepJoinCountCache() { lock (SyncRoot) { if (DateTime.Now - LastThreadWaitSleepJoinCountCacheUpdate > ThreadWaitSleepJoinCountUpdateInterval) { ; foreach (var thread in threadDict) { if (workerInfoDict[thread.Key].InsideCallBack && thread.Value.ThreadState == ThreadState.WaitSleepJoin) returnValue++; } CurrentNumWaitSleepJoinThreadsCache = returnValue; LastThreadWaitSleepJoinCountCacheUpdate = DateTime.Now; } } } /// <summary> /// Provides a brief string summarisation the state of the thread pool /// 提供了一个简短的字符串解释线程池中的线程的状态 /// </summary> /// <returns></returns> public override string ToString() { lock (SyncRoot) { UpdateThreadWaitSleepJoinCountCache(); return "TotalTs:" + CurrentNumTotalThreads.ToString() + ", IdleTs:" + CurrentNumIdleThreads.ToString() + ", SleepTs:" + CurrentNumWaitSleepJoinThreadsCache.ToString() + ", Q:" + QueueCount.ToString(); } } } /// <summary> /// A private wrapper used by CommsThreadPool /// 工作者信息类 提供给CommsThreadPool线程池使用 /// </summary> class WorkerInfo { public int ThreadId { get; private set; } public AutoResetEvent ThreadSignal { get; private set; } public bool ThreadIdle { get; private set; } public DateTime LastActiveTime { get; private set; } public WaitCallBackWrapper CurrentCallBackWrapper { get; private set; } public bool InsideCallBack { get; private set; } public WorkerInfo(int threadId, WaitCallBackWrapper initialisationCallBackWrapper) { ThreadSignal = new AutoResetEvent(false); ThreadIdle = false; ThreadId = threadId; LastActiveTime = DateTime.Now; this.CurrentCallBackWrapper = initialisationCallBackWrapper; } public void UpdateCurrentCallBackWrapper(WaitCallBackWrapper waitCallBackWrapper) { CurrentCallBackWrapper = waitCallBackWrapper; } public void UpdateLastActiveTime() { LastActiveTime = DateTime.Now; } public void ClearCallBackWrapper() { CurrentCallBackWrapper = null; } /// <summary> /// Set InsideCallBack to true /// 设置内部回调为True /// </summary> public void SetInsideCallBack() { InsideCallBack = true; } /// <summary> /// Set InsideCallBack to false /// 设置内部回调为False /// </summary> public void ClearInsideCallBack() { InsideCallBack = false; } /// <summary> /// Set threadIdle to true /// 设置线程空闲为True /// </summary> public void SetThreadIdle() { this.ThreadIdle = true; } /// <summary> /// Set threadIdle to false /// 设置线程空闲为False /// </summary> public void ClearThreadIdle() { this.ThreadIdle = false; } } #endif /// <summary> /// A private wrapper used by CommsThreadPool /// 一个回调的包装器 在自定义线程池中使用 /// </summary> class WaitCallBackWrapper { public WaitCallback WaitCallBack { get; private set; } public object State { get; private set; } public WaitCallBackWrapper(WaitCallback waitCallBack, object state) { this.WaitCallBack = waitCallBack; this.State = state; } }