asp.net core C#設計一個實用的執行緒池
菜菜呀，我最近研究技術呢，發現線上一個任務程式執行緒數有點多呀
x總，你學程式設計呢？
作為公司總負責人，我以後還要管理技術部門呢，怎麼能不會技術呢
（技術部完了）。。。。。。。
趕緊看看線上那個執行緒特別多的程式，給你2個小時優化一下
x總，我想辭職
菜菜呀，心不要浮躁，學學小馬，心平氣和養養生
............................
好了，給你半天時間把執行緒多的問題優化一下，要不然扣你績效
（嘞了個擦）。。。。。。
經過一個多小時的程式碼排查終於查明瞭線上程式執行緒數過多的原因：這是一個接收mq訊息的一個服務，程式大體思路是這樣的，監聽的執行緒每次收到一條訊息，就啟動一個執行緒去執行，每次啟動的執行緒都是新的。說到這裡，咱們就談一談這個程式有哪些弊端呢：
1. 每次收到一條訊息都建立一個新的執行緒，要知道執行緒的資源對於系統來說是很昂貴的，訊息處理完成還要銷燬這個執行緒。
2. 這個程式用到的執行緒數量是沒有限制的。當執行緒到達一定數量，程式反而因執行緒在cpu切換開銷的原因處理效率降低。無論的你的伺服器cpu是多少核心，這個現象都有發生的可能。
執行緒多的問題該怎麼解決呢，增加cpu核心數？治標不治本。對於開發者而言，最為常用也最為有效的是執行緒池化，也就是說執行緒池。
執行緒池是一種多執行緒處理形式，處理過程中將任務新增到佇列，然後在建立執行緒後自動啟動這些任務。這避免了在處理短時間任務時建立與銷燬執行緒的代價。執行緒池不僅能夠保證核心的充分利用，還能防止過分排程。可用執行緒數量應該取決於可用的併發處理器、處理器核心、記憶體、網路sockets等的數量。 例如，執行緒數一般取cpu數量+2比較合適，執行緒數過多會導致額外的執行緒切換開銷。
執行緒池其中一項很重要的技術點就是任務的佇列，佇列雖然屬於一種基礎的資料結構，但是發揮了舉足輕重的作用。
佇列是一種特殊的線性表，特殊之處在於它只允許在表的前端（front）進行刪除操作，而在表的後端（rear）進行插入操作，和棧一樣，佇列是一種操作受限制的線性表。進行插入操作的端稱為隊尾，進行刪除操作的端稱為隊頭。
佇列是一種採用的FIFO(first in first out)方式的線性表，也就是經常說的先進先出策略。
實現
陣列
佇列可以用陣列Q[1…m]來儲存，陣列的上界m即是佇列所容許的最大容量。在佇列的運算中需設兩個指標：head，隊頭指標，指向實際隊頭元素+1的位置；tail，隊尾指標，指向實際隊尾元素位置。一般情況下，兩個指標的初值設為0，這時佇列為空，沒有元素。以下為一個簡單的例項（生產環境需要優化）：
public class QueueArray<T>
{
//佇列元素的陣列容器
T[] container = null;
int IndexHeader, IndexTail;
public QueueArray(int size)
{
container = new T[size];
IndexHeader = 0;
IndexTail = 0;
}
public void Enqueue(T item)
{
//入隊的元素放在頭指標的指向位置，然後頭指標前移
container[IndexHeader] = item;
IndexHeader++;
}
public T Dequeue()
{
//出隊：把尾元素指標指向的元素取出並清空（不清空也可以）對應的位置，尾指標前移
T item = container[IndexTail];
container[IndexTail] = default(T);
IndexTail++;
return item;
}
}
連結串列
佇列採用的FIFO(first in first out)，新元素總是被插入到連結串列的尾部，而讀取的時候總是從連結串列的頭部開始讀取。每次讀取一個元素，釋放一個元素。所謂的動態建立，動態釋放。因而也不存在溢位等問題。由於連結串列由元素連線而成，遍歷也方便。以下是一個例項僅供參考：
public class QueueLinkList<T>
{
LinkedList<T> contianer = null;
public QueueLinkList()
{
contianer = new LinkedList<T>();
}
public void Enqueue(T item)
{
//入隊的元素其實就是加入到隊尾
contianer.AddLast(item);
}
public T Dequeue()
{
//出隊：取連結串列第一個元素，然後把這個元素刪除
T item = contianer.First.Value;
contianer.RemoveFirst();
return item;
}
}
佇列的擴充套件閱讀
1. 佇列通過陣列來實現的話有什麼問題嗎？是的。首先基於陣列不可變本質的因素（具體可參考菜菜之前的文章），當一個佇列的元素把陣列沾滿的時候，陣列擴容是有效能問題的，陣列的擴容過程不只是開闢新空間分配記憶體那麼簡單，還要有陣列元素的copy過程，更可怕的是會給GC造成極大的壓力。如果陣列比較小可能影響比較小，但是當一個數組比較大的時候，比如佔用500M記憶體的一個數組，資料copy其實會造成比較大的效能損失。
2. 佇列通過陣列來實現，隨著頭指標和尾指標的位置移動，尾指標最終會指向第一個元素的位置，也就是說沒有元素可以出隊了，其實要解決這個問題有兩種方式，其一：在出隊或者入隊的過程中不斷的移動所有元素的位置，避免上邊所說的極端情況發生；其二：可以把陣列的首尾元素連線起來，使其成為一個環狀，也就是經常說的迴圈佇列。
3. 佇列在一些特殊場景下其實還有一些變種，比如說迴圈佇列，阻塞佇列，併發佇列等，有興趣的同學可以去研究一下，這裡不在展開討論。這裡說到阻塞佇列就多說一句，其實用阻塞佇列可以實現一個最基本的生產者消費者模式。
4. 當佇列用連結串列方式實現的時候，由於連結串列的首尾操作時間複雜度都是O（1），而且沒有空間大小的限制，所以一般的佇列用連結串列實現更簡單
5. 當佇列中無元素可出隊或者沒有空間可入隊的時候，是阻塞當前的操作還是返回錯誤資訊，取決於在座各位佇列的設計者了。
Net Core C# 版本
//執行緒池
public class ThreadPool
{
bool PoolEnable = false; //執行緒池是否可用
List<Thread> ThreadContainer = null; //執行緒的容器
ConcurrentQueue<ActionData> JobContainer = null; //任務的容器
public ThreadPool(int threadNumber)
{
PoolEnable = true;
ThreadContainer = new List<Thread>(threadNumber);
JobContainer = new ConcurrentQueue<ActionData>();
for (int i = 0; i < threadNumber; i++)
{
var t = new Thread(RunJob);
ThreadContainer.Add(t);
t.Start();
}
}
//向執行緒池新增一個任務
public void AddTask(Action<object> job,object obj, Action<Exception> errorCallBack=null)
{
if (JobContainer != null)
{
JobContainer.Enqueue(new ActionData { Job = job, Data = obj , ErrorCallBack= errorCallBack });
}
}
//終止執行緒池
public void FinalPool()
{
PoolEnable = false;
JobContainer = null;
if (ThreadContainer != null)
{
foreach (var t in ThreadContainer)
{
//強制執行緒退出並不好，會有異常
//t.Abort();
t.Join();
}
ThreadContainer = null;
}
}
private void RunJob()
{
while (true&& JobContainer!=null&& PoolEnable)
{
//任務列表取任務
ActionData job=null;
JobContainer?.TryDequeue(out job);
if (job == null)
{
//如果沒有任務則休眠
Thread.Sleep(10);
continue;
}
try
{
//執行任務
job.Job.Invoke(job.Data);
}
catch(Exception error)
{
//異常回調
job?.ErrorCallBack(error);
}
}
}
}
public class ActionData
{
//執行任務的引數
public object Data { get; set; }
//執行的任務
public Action<object> Job { get; set; }
//發生異常時候的回撥方法
public Action<Exception> ErrorCallBack { get; set; }
}使用方法
ThreadPool pool = new ThreadPool(100);
for (int i = 0; i < 5000; i++)
{
pool.AddTask((obj) =>
{
Console.WriteLine($"{obj}__{System.Threading.Thread.CurrentThread.ManagedThreadId}");
}, i, (e) =>
{
Console.WriteLine(e.Message);
});
}
pool.FinalPool();
Console.Read();
