C#中幾種常見的異步處理的方法

先大概看一下控制臺應(yīng)用程序的Main方法的主要代碼：

001 static bool done = false;

002 static decimal count2 = 0;

003 static int threadDone = 0;//標(biāo)志啟用線程數(shù)?

004 static System.Timers.Timer timer = new System.Timers.Timer(1000);

005

006 static decimal[] threadPoolCounters = new decimal[10];

007 static Thread[] threads = new Thread[10];

008 static System.Timers.Timer[] threadTimers = new System.Timers.Timer[10];

009

010 static void Main(string[] args)

011 {

012 timer.Stop();

013 /*當(dāng) AutoReset 設(shè)置為 false 時，Timer 只在第一個 Interval 過后引發(fā)一次 Elapsed 事件。

014 若要保持以 Interval 時間間隔引發(fā) Elapsed 事件，請將 AutoReset 設(shè)置為 true。*/

015 timer.AutoReset = false;

016 timer.Elapsed += new ElapsedEventHandler(OnTimerEvent);//當(dāng)timer.Start()時，觸發(fā)事件

017 decimal total = 0;

018

019 // raw test

020 decimal count1 = SingleThreadTest();//單一線程，一跑到底

021 Console.WriteLine("Single thread count = " + count1.ToString());

022

023 // create one thread, increment counter, destroy thread, repeat

024 Console.WriteLine();

025 CreateAndDestroyTest();//創(chuàng)建一個線程，運算，然后銷毀該線程 重復(fù)前面的動作

026 Console.WriteLine("Create and destroy per count = " + count2.ToString());

027

028 // Create 10 threads and run them simultaneously

029 //一次性創(chuàng)建10個線程，然后遍歷使線程執(zhí)行運算

030 Console.WriteLine();

031 InitThreadPoolCounters();

032 InitThreads();

033 StartThreads();

034 while (threadDone != 10) { };

035 Console.WriteLine("10 simultaneous threads:");

036 for (int i = 0; i < 10; i++)

037 {

038 Console.WriteLine("T" + i.ToString() + " = " + threadPoolCounters[i].ToString() + " ");

039 total += threadPoolCounters[i];

040 }

041 Console.WriteLine("Total = " + total.ToString());

042 Console.WriteLine();

043

044 Console.WriteLine("http:///////////////////////////////////////////////////");

045

046 // using ThreadPool

047 //直接通過線程池的QueueUserWorkItem方法，按隊列執(zhí)行10個任務(wù)

048 Console.WriteLine();

049 Console.WriteLine("ThreadPool:");

050 InitThreadPoolCounters();

051 QueueThreadPoolThreads();

052 while (threadDone != 10) { };

053 Console.WriteLine("ThreadPool: 10 simultaneous threads:");

054 total = 0;

055 for (int i = 0; i < 10; i++)

056 {

057 // threadTimers[i].Stop();

058 // threadTimers[i].Dispose();

059 Console.WriteLine("T" + i.ToString() + " = " + threadPoolCounters[i].ToString() + " ");

060 total += threadPoolCounters[i];

061 }

062 Console.WriteLine("Total = " + total.ToString());

063

064 // using SmartThreadPool

065 //通過Amir Bar的SmartThreadPool線程池，利用QueueUserWorkItem方法，按隊列執(zhí)行10個任務(wù)

066 Console.WriteLine();

067 Console.WriteLine("SmartThreadPool:");

068 InitThreadPoolCounters();

069 QueueSmartThreadPoolThreads();

070 while (threadDone != 10) { };

071 Console.WriteLine("SmartThreadPool: 10 simultaneous threads:");

072 total = 0;

073 for (int i = 0; i < 10; i++)

074 {

075 Console.WriteLine("T" + i.ToString() + " = " + threadPoolCounters[i].ToString() + " ");

076 total += threadPoolCounters[i];

077 }

078 Console.WriteLine("Total = " + total.ToString());

079

080 // using ManagedThreadPool

081 //通過Stephen Toub改進(jìn)后的線程池，利用QueueUserWorkItem方法，按隊列執(zhí)行10個任務(wù)

082 Console.WriteLine();

083 Console.WriteLine("ManagedThreadPool:");

084 InitThreadPoolCounters();

085 QueueManagedThreadPoolThreads();

086 while (threadDone != 10) { };

087 Console.WriteLine("ManagedThreadPool: 10 simultaneous threads:");

088 total = 0;

089 for (int i = 0; i < 10; i++)

090 {

091 Console.WriteLine("T" + i.ToString() + " = " + threadPoolCounters[i].ToString() + " ");

092 total += threadPoolCounters[i];

093 }

094 Console.WriteLine("Total = " + total.ToString());

095

096 // using C#4.0 Parallel

097 //通過Tasks.Parallel.For進(jìn)行并行運算

098 Console.WriteLine();

099 Console.WriteLine("Parallel:");

100 InitThreadPoolCounters();

101 UseParallelTasks();

102 while (threadDone != 10) { };

103 Console.WriteLine("Parallel: 10 simultaneous threads:");

104 total = 0;

105 for (int i = 0; i < 10; i++)

106 {

107 Console.WriteLine("T" + i.ToString() + " = " + threadPoolCounters[i].ToString() + " ");

108 total += threadPoolCounters[i];

109 }

110 Console.WriteLine("Total = " + total.ToString());

111 }

我們可以先熟悉一下大致思路。代碼中，我們主要依靠輸出的數(shù)字count或者total來判斷哪個方法執(zhí)行效率更高（原文是How Hign Can I Count?），通常輸出的數(shù)字越大，我們就認(rèn)為它”干的活越多“，效率越高。主要實現(xiàn)過程就是通過一個靜態(tài)的System.Timers.Timer對象的timer實例，設(shè)置它的Interval屬性和ElapsedEventHandler事件：

1 static System.Timers.Timer timer = new System.Timers.Timer(1000);

2 /*當(dāng) AutoReset 設(shè)置為 false 時，Timer 只在第一個 Interval 過后引發(fā)一次 Elapsed 事件。

3 若要保持以 Interval 時間間隔引發(fā) Elapsed 事件，請將 AutoReset 設(shè)置為 true。*/

4 timer.AutoReset = false;

5 timer.Elapsed += new ElapsedEventHandler(OnTimerEvent);//當(dāng)timer.Start()時，觸發(fā)事件

其中，timer的事件觸發(fā)的函數(shù)：

?1 static void OnTimerEvent(object src, ElapsedEventArgs e)

2 {

3 done = true;

4 }

每次timer.Start執(zhí)行的時候，一次測試就將開始，這樣可以確保測試的不同方法都在1000毫秒內(nèi)跑完。

下面開始具體介紹幾個方法：

A、線程

這個非常簡單，就是通過主線程計算在1000毫秒內(nèi)，count從0遞增加到了多少：

01 /// <summary>

02 /// 單一線程，一跑到底

03 /// </summary>

04 /// <returns></returns>

05 static decimal SingleThreadTest()

06 {

07 done = false;

08 decimal counter = 0;

09 timer.Start();

10 while (!done)

11 {

12 ++counter;

13 }

14 return counter;

15 }

while判斷可以保證方法在1000毫秒內(nèi)執(zhí)行完成。



B、多線程

這個多線程方法比較折騰，先創(chuàng)建線程，然后運行，最后銷毀線程，這就是一個線程執(zhí)行單元，重復(fù)10次這個線程執(zhí)行單元。

01 /// <summary>

02 /// 創(chuàng)建一個線程，運算，然后銷毀該線程 重復(fù)前面的動作

03 /// </summary>

04 static void CreateAndDestroyTest()

05 {

06 done = false;

07 timer.Start();

08 while (!done)

09 {

10 Thread counterThread = new Thread(new ThreadStart(Count1Thread));

11 counterThread.IsBackground = true;//后臺線程

12 counterThread.Start();

13 while (counterThread.IsAlive) { };

14 }

15 }

那個ThreadStart委托對應(yīng)的方法Count1Thread如下：

?1 static void Count1Thread()

2 {

3 ++count2; //靜態(tài)字段count2自增

4 }

從表面上看，大家估計都可以猜到，效果可能不佳。



C、還是多線程

這個方法不判斷線程的執(zhí)行狀態(tài)，不用等到一個線程銷毀后再創(chuàng)建一個線程，然后執(zhí)行線程方法。線程執(zhí)行的方法就是根據(jù)線程的Name找到一個指定數(shù)組的某一索引，并累加改變數(shù)組的值：

01 /// <summary>

02 /// 將數(shù)組和線程數(shù)標(biāo)志threadDone回到初始狀態(tài)

03 /// </summary>

04 static void InitThreadPoolCounters()

05 {

06 threadDone = 0;

07 for (int i = 0; i < 10; i++)

08 {

09 threadPoolCounters[i] = 0;

10 }

11 }

12

13 /// <summary>

14 /// 初始化10個線程

15 /// </summary>

16 static void InitThreads()

17 {

18 for (int i = 0; i < 10; i++)

19 {

20 threads[i] = new Thread(new ThreadStart(Count2Thread));

21 threads[i].IsBackground = true;

22 threads[i].Name = i.ToString();//將當(dāng)前線程的Name賦值為數(shù)組索引，在Count2Thread方法中獲取對應(yīng)數(shù)組

23 }

24 }

25

26 /// <summary>

27 /// 開始多線程運算

28 /// </summary>

29 static void StartThreads()

30 {

31 done = false;

32 timer.Start();

33 for (int i = 0; i < 10; i++)

34 {

35 threads[i].Start();

36 }

37 }

其中，每一個線程需要執(zhí)行的委托方法

1 static void Count2Thread()

2 {

3 int n = Convert.ToInt32(Thread.CurrentThread.Name);//取數(shù)組索引

4 while (!done)

5 {

6 ++threadPoolCounters[n];

7 }

8 Interlocked.Increment(ref threadDone);//以原子操作的形式保證threadDone遞增

9 }

在測試過程中，我們看代碼：

01 // Create 10 threads and run them simultaneously

02 //一次性創(chuàng)建10個線程，然后遍歷使線程執(zhí)行運算

03 Console.WriteLine();

04 InitThreadPoolCounters();

05 InitThreads();

06 StartThreads();

07 while (threadDone != 10) { };

08 Console.WriteLine("10 simultaneous threads:");

09 for (int i = 0; i < 10; i++)

10 {

11 Console.WriteLine("T" + i.ToString() + " = " + threadPoolCounters[i].ToString() + " ");

12 total += threadPoolCounters[i];

13 }

14 Console.WriteLine("Total = " + total.ToString());

15 Console.WriteLine();

最后算出這個數(shù)組的所有元素的總和，就是這10個線程在1000毫秒內(nèi)所做的事情。其中， while (threadDone != 10) { };這個判斷非常重要。這個方法看上去沒心沒肺，線程創(chuàng)建好就不管它的死活了（還是管活不管死？），所以效率應(yīng)該不低。

實際上，我在本地測試并看了一下輸出，表面看來，按count大小逆序排列：C>A>B，這就說明多線程并不一定比單線程運行效率高。其實B之所以效率不佳，主要是由于這個方法大部分的”精力“花在線程的執(zhí)行狀態(tài)和銷毀處理上。

注意，其實C和A、B都沒有可比性，因為C計算的是數(shù)組的總和，而A和B只是簡單的對一個數(shù)字進(jìn)行自加。

ps：C這一塊說的沒有中心，想到哪寫到哪，所以看起來寫得很亂，如果看到這里您還覺著不知所云，建議先下載最后的demo，先看代碼，再對照這篇文章。

好了，到這里，我們對線程的創(chuàng)建和使用應(yīng)該有了初步的了解。細(xì)心的人可能會發(fā)現(xiàn)，我們new一個Thread，然后給線程實例設(shè)置屬性，比如是否后臺線程等等，其實這部分工作可以交給下面介紹的線程池ThreadPool來做（D、E和F主要介紹線程池）。



D、線程池ThreadPool

在實際的項目中大家可能使用最多最熟悉的就是這個類了，所以沒什么可說的：

01 /// <summary>

02 /// ThreadPool測試

03 /// </summary>

04 static void QueueThreadPoolThreads()

05 {

06 done = false;

07 for (int i = 0; i < 10; i++)

08 {

09 ThreadPool.QueueUserWorkItem(new WaitCallback(Count3Thread), i);

10 }

11

12 timer.Start();

13 }

14

15 static void Count3Thread(object state)

16 {

17 int n = (int)state;

18 while (!done)

19 {

20 ++threadPoolCounters[n];

21 }

22 Interlocked.Increment(ref threadDone);

23 }

我們知道線程池里的線程默認(rèn)都是后臺線程，所以它實際上簡化了線程的屬性設(shè)置，更方便異步編程。

需要說明的是，線程池使用過程中會有這樣那樣的缺陷（雖然本文的幾個線程池任務(wù)都不會受這種缺陷影響）。比如，我們一次性向線程池中加入100個任務(wù)，但是當(dāng)前的系統(tǒng)可能只支持25個線程，并且每個線程正處于”忙碌“狀態(tài)，如果一次性加入池中系統(tǒng)會處理不過來，那么多余的任務(wù)必須等待，這就造成等待的時間過長，系統(tǒng)無法響應(yīng)。還好，ThreadPool提供了GetAvailableThreads方法，可以讓你知道當(dāng)前可用的工作線程數(shù)量。

01 static void QueueThreadPoolThreads()

02 {

03 done = false;

04 for (int i = 0; i < 10; i++)

05 {

06 //ThreadPool.QueueUserWorkItem(new WaitCallback(Count3Thread), i); //直接給程序池添加任務(wù)有時是很草率的

07

08 WaitCallback wcb = new WaitCallback(Count3Thread);

09 int workerThreads, availabeThreads;

10 ThreadPool.GetAvailableThreads(out workerThreads, out availabeThreads);

11 if (workerThreads > 0)//可用線程數(shù)>0

12 {

13 ThreadPool.QueueUserWorkItem(wcb, i);

14 }

15 else

16 {

17 //to do 可以采取一種策略，讓這個任務(wù)合理地分配給線程

18 }

19 }

如果沒有可用的工作線程數(shù)，必須設(shè)計一定的策略，讓這個任務(wù)合理地分配給線程。

也許就是類似于上面那樣的限制，很多開發(fā)者都自己創(chuàng)建自己的線程池，同時也就有了后面的SmartThreadPool和ManagedThreadPool大展身手的機(jī)會。



E、線程池SmartThreadPool

大名鼎鼎的SmartThreadPool，但是我從來沒在項目中使用過，所以只是找了一段簡單的代碼測試一下：

01 /// <summary>

02 /// SmartThreadPool測試

03 /// </summary>

04 static void QueueSmartThreadPoolThreads()

05 {

06 SmartThreadPool smartThreadPool = new SmartThreadPool();

07 // Create a work items group that processes

08 // one work item at a time

09 IWorkItemsGroup wig = smartThreadPool.CreateWorkItemsGroup(1);

10

11 done = false;

12 timer.Start();

13 for (int i = 0; i < 10; i++)

14 {

15 wig.QueueWorkItem(new WorkItemCallback(Count4Thread), i);

16 }

17 // Wait for the completion of all work items in the work items group

18 wig.WaitForIdle();

19 smartThreadPool.Shutdown();

20 }

21

22 static object Count4Thread(object state)

23 {

24 int n = (int)state;

25 while (!done)

26 {

27 ++threadPoolCounters[n];

28 }

29 Interlocked.Increment(ref threadDone);

30 return null;

31 }

自從收藏這個SmartThreadPool.dll后，我還從沒有在項目中使用過。查看它的源碼注釋挺少也挺亂的，不知道有沒有高人知道它的一個效率更好的方法。您也可以看看英文原文，自己嘗試體驗一下。如果您熟悉使用SmartThreadPool，歡迎討論。



F、線程池ManagedThreadPool

Stephen Toub這個完全用C#托管代碼實現(xiàn)的線程池也非常有名，在Marc Clifton的英文原文中，作者也不吝溢美之詞，贊它“quite excellent”，于我心有戚戚焉：

01 /// <summary>

02 /// ManagedThreadPool測試

03 /// </summary>

04 static void QueueManagedThreadPoolThreads()

05 {

06 done = false;

07 timer.Start();

08 for (int i = 0; i < 10; i++)

09 {

10 Toub.Threading.ManagedThreadPool.QueueUserWorkItem(new WaitCallback(Count5Thread), i);

11 }

12 }

13 static void Count5Thread(object state)

14 {

15 int n = (int)state;

16 while (!done)

17 {

18 ++threadPoolCounters[n];

19 }

20 Interlocked.Increment(ref threadDone);

21 }


對于這個托管的線程池，我個人的理解，就是它在管理線程的時候，這個池里還有一個緩存線程的池，即一個ArrayList對象。它一開始就初始化了一定數(shù)量的線程，并通過ProcessQueuedItems方法保證異步執(zhí)行進(jìn)入池中的隊列任務(wù)（那個死循環(huán)有時可能導(dǎo)致CPU過分忙碌），這樣在分配異步任務(wù)的時候，就省去了頻繁去創(chuàng)建（new）一個線程。同時它在實現(xiàn)信號量（Semaphore）的同步和線程出入隊列的設(shè)計上都可圈可點，非常巧妙，強(qiáng)烈推薦您閱讀它的源碼。


G、并行運算

下面的示例，我只使用了簡單的System.Threading.Tasks.Parallel.For 對應(yīng)的for 循環(huán)的并行運算：

01 /// <summary>

02 /// 并行運算測試

03 /// </summary>

04 static void UseParallelTasks()

05 {

06 done = false;

07 timer.Start();

08 // System.Threading.Tasks.Parallel.For - for 循環(huán)的并行運算

09 System.Threading.Tasks.Parallel.For(0, 10, (i) => { Count6Thread(i); });

10 }

11 static void Count6Thread(object state)

12 {

13 int n = (int)state;

14 while (!done)

15 {

16 ++threadPoolCounters[n];

17 }

18 Interlocked.Increment(ref threadDone);

19 }

沒有什么要特殊說明的，就是新類庫的使用�？创a，好像比使用線程或線程池更加簡單直接，有機(jī)會爭取多用一用。我在本地測試的時候，在Release版本下，按照count的大小逆序排列，總體上G>D>F>E。需要注意到一件事，就是SmartThreadPool中排入隊列的任務(wù)是一個返回值為Object的委托類型，這和其他的幾個沒有返回的（void類型）不同。SmartThreadPool口碑還是不錯的，也許是我沒有正確使用它。


最后小結(jié)一下：本文主要列舉了C#中我所知道的幾種常見的異步處理的方法，歡迎大家糾錯或補(bǔ)充。