寫這個系列原本的想法是討論一下.NET中異步編程風格的變化，特別是F#中的異步工作流以及未來的.NET 5.0中的基于任務的異步編程模型。但經過前幾篇文章（為什么需要異步，傳統的異步編程，使用CPS及yield實現異步）的發表后，很多人對IO異步背后實現的原理以及為什么這樣能提高性能很感興趣。其實我本不想花更多的文字在這些底層實現的細節上，一來我并不擅長這些方面，二來我們使用.NET的異步IO就不需要關心這些底層東西，因為已經為你封裝完備了。不過為了避免大家一再在這上面商討，我還是在這個系列中間插入了一篇來解釋一下。

　　本文我將從內核對象IO完成端口開始介紹，然后來瞧瞧.NET BCL中的FileStream.BeginRead是如何利用IO完成端口來實現的。

　　IO完成端口(IO Completion Port)

　　大多數人應該或多或少地聽說過IO完成端口這么個東西，而且也知道它是實現高性能IO，高伸縮性應用的尚方寶劍。IO完成端口是一個非常復雜的內核對象，其實現的也非常巧妙，細細琢磨還是非常有意思的。

　　創建高伸縮性的應用的一個基本原則就是：創建更少的線程。線程數更少首先消耗的資源就少，每個線程的創建除了要浪費CPU時間外，還要創建一系列的數據結構用來保存線程相關的一些信息：用戶棧，線程上下文，內核棧等。這個總共加起來大概1.5M左右，那么你算算你的32位機器總共能使用多少內存?那么對應地能創建多少線程?可能有人講那對于64位的就無所謂了。嗯，在資源占用這方面64位確實不用擔心。但是系統中可運行的線程數越多，你的CPU數又是有限的(8個?80個?)。Windows的任務調度機制是每個線程會運行一個時間片，然后Windows搶占式的調度另一個線程運行。那么線程數越多，Windows勢必要進行更頻繁的線程上下文切換。線程上下文切換對系統性能的影響在這里我就不多說了，你可以搜搜資料。

　　那么如何做到創建更少的線程，而又干更多的事兒呢?答案就是“不等待”。相對CPU來說，IO設備的速度簡直低的要命。就好像飛機和拖拉機的差別一樣，我們可不能讓拖拉機拖了飛機的后退兒。而IO完成端口就是為了這個而生的：創建更少的線程，干更多的事兒。

　　IO完成端口首先不是一個我們看得見摸得著的什么插口，也和我們常說的80這樣的端口不同。你可以將其理解為一個數據結構或一個對象(下面我會用C#的代碼來輔助講解IO完成端口，僅僅是講解，這些代碼并不是真實的實現)：

　　Windows提供了一個CreateIoCompletionPort API來創建IO完成端口，實際上這個API有兩個作用：創建IO完成端口和將一個IO設備與該端口綁定。創建IO完成端口時有一個很重要的參數：指定同時最多能有多少個線程并行運行，這就是為了保證更少的線程，如果你將這個數值指定為0，那么默認值就會是你機器的CPU數。IO端口里還有一個IO設備句柄列表，你可以將很多設備句柄與這個端口綁定(文件、Socket等)：

//函數原型
HANDLE CreateIoCompletionPort(
//設備句柄
HANDLE hFile,
//已有的IO完成端口句柄，如果這里已經指定，則是將前面指定的設備與該端口綁定
HANDLE hExistingCompletionPort,
//因為一個IO完成端口可以綁定很多設備，可以用這個來區分
ULONG_PTR CompletionKey,
//允許同時運行的線程數
DWORD dwNumberOfConcurrentThreads
);
//創建一個IO完成端口
HANDLE hIoPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE,NULL,0,2);
//創建文件，如果要異步訪問文件則需要指定FILE_FLAG_OVERLAPPED
HANDLE hFile = CreateFile(..);
//將上面創建的文件句柄與剛才創建的IO完成端口綁定，不僅僅是文件可以
CreateIoCompletionPort(hFile,hIoPort,1,2);

　　除此之外，我們還要為該端口創建一些供使用的線程。然后讓這些線程調用Windows提供的GetQueuedCompletionStatus方法。這些線程調用了該方法后會被放到IO完成端口另外一個數據結構中：一個后進先出的隊列(我們將其稱為等待隊列吧)。然后該線程會休眠起來，不占用CPU。然后我們可以調用像ReadFile這樣的方法發起一個IO請求：

BOOL ReadFile(
HANDLE hFile,
PVOID pvBuffer,
DWORD nNumBytesToRead,
PDWORD pdwNumBytes,
OVERLAPPED* pOverlapped);

ReadFile(..
&overlapped);

　　上面代碼中的OVERLAPPED是一個非常重要的數據結構，后面會提到。

　　現在假設你的某個IO設備收到了一個數據包，Windows就會檢查這個IO設備是否跟一個IO完成端口關聯了，如果關聯了Windows就會把這個數據包投遞到這個IO完成端口。IO完成端口里還有另外一個先進先出的隊列，用來保存這些IO完成的數據。IO完成端口一看，唔，有個IO完成包投遞到我這兒來了，那我看看我的那個等待隊列里有沒有線程還在休息，如果有就叫它起來干活兒。

　　嘿，還真有一個家伙還在睡覺，如是IO完成端口就喚醒該線程，實際上就是上面的那個GetQueuedCompletionStatus方法返回了。該方法返回時還會得到一些別的信息：接收了多少個字節啊，是哪個設備啊，最重要的是上面提到的OVERLAPPED這個結構等等。起來后的線程就會拿著這些信息干一些后續的事兒：

BOOL GetQueuedCompletionStatus(
HANDLE hCompletionPort,
PDWORD pdwNumberOfBytesTransferred,
PULONG_PTR pCompletionKey,
OVERLAPPED **ppOverlapped,
DWORD dwMilliseconds);
//類似于下面的過程
//創建一個線程
Thread thread = new Thread(()=>{
while(true){
//如果沒有IO完成通知到達，該線程就在這里休眠了
if(GetQueuedCompletionStatus(hIoPort,..ppOverlapped..)){
//從ppOverlapped里取出所需的信息，比如可能設置了一個回調函數的指針等
}else{
//
}
}
});
thread.Start();

?　干完這個事兒后，這個線程又會回到剛才那個隊列繼續躺起來(其實是再次調用一下那個方法)。我們要注意的是，這個等待隊列是后進先出的，也就是說如果下次有消息來了很有可能還是上會那個線程來處理。這樣做的目的還是為了提高性能：不需要進行線程上下文切換。因為CPU的速度比IO設備的高出很多，大部分時候我們只需要一兩個線程就可以處理很多IO請求。

　　現在假設我們的機器有2個CPU，創建IO完成端口時我們指定了同時可以有2個線程運行。我們創建了4個線程放到等待隊列里。現在有4個IO完成包投遞過來了，放在那個隊列里。實際上IO完成端口只會喚醒兩個線程去執行，因為你指定了同時只能有兩個線程運行，那兩個線程運行完就會立馬回來繼續運行別的。但是現在出了一個狀況，其中有一個線程執行過程中因為等待某個資源被阻塞了。那現在只有一個線程執行了，那這個線程就有點吃力了。其實IO完成端口非常聰明，它內部還有一個暫停運行的線程列表和一個正在運行的線程列表。如果某個線程正在運行，它就把這個線程ID放到這個隊列里，當這個線程因為某個事兒暫停運行了它就會將其移動到另外一個列表中。IO完成端口會保證正在運行的線程列表里的數目不會超過你指定的最大并發數。一旦這個列表里的數目少于這個數，而IO完成包隊列里又有未處理的包，IO完成端口就會看看還有沒有在睡覺的線程，如果有就將其喚醒干活兒。

　　IO完成端口盡量的控制同時運行的線程數，減少上下文切換浪費的時間和資源，并且讓線程盡量的忙起來。

　　這里還有一個有意思的地方，假設現在正在運行的兩個線程其中一個調用Thread.Sleep休眠了，然后IO完成端口喚醒另外一個線程，讓同時運行的線程數保持為2個，不過過了一會兒剛才調用Sleep休眠的線程醒過來了，有意思的事情發生了：現在有三個線程同時運行，超過了我們設置的最大并行數。這個時候IO完成端口是不會殺掉一個線程的，它會讓它們繼續執行，然后等到執行完了再讓這個并行數降下去。

　　實際上，IO完成端口不僅僅可以用來處理這種異步的IO，它完全可以作為一種線程間的通訊機制來使用(與IO一點關系都沒有)，我們可以調用Win API PostQueuedCompletionStatus來模擬一次IO完成，這樣我們的IO完成端口就會接到通知，然后調用線程執行。熟悉并發里的Actor模型的同學可能覺得這有點Actor的影子了。

　　BeginRead&EndRead

　　那么，既然有IO完成端口這么個好東西，如是有很多人想在.NET里也利用利用。其實大可不必，在.NET里異步的IO內部就是使用了IO完成端口。每個CLR初始化后都會創建一個IO完成端口，用來處理IO請求。很多人應該知道ThreadPool里的線程分為兩類：worker thread和io completion thread，這里的io completion thread就是上一節說的跟IO完成端口相關聯的那些thread。要說它跟其他的thread有什么不同?沒什么不同，只是受IO完成端口控制而已。

　　為了看看在.NET中是如何利用IO完成端口的，我們將FileStream.BeginRead作為我們的入口點。在FileStream的Init方法里我們會看到這么一段代碼：

if (this._isAsync)
{
//...
try
{
flag4 = ThreadPool.BindHandle(this._handle);
}
finally
{
CodeAccessPermission.RevertAssert();
}
//...
}

　　我們感興趣的就是ThreadPool.BindHandle。還記得上面對IO完成端口的描述么?其實這里做的事兒就是將該文件句柄與每個CLR都初始化了的那個IO完成端口綁定。也就是說如果我們創建一個FileStream時指定了異步，那么IO完成端口就會“監視”這個文件。

　　我們再來看看BeginRead這個方法。該方法是用來發起異步IO請求的方法，該方法執行后會立即返回，不阻塞線程。

　　首先，看這么段代碼：

if (!this._isAsync)
{
return base.BeginRead(array, offset, numBytes, userCallback, stateObject);
}

　　也就是說如果我們創建FileStream時，沒有指定為異步，就會調用基類的BeginRead方法，那基類的這個方法又是如何實現的呢?

[HostProtection(SecurityAction.LinkDemand, ExternalThreading=true)]
public virtual IAsyncResult BeginRead(byte[] buffer, int offset, int count, AsyncCallback callback, object state)
{
//...
ReadDelegate delegate2 = new ReadDelegate(this.Read);
//...
return delegate2.BeginInvoke(buffer, offset, count, callback, state);
}

　　其實是創建一個調用同步的Read方法的委托，然后調用一下BeginInvoke方法(在第二篇文章已經說過，這樣的調用實際上還是讓線程池里的一個線程來調用，我們可以稱之為一種偽異步IO)。這里可以得出一個結論：如果你想用BeginRead，那么初始化FileStream的時候就指定異步，否則就不用直接用Read。

　　那么如果創建FileStream的時候指定了異步會是什么結果呢(這里的實現在BeginReadCore方法里)?

[SecuritySafeCritical]
private unsafe FileStreamAsyncResult BeginReadCore(byte[] bytes, int offset, int numBytes, AsyncCallback userCallback, object stateObject, int numBufferedBytesRead)
{
NativeOverlapped* overlappedPtr;
FileStreamAsyncResult ar = new FileStreamAsyncResult {
_handle = this._handle,
_userCallback = userCallback,
_userStateObject = stateObject,
_isWrite = false,
_numBufferedBytes = numBufferedBytesRead
};
ManualResetEvent event2 = new ManualResetEvent(false);
ar._waitHandle = event2;
Overlapped overlapped = new Overlapped(0, 0, IntPtr.Zero, ar);
//...
overlappedPtr = overlapped.Pack(IOCallback, bytes);
//...
ar._overlapped = overlappedPtr;


//...
ReadFileNative(this._handle, bytes, offset, numBytes, overlappedPtr, out hr)

　　上面代碼中的NativeOverlapped就是在上一節我們提到的保存有回調等信息的OVERLAPPED結構，在這里也是一樣，它保存有我們的userCallback回調。然后通過調用ReadNative發起IO請求，并將這個數據結構傳遞進去，這里的ReadNative就是對Win32 的ReadFile的封裝。發起異步IO請求完畢，BeginRead返回，過了一會兒磁盤驅動程序將數據讀回來了，對應的IO完成端口收到通知，IO完成端口把剛才傳遞進去的NativeOverlapped結構傳遞給IO線程，IO線程從中取出IOCallback回調，IOCallback回調里有對我們的userCallback回調的調用：

IOCallback = new IOCompletionCallback(FileStream.AsyncFSCallback);
private static unsafe void AsyncFSCallback(uint errorCode, uint numBytes, NativeOverlapped* pOverlapped)
{
FileStreamAsyncResult asyncResult = (FileStreamAsyncResult) Overlapped.Unpack(pOverlapped).AsyncResult;
//...
AsyncCallback callback = asyncResult._userCallback;
if (callback != null)
{
callback(asyncResult);
}
}

　　在這個回調里我們會對EndRead進行調用，我們看看EndRead的代碼會發現其他一些東西：

public override unsafe int EndRead(IAsyncResult asyncResult)
{
//...
WaitHandle handle = result._waitHandle;
if (handle != null)
{
try
{
handle.WaitOne();
}
finally
{
handle.Close();
}
}
NativeOverlapped* nativeOverlappedPtr = result._overlapped;
if (nativeOverlappedPtr != null)
{
Overlapped.Free(nativeOverlappedPtr);
}
//...
return (result._numBytes + result._numBufferedBytes);
}

　　首先是銷毀我們在BeginRead里初始化的WaitHandle內核對象，然后將NativeOverlapped結構也銷毀。所以EndRead除了取回讀了多少個字節的作用外，還起了銷毀資源的作用。所以有的時候我們想進行這么一個操作：異步的發起請求，但是我們并不關心該請求是否成功。如是我們就假想能不能只調用BeginXXX方法就可以了?從這里看我們不能簡單的調用一下BeginXXX就了事了，因為在BeginXXX里分配的一些句柄和內核資源需要在EndXXX里銷毀，不然會造成資源泄露。

　　總結

　　本文先介紹了一下IO完成端口的原理，然后打開FileStream的源代碼，看看.NET是如何利用IO完成端口進行異步IO請求的。IO完成端口是一種非常高效的編程方式，所以如果我們想構建界面響應靈敏或高可伸縮性的服務應用，如果你的應用又是IO密集型的，那么你應該仔細的設計你的應用，利用異步IO的優勢。

　　但是我們切忌拿著錘子就是釘子，發現IO完成端口這把利器就到處使用，甚至都不去思考是否值得。從前面幾篇文章以及本文的分析來看，如果我們使用同步的方式那么一切都是在同一個方法內部完成，分配的一些對象的引用也都是在棧上完成，所以本方法退出后這些分配的資源都可以高效的回收。

　　但是如果使用異步IO的話我們無法在一個方法內完成，所以很多東西的生命周期無形的延長了，本方法退出后還是不能銷毀;這是其一，其二是要利用IO完成端口就必須進行額外的平臺調用(從.NET調入到CLR甚至到Windows內核)，這些調用都是非常昂貴的。所以如果我們在構建服務器應用時，如果應用的規模并不是非常大，我們還是應該首選同步的方式，這樣編程更容易，消耗的資源也更少。當然，這都是基于你實際的應用經過不斷的嘗試和調整得出的。

　　我希望本文能對某些同學心中懷疑BeginRead到底占不占用線程，IO線程又是個什么東西起到釋疑的作用。

總結

以上是生活随笔為你收集整理的一起谈.NET技术，.NET异步编程：IO完成端口与BeginRead的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。