淺談.NET下的多線程和并行計算（十）.NET異步程式設計模型基礎上

談多線程談到現在，我們要明确多線程的一個好處是可以進行并行的運算（充分利用多核處理器，對于桌面應用程式來說就更重要一點了，沒有WEB伺服器，利用多核隻能靠自己），還有一個好處就是異步操作，就是我們可以讓某個長時間的操作獨立運作，不妨礙主線程繼續進行一些計算，然後異步的去傳回結果（也可以不傳回）。前者能提高性能是因為能利用到多核，而後者能提高性能是因為能讓CPU不在等待中白白浪費，其實異步從廣義上來說也可以了解為某種并行的運算。在之前的這麼多例子中，我們大多采用手工方式來新開線程，之前也說過了，在大并發的環境中随便開始和結束線程的代價太大，需要利用線程池，使用線程池的話又覺得少了一些控制。現在讓我們來總結一下大概會有哪幾種常見的異步程式設計應用模式：

1） 新開一個A線程執行一個任務，然後主線程執行另一個任務後等待線程傳回結果後繼續

2） 新開一個A線程執行一個任務，然後主線程不斷輪詢A線程是否執行完畢，如果沒有的話可以選擇等待或是再進行一些操作

3） 新開一個A線程執行一個任務，執行完畢之後立即執行一個回調方法去更新一些狀态變量，主線程和A線程不一定有直接互動

4） 新開一個A線程執行一個任務，執行完畢之後啥都不做

（補充一句，異步程式設計不一定是依賴于線程的，從廣義上來說，使用隊列異步處理資料也可以算是一種異步程式設計模式）

對于這任何一種，我們要使用線程池來編寫應用的話都是比較麻煩的，比如如下的代碼實作了1）這種應用：

class AsyncObj
{
    public EventWaitHandle AsyncWaitHandle { get; set; }
    public object Result { get; set; }

    public AsyncObj()
    {
        AsyncWaitHandle = new AutoResetEvent(false);
    }
}      

AsyncObj ao = new AsyncObj();
ThreadPool.QueueUserWorkItem(state =>
{
    AsyncObj obj = state as AsyncObj;
    Console.WriteLine("asyc operation started @ " + DateTime.Now.ToString("mm:ss"));
    Thread.Sleep(2000);
    Console.WriteLine("asyc operation completed @ " + DateTime.Now.ToString("mm:ss"));
    obj.Result = 100;
    obj.AsyncWaitHandle.Set();
}, ao);

Console.WriteLine("main operation started @ " + DateTime.Now.ToString("mm:ss"));
Thread.Sleep(1000);
Console.WriteLine("main operation completed @ " + DateTime.Now.ToString("mm:ss"));
ao.AsyncWaitHandle.WaitOne();
Console.WriteLine("get syc operation result : " + ao.Result.ToString() + " @ " + DateTime.Now.ToString("mm:ss"));      

結果如下：

對于2）-4）等情況又是另外一套了，這樣我們的代碼可能會變得亂七八糟，在.NET中我們的委托以及很多IO操作相關的類庫都支援一種叫做異步程式設計模型APM的程式設計模型。不僅僅友善了我們進行多線程應用，而且我們如果自己要設計類庫的話也可以遵從這個模式（基于APM的接口實作我們自己的類庫）。.NET提供了基于IAsyncResult的異步程式設計模型和基于事件的異步程式設計模型，這節我們來看看基于IAsyncResult也就是BeginInvoke和EndInvoke（對于非同用的操作來說就是BeginXXX和EndXXX）的程式設計模型的各種使用方法，可以說這麼多種使用方法可以滿足我們大部分的要求。

首先來定義一個異步操作：

static int AsyncOperation(int x, int y)
{
    Console.WriteLine("asyc operation started @ " + DateTime.Now.ToString("mm:ss"));
    Thread.Sleep(2000);
    int a, b;
    ThreadPool.GetAvailableThreads(out a, out b);
    Console.WriteLine(string.Format("({0}/{1}) #{2}", a, b, Thread.CurrentThread.ManagedThreadId));
    Console.WriteLine("asyc operation completed @ " + DateTime.Now.ToString("mm:ss"));
    return x + y;
}      

我們需要開兩個線程同時計算兩個異步操作，然後主線程等待兩個線程執行完畢後擷取結果并且輸出它們的和，難以想象代碼是多麼簡單：

var func = new Func<int, int, int>(AsyncOperation);
var result1 = func.BeginInvoke(100, 200, null, null);
var result2 = func.BeginInvoke(300, 400, null, null);
Console.WriteLine("main operation started @ " + DateTime.Now.ToString("mm:ss"));
Thread.Sleep(1000);
Console.WriteLine("main operation completed @ " + DateTime.Now.ToString("mm:ss"));
int result = func.EndInvoke(result1) + func.EndInvoke(result2);
Console.WriteLine("get syc operation result : " + result + " @ " + DateTime.Now.ToString("mm:ss"));      

主線程的計算需要1秒，兩個異步線程都需要2秒，整個程式理論上需要2秒執行完畢，看看結果如何：

當然，在之前我們限制了線程池的線程數為2-4：

ThreadPool.SetMinThreads(2, 2);
ThreadPool.SetMaxThreads(4, 4);      

從結果中可以看出，使用委托來異步調用方法基于線程池，調用EndInvoke的時候阻塞了主線程，得到結果後主線程繼續。在代碼中沒看到Thread沒看到ThreadPool沒看到信号量，我們卻完成了一個異步操作，實作了一開始說的1）場景。現在再來看看第二種使用方式：

var func = new Func<string, int, string>(AsyncOperation2);
var result1 = func.BeginInvoke("hello ", 2000, null, null);
var result2 = func.BeginInvoke("world ", 3000, null, null);
Console.WriteLine("main operation started @ " + DateTime.Now.ToString("mm:ss"));
Thread.Sleep(1000);
Console.WriteLine("main operation completed @ " + DateTime.Now.ToString("mm:ss"));
WaitHandle.WaitAny(new WaitHandle[] { result1.AsyncWaitHandle, result2.AsyncWaitHandle });
string r1 = result1.IsCompleted ? func.EndInvoke(result1) : string.Empty;
string r2 = result2.IsCompleted ? func.EndInvoke(result2) : string.Empty;
if (string.IsNullOrEmpty(r1))
{
    Console.WriteLine("get syc operation result : " + r2 + " @ " + DateTime.Now.ToString("mm:ss"));
    func.EndInvoke(result1);
}
if (string.IsNullOrEmpty(r2))
{
    Console.WriteLine("get syc operation result : " + r1 + " @ " + DateTime.Now.ToString("mm:ss"));
    func.EndInvoke(result2);
}      

BeginInvoke傳回的是一個IAsyncResult，通過其AsyncWaitHandle 屬性來擷取WaitHandle。異步調用完成時會發出信号量。這樣我們就可以更靈活一些了，可以在需要的時候去WaitOne()（可以設定逾時時間），也可以WaitAny()或是WaitAll()，上例我們實作的效果是開了2個線程一個3秒，一個2秒，隻要有任何一個完成就擷取其結果，主線程任務完成之後再去EndInvoke沒完成的那個來釋放資源（比如有兩個排序算法，它們哪個快取決于資料源，我們一起執行并且隻要有一個得到結果就繼續）。在這裡我們的工作方法AsyncOperation2的定義如下：

static string AsyncOperation2(string s, int time)
{
    Console.WriteLine("asyc operation started @ " + DateTime.Now.ToString("mm:ss:fff"));
    Thread.Sleep(time);
    int a, b;
    ThreadPool.GetAvailableThreads(out a, out b);
    Console.WriteLine(string.Format("({0}/{1}) #{2}", a, b, Thread.CurrentThread.ManagedThreadId));
    Console.WriteLine("asyc operation completed @ " + DateTime.Now.ToString("mm:ss:fff"));
    return s.ToUpper();
}      

這段程式運作結果如下：

可以看到，在2秒的那個線程結束後，主線程就繼續了，然後再是3秒的那個線程結束。再來看看第三種，也就是使用輪詢的方式來等待結果：

var func = new Func<int, int, int>(AsyncOperation);
var result = func.BeginInvoke(100, 200, null, null);
Console.WriteLine("main operation started @ " + DateTime.Now.ToString("mm:ss"));
Thread.Sleep(1000);
Console.WriteLine("main operation completed @ " + DateTime.Now.ToString("mm:ss"));
while (!result.IsCompleted)
{
    Console.WriteLine("main thread wait again");
    Thread.Sleep(500);
}
int r = func.EndInvoke(result);
Console.WriteLine("get syc operation result : " + r + " @ " + DateTime.Now.ToString("mm:ss"));      

程式的輸出結果如下，這對應我們一開始提到的第二種場景，在等待的時候我們的主線程還可以做一些（不依賴于傳回結果的）計算呢：

再來看看第四種，采用回調的方式來擷取結果，線程在結束後自動調用回調方法，我們可以在回調方法中進行EndInvoke：

var func = new Func<int, int, int>(AsyncOperation);
var result = func.BeginInvoke(100, 200, CallbackMethod, func);
Console.WriteLine("main operation started @ " + DateTime.Now.ToString("mm:ss"));
Thread.Sleep(1000);
Console.WriteLine("main operation completed @ " + DateTime.Now.ToString("mm:ss"));
Console.ReadLine();      

BeginInvoke的第三個參數是回調方法，第四個參數是傳給工作方法的狀态變量，這裡我們把工作方法的委托傳給它，這樣我們可以在回調方法中擷取到這個委托：

static void CallbackMethod(IAsyncResult ar)
{
    Console.WriteLine(string.Format("CallbackMethod runs on #{0}", Thread.CurrentThread.ManagedThreadId));
    var caller = (Func<int, int, int>)ar.AsyncState;
    int r = caller.EndInvoke(ar);
    Console.WriteLine("get syc operation result : " + r + " @ " + DateTime.Now.ToString("mm:ss"));
}      

程式的輸出結果如下：

可以看到，主線程并沒有因為工作線程而阻塞，它沒有等待它的結果，異步方法結束後自動調用回調方法（運作于新線程），在回調方法中我們把狀态變量進行類型轉換後得到方法委托，然後通過這個委托來調用EndInvoke獲得結果。這裡符合我們第3）種應用，這種情況下主線程不一定需要和異步方法進行直接的互動（也就無需等待），當然主線程也完全可以再結合使用輪詢或等待信号量等待異步線程完成後從共享變量（需要回調方法把結果寫入共享變量）來擷取結果。

至于一開始說的第4）種應用需要注意，我們完全可以直接采用線程池來做，如果采用異步程式設計模型的話，即使不需要得到結果也别忘記調用EndInvoke來釋放資源，這是一個好習慣，因為.NET中很多涉及到IO和網絡操作的類庫都采用了APM方式，對于這些應用如果我們不調用EndInvoke來釋放非托管資源的話，GC恐怕無能為力的。下節繼續讨論基于事件的異步程式設計模式。

作者：

lovecindywang

本文版權歸作者和部落格園共有，歡迎轉載，但未經作者同意必須保留此段聲明，且在文章頁面明顯位置給出原文連接配接，否則保留追究法律責任的權利。