C#多线程编程系列（四）- 使用线程池
目錄
- 1.1 簡(jiǎn)介
- 1.2 在線程池中調(diào)用委托
- 1.3 向線程池中放入異步操作
- 1.4 線程池與并行度
- 1.5 實(shí)現(xiàn)一個(gè)取消選項(xiàng)
- 1.6 在線程池中使用等待事件處理器及超時(shí)
- 1.7 使用計(jì)時(shí)器
- 1.8 使用BackgroundWorker組件
- 參考書籍
- 筆者水平有限，如果錯(cuò)誤歡迎各位批評(píng)指正！
1.1 簡(jiǎn)介#
在本章中，主要介紹線程池(ThreadPool)的使用；在C#中它叫System.Threading.ThreadPool，在使用線程池之前首先我們得明白一個(gè)問題，那就是為什么要使用線程池。其主要原因是創(chuàng)建一個(gè)線程的代價(jià)是昂貴的，創(chuàng)建一個(gè)線程會(huì)消耗很多的系統(tǒng)資源。
那么線程池是如何解決這個(gè)問題的呢？線程池在初始時(shí)會(huì)自動(dòng)創(chuàng)建一定量的線程供程序調(diào)用，使用時(shí)，開發(fā)人員并不直接分配線程，而是將需要做的工作放入線程池工作隊(duì)列中，由線程池分配已有的線程進(jìn)行處理，等處理完畢后線程不是被銷毀，而是重新回到線程池中，這樣節(jié)省了創(chuàng)建線程的開銷。
但是在使用線程池時(shí)，需要注意以下幾點(diǎn)，這將非常重要。
- 線程池不適合處理長(zhǎng)時(shí)間運(yùn)行的作業(yè)，或者處理需要與其它線程同步的作業(yè)。
- 避免將線程池中的工作線程分配給I/O首先的任務(wù)，這種任務(wù)應(yīng)該使用TPL模型。
- 如非必須，不要手動(dòng)設(shè)置線程池的最小線程數(shù)和最大線程數(shù)，CLR會(huì)自動(dòng)的進(jìn)行線程池的擴(kuò)張和收縮，手動(dòng)干預(yù)往往讓性能更差。
1.2 在線程池中調(diào)用委托#
本節(jié)展示的是如何在線程池中如何異步的執(zhí)行委托，然后將介紹一個(gè)叫異步編程模型(Asynchronous Programming Model，簡(jiǎn)稱APM)的異步編程方式。
在本節(jié)及以后，為了降低代碼量，在引用程序集聲明位置默認(rèn)添加了using static System.Console和using static System.Threading.Thead聲明，這樣聲明可以讓我們?cè)诔绦蛑猩傩┮恍┮饬x不大的調(diào)用語句。
演示代碼如下所示，使用了普通創(chuàng)建線程和APM方式來執(zhí)行同一個(gè)任務(wù)。
Copy
static void Main(string[] args) { int threadId = 0; RunOnThreadPool poolDelegate = Test; var t = new Thread(() => Test(out threadId)); t.Start(); t.Join(); WriteLine($"手動(dòng)創(chuàng)建線程 Id: {threadId}"); // 使用APM方式 進(jìn)行異步調(diào)用 異步調(diào)用會(huì)使用線程池中的線程 IAsyncResult r = poolDelegate.BeginInvoke(out threadId, Callback, "委托異步調(diào)用"); r.AsyncWaitHandle.WaitOne(); // 獲取異步調(diào)用結(jié)果 string result = poolDelegate.EndInvoke(out threadId, r); WriteLine($"Thread - 線程池工作線程Id: {threadId}"); WriteLine(result); Console.ReadLine(); } // 創(chuàng)建帶一個(gè)參數(shù)的委托類型 private delegate string RunOnThreadPool(out int threadId); private static void Callback(IAsyncResult ar) { WriteLine("Callback - 開始運(yùn)行Callback..."); WriteLine($"Callback - 回調(diào)傳遞狀態(tài): {ar.AsyncState}"); WriteLine($"Callback - 是否為線程池線程: {CurrentThread.IsThreadPoolThread}"); WriteLine($"Callback - 線程池工作線程Id: {CurrentThread.ManagedThreadId}"); } private static string Test(out int threadId) { string isThreadPoolThread = CurrentThread.IsThreadPoolThread ? "ThreadPool - ": "Thread - "; WriteLine($"{isThreadPoolThread}開始運(yùn)行..."); WriteLine($"{isThreadPoolThread}是否為線程池線程: {CurrentThread.IsThreadPoolThread}"); Sleep(TimeSpan.FromSeconds(2)); threadId = CurrentThread.ManagedThreadId; return $"{isThreadPoolThread}線程池工作線程Id: {threadId}"; }
運(yùn)行結(jié)果如下圖所示，其中以Thread開頭的為手動(dòng)創(chuàng)建的線程輸出的信息，而TheadPool為開始線程池任務(wù)輸出的信息，Callback為APM模式運(yùn)行任務(wù)結(jié)束后，執(zhí)行的回調(diào)方法，可以清晰的看到，Callback的線程也是線程池的工作線程。
在上文中，使用BeginOperationName/EndOperationName方法和.Net中的IAsyncResult對(duì)象的方式被稱為異步編程模型(或APM模式)，這樣的方法被稱為異步方法。使用委托的BeginInvoke方法來運(yùn)行該委托，BeginInvoke接收一個(gè)回調(diào)函數(shù)，該回調(diào)函數(shù)會(huì)在任務(wù)處理完成后背調(diào)用，并且可以傳遞一個(gè)用戶自定義的狀態(tài)給回調(diào)函數(shù)。
現(xiàn)在這種APM編程方式用的越來越少了，更推薦使用任務(wù)并行庫(kù)(Task Parallel Library，簡(jiǎn)稱TPL)來組織異步API。
1.3 向線程池中放入異步操作#
本節(jié)將介紹如何將異步操作放入線程池中執(zhí)行，并且如何傳遞參數(shù)給線程池中的線程。本節(jié)中主要用到的是ThreadPool.QueueUserWorkItem()方法，該方法可將需要運(yùn)行的任務(wù)通過委托的形式傳遞給線程池中的線程，并且允許傳遞參數(shù)。
使用比較簡(jiǎn)單，演示代碼如下所示。演示了線程池使用中如何傳遞方法和參數(shù)，最后需要注意的是使用了Lambda表達(dá)式和它的閉包機(jī)制。
Copy
static void Main(string[] args) { const int x = 1; const int y = 2; const string lambdaState = "lambda state 2"; // 直接將方法傳遞給線程池 ThreadPool.QueueUserWorkItem(AsyncOperation); Sleep(TimeSpan.FromSeconds(1)); // 直接將方法傳遞給線程池 并且 通過state傳遞參數(shù) ThreadPool.QueueUserWorkItem(AsyncOperation, "async state"); Sleep(TimeSpan.FromSeconds(1)); // 使用Lambda表達(dá)式將任務(wù)傳遞給線程池 并且通過 state傳遞參數(shù) ThreadPool.QueueUserWorkItem(state => { WriteLine($"Operation state: {state}"); WriteLine($"工作線程 id: {CurrentThread.ManagedThreadId}"); Sleep(TimeSpan.FromSeconds(2)); }, "lambda state"); // 使用Lambda表達(dá)式將任務(wù)傳遞給線程池 通過 **閉包** 機(jī)制傳遞參數(shù) ThreadPool.QueueUserWorkItem(_ => { WriteLine($"Operation state: {x + y}, {lambdaState}"); WriteLine($"工作線程 id: {CurrentThread.ManagedThreadId}"); Sleep(TimeSpan.FromSeconds(2)); }, "lambda state"); ReadLine(); } private static void AsyncOperation(object state) { WriteLine($"Operation state: {state ?? "(null)"}"); WriteLine($"工作線程 id: {CurrentThread.ManagedThreadId}"); Sleep(TimeSpan.FromSeconds(2)); }
運(yùn)行結(jié)果如下圖所示。
1.4 線程池與并行度#
在本節(jié)中，主要是使用普通創(chuàng)建線程和使用線程池內(nèi)的線程在任務(wù)量比較大的情況下有什么區(qū)別，我們模擬了一個(gè)場(chǎng)景，創(chuàng)建了很多不同的線程，然后分別使用普通創(chuàng)建線程方式和線程池方式看看有什么不同。
Copy
static void Main(string[] args) { const int numberOfOperations = 500; var sw = new Stopwatch(); sw.Start(); UseThreads(numberOfOperations); sw.Stop(); WriteLine($"使用線程執(zhí)行總用時(shí): {sw.ElapsedMilliseconds}"); sw.Reset(); sw.Start(); UseThreadPool(numberOfOperations); sw.Stop(); WriteLine($"使用線程池執(zhí)行總用時(shí): {sw.ElapsedMilliseconds}"); Console.ReadLine(); } static void UseThreads(int numberOfOperations) { using (var countdown = new CountdownEvent(numberOfOperations)) { WriteLine("通過創(chuàng)建線程調(diào)度工作"); for (int i = 0; i < numberOfOperations; i++) { var thread = new Thread(() => { Write($"{CurrentThread.ManagedThreadId},"); Sleep(TimeSpan.FromSeconds(0.1)); countdown.Signal(); }); thread.Start(); } countdown.Wait(); WriteLine(); } } static void UseThreadPool(int numberOfOperations) { using (var countdown = new CountdownEvent(numberOfOperations)) { WriteLine("使用線程池開始工作"); for (int i = 0; i < numberOfOperations; i++) { ThreadPool.QueueUserWorkItem(_ => { Write($"{CurrentThread.ManagedThreadId},"); Sleep(TimeSpan.FromSeconds(0.1)); countdown.Signal(); }); } countdown.Wait(); WriteLine(); } }
執(zhí)行結(jié)果如下，可見使用原始的創(chuàng)建線程執(zhí)行，速度非常快。只花了2秒鐘，但是創(chuàng)建了500多個(gè)線程，而使用線程池相對(duì)來說比較慢，花了9秒鐘，但是只創(chuàng)建了很少的線程，為操作系統(tǒng)節(jié)省了線程和內(nèi)存空間，但花了更多的時(shí)間。
1.5 實(shí)現(xiàn)一個(gè)取消選項(xiàng)#
在之前的文章中有提到，如果需要終止一個(gè)線程的執(zhí)行，那么可以使用Abort()方法，但是有諸多的原因并不推薦使用Abort()方法。
這里推薦的方式是使用協(xié)作式取消(cooperative cancellation)，這是一種可靠的技術(shù)來安全取消不再需要的任務(wù)。其主要用到CancellationTokenSource和CancellationToken兩個(gè)類，具體用法見下面演示代碼。
以下延時(shí)代碼主要是實(shí)現(xiàn)了使用CancellationToken和CancellationTokenSource來實(shí)現(xiàn)任務(wù)的取消。但是任務(wù)取消后可以進(jìn)行三種操作，分別是：直接返回、拋出ThrowIfCancellationRequesed異常和執(zhí)行回調(diào)。詳細(xì)請(qǐng)看代碼。
Copy
static void Main(string[] args) { // 使用CancellationToken來取消任務(wù) 取消任務(wù)直接返回 using (var cts = new CancellationTokenSource()) { CancellationToken token = cts.Token; ThreadPool.QueueUserWorkItem(_ => AsyncOperation1(token)); Sleep(TimeSpan.FromSeconds(2)); cts.Cancel(); } // 取消任務(wù) 拋出 ThrowIfCancellationRequesed 異常 using (var cts = new CancellationTokenSource()) { CancellationToken token = cts.Token; ThreadPool.QueueUserWorkItem(_ => AsyncOperation2(token)); Sleep(TimeSpan.FromSeconds(2)); cts.Cancel(); } // 取消任務(wù) 并 執(zhí)行取消后的回調(diào)函數(shù) using (var cts = new CancellationTokenSource()) { CancellationToken token = cts.Token; token.Register(() => { WriteLine("第三個(gè)任務(wù)被取消，執(zhí)行回調(diào)函數(shù)。"); }); ThreadPool.QueueUserWorkItem(_ => AsyncOperation3(token)); Sleep(TimeSpan.FromSeconds(2)); cts.Cancel(); } ReadLine(); } static void AsyncOperation1(CancellationToken token) { WriteLine("啟動(dòng)第一個(gè)任務(wù)."); for (int i = 0; i < 5; i++) { if (token.IsCancellationRequested) { WriteLine("第一個(gè)任務(wù)被取消."); return; } Sleep(TimeSpan.FromSeconds(1)); } WriteLine("第一個(gè)任務(wù)運(yùn)行完成."); } static void AsyncOperation2(CancellationToken token) { try { WriteLine("啟動(dòng)第二個(gè)任務(wù)."); for (int i = 0; i < 5; i++) { token.ThrowIfCancellationRequested(); Sleep(TimeSpan.FromSeconds(1)); } WriteLine("第二個(gè)任務(wù)運(yùn)行完成."); } catch (OperationCanceledException) { WriteLine("第二個(gè)任務(wù)被取消."); } } static void AsyncOperation3(CancellationToken token) { WriteLine("啟動(dòng)第三個(gè)任務(wù)."); for (int i = 0; i < 5; i++) { if (token.IsCancellationRequested) { WriteLine("第三個(gè)任務(wù)被取消."); return; } Sleep(TimeSpan.FromSeconds(1)); } WriteLine("第三個(gè)任務(wù)運(yùn)行完成."); }
運(yùn)行結(jié)果如下所示，符合預(yù)期結(jié)果。
1.6 在線程池中使用等待事件處理器及超時(shí)#
本節(jié)將介紹如何在線程池中使用等待任務(wù)和如何進(jìn)行超時(shí)處理，其中主要用到ThreadPool.RegisterWaitForSingleObject()方法，該方法允許傳入一個(gè)WaitHandle對(duì)象，和需要執(zhí)行的任務(wù)、超時(shí)時(shí)間等。通過使用這個(gè)方法，可完成線程池情況下對(duì)超時(shí)任務(wù)的處理。
演示代碼如下所示，運(yùn)行了兩次使用ThreadPool.RegisterWaitForSingleObject()編寫超時(shí)代碼的RunOperations()方法，但是所傳入的超時(shí)時(shí)間不同，所以造成一個(gè)必然超時(shí)和一個(gè)不會(huì)超時(shí)的結(jié)果。
Copy
static void Main(string[] args) { // 設(shè)置超時(shí)時(shí)間為 5s WorkerOperation會(huì)延時(shí) 6s 肯定會(huì)超時(shí) RunOperations(TimeSpan.FromSeconds(5)); // 設(shè)置超時(shí)時(shí)間為 7s 不會(huì)超時(shí) RunOperations(TimeSpan.FromSeconds(7)); } static void RunOperations(TimeSpan workerOperationTimeout) { using (var evt = new ManualResetEvent(false)) using (var cts = new CancellationTokenSource()) { WriteLine("注冊(cè)超時(shí)操作..."); // 傳入同步事件 超時(shí)處理函數(shù) 和 超時(shí)時(shí)間 var worker = ThreadPool.RegisterWaitForSingleObject(evt , (state, isTimedOut) => WorkerOperationWait(cts, isTimedOut) , null , workerOperationTimeout , true); WriteLine("啟動(dòng)長(zhǎng)時(shí)間運(yùn)行操作..."); ThreadPool.QueueUserWorkItem(_ => WorkerOperation(cts.Token, evt)); Sleep(workerOperationTimeout.Add(TimeSpan.FromSeconds(2))); // 取消注冊(cè)等待的操作 worker.Unregister(evt); ReadLine(); } } static void WorkerOperation(CancellationToken token, ManualResetEvent evt) { for (int i = 0; i < 6; i++) { if (token.IsCancellationRequested) { return; } Sleep(TimeSpan.FromSeconds(1)); } evt.Set(); } static void WorkerOperationWait(CancellationTokenSource cts, bool isTimedOut) { if (isTimedOut) { cts.Cancel(); WriteLine("工作操作超時(shí)并被取消."); } else { WriteLine("工作操作成功."); } }
運(yùn)行結(jié)果如下圖所示，與預(yù)期結(jié)果相符。
1.7 使用計(jì)時(shí)器#
計(jì)時(shí)器是FCL提供的一個(gè)類，叫System.Threading.Timer，可要結(jié)果與創(chuàng)建周期性的異步操作。該類使用比較簡(jiǎn)單。
以下的演示代碼使用了定時(shí)器，并設(shè)置了定時(shí)器延時(shí)啟動(dòng)時(shí)間和周期時(shí)間。
Copy
static void Main(string[] args) { WriteLine("按下回車鍵，結(jié)束定時(shí)器..."); DateTime start = DateTime.Now; // 創(chuàng)建定時(shí)器 _timer = new Timer(_ => TimerOperation(start), null , TimeSpan.FromSeconds(1) , TimeSpan.FromSeconds(2)); try { Sleep(TimeSpan.FromSeconds(6)); _timer.Change(TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(4)); ReadLine(); } finally { //實(shí)現(xiàn)了IDispose接口 要及時(shí)釋放 _timer.Dispose(); } } static Timer _timer; static void TimerOperation(DateTime start) { TimeSpan elapsed = DateTime.Now - start; WriteLine($"離 {start} 過去了 {elapsed.Seconds} 秒. " + $"定時(shí)器線程池 線程 id: {CurrentThread.ManagedThreadId}"); }
運(yùn)行結(jié)果如下所示，可見定時(shí)器根據(jù)所設(shè)置的周期時(shí)間循環(huán)的調(diào)用TimerOperation()方法。
1.8 使用BackgroundWorker組件#
本節(jié)主要介紹BackgroundWorker組件的使用，該組件實(shí)際上被用于Windows窗體應(yīng)用程序(Windows Forms Application，簡(jiǎn)稱 WPF)中，通過它實(shí)現(xiàn)的代碼可以直接與UI控制器交互，更加自認(rèn)和好用。
演示代碼如下所示，使用BackgroundWorker來實(shí)現(xiàn)對(duì)數(shù)據(jù)進(jìn)行計(jì)算，并且讓其支持報(bào)告工作進(jìn)度，支持取消任務(wù)。
Copy
static void Main(string[] args) { var bw = new BackgroundWorker(); // 設(shè)置可報(bào)告進(jìn)度更新 bw.WorkerReportsProgress = true; // 設(shè)置支持取消操作 bw.WorkerSupportsCancellation = true; // 需要做的工作 bw.DoWork += Worker_DoWork; // 工作處理進(jìn)度 bw.ProgressChanged += Worker_ProgressChanged; // 工作完成后處理函數(shù) bw.RunWorkerCompleted += Worker_Completed; bw.RunWorkerAsync(); WriteLine("按下 `C` 鍵 取消工作"); do { if (ReadKey(true).KeyChar == 'C') { bw.CancelAsync(); } } while (bw.IsBusy); } static void Worker_DoWork(object sender, DoWorkEventArgs e) { WriteLine($"DoWork 線程池 線程 id: {CurrentThread.ManagedThreadId}"); var bw = (BackgroundWorker)sender; for (int i = 1; i <= 100; i++) { if (bw.CancellationPending) { e.Cancel = true; return; } if (i % 10 == 0) { bw.ReportProgress(i); } Sleep(TimeSpan.FromSeconds(0.1)); } e.Result = 42; } static void Worker_ProgressChanged(object sender, ProgressChangedEventArgs e) { WriteLine($"已完成{e.ProgressPercentage}%. " + $"處理線程 id: {CurrentThread.ManagedThreadId}"); } static void Worker_Completed(object sender, RunWorkerCompletedEventArgs e) { WriteLine($"完成線程池線程 id: {CurrentThread.ManagedThreadId}"); if (e.Error != null) { WriteLine($"異常 {e.Error.Message} 發(fā)生."); } else if (e.Cancelled) { WriteLine($"操作已被取消."); } else { WriteLine($"答案是 : {e.Result}"); } }
運(yùn)行結(jié)果如下所示。
在本節(jié)中，使用了C#中的另外一個(gè)語法，叫事件(event)。當(dāng)然這里的事件不同于之前在線程同步章節(jié)中提到的事件，這里是觀察者設(shè)計(jì)模式的體現(xiàn)，包括事件源、訂閱者和事件處理程序。因此，除了異步APM模式意外，還有基于事件的異步模式(Event-based Asynchronous Pattern，簡(jiǎn)稱 EAP)。
參考書籍
本文主要參考了以下幾本書，在此對(duì)這些作者表示由衷的感謝你們提供了這么好的資料。
源碼下載點(diǎn)擊鏈接?示例源碼下載
筆者水平有限，如果錯(cuò)誤歡迎各位批評(píng)指正！
作者：InCerry
出處：https://www.cnblogs.com/InCerry/p/9432804.html
版權(quán)：本文采用「署名 4.0 國(guó)際」知識(shí)共享許可協(xié)議進(jìn)行許可。
總結(jié)
以上是生活随笔為你收集整理的C#多线程编程系列（四）- 使用线程池的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： UML之用例图(use case)箭头方
- 下一篇： C#的变迁史05 - C# 4.0篇