<<3D游戲程序設計大師技巧>>這本書.
網上討論完成端口資料很多,大寶收集的最多,有些是錯誤的,有些說的比較模糊.我就進可能清晰說明一下完成端口在游戲開發中一般模型.并解

決幾個難點問題.
由于完成端口是多線模型(當然可以把工作線程設定為一個)所以設計到資源的時候要線程安全,所以我開始簡單封裝了一下幾個stl的容器.
/***************************************以下代碼要與后邊的代碼一起編譯*******************************************/
//memorypool.h
//particle_allocator.h
#pragma once
#include <iostream>
#include <list>
#include <queue>
#include <windows.h>
using namespace std;

//關鍵區鎖
class CLock
{
????CRITICAL_SECTION _crisection;
public:
????CLock()????
????{
????InitializeCriticalSection( &_crisection );
????}
????~CLock()??
????{
????DeleteCriticalSection( &_crisection );????
????}
????void Lock()
????{
????EnterCriticalSection( &_crisection );??????
????}
????void Unlock()
????{
????LeaveCriticalSection( &_crisection );??????
????}
};

//鏈表模板
template <class T>
class CList : public list<T>
{

public:
????CList(void){};
????virtual ~CList(void){};
????CLock _guard;

????DWORD Size()
????{
????DWORD dwSize = 0;
????_guard.Lock();
????dwSize = size();
????_guard.Unlock();
????return dwSize;
????};
????void Clear()
????{
????_guard.Lock();
????clear();
????_guard.Unlock();
????}

????//添加數據到鏈表末尾，并返回添加后的該數據的節點指針
????iterator Push_Back(T lpData)
????{
????iterator lpNode = NULL;
????_guard.Lock();
????lpNode = insert(end(), lpData);
????_guard.Unlock();
????return lpNode;
????};

????//刪除一個節點
????void Erase(iterator lpNode)
????{
????_guard.Lock();
????erase(lpNode);
????_guard.Unlock();
????};

};

template <class T>
class CQueue : private CLock
{
private:
????queue<T> m_Queue;

public:
????CQueue(void) {};
????virtual ~CQueue(void) {};

????DWORD GetSize()
????{
????return (DWORD)m_Queue.size();
????};

????virtual void Push(T lpData)
????{
????Lock();
????m_Queue.push(lpData);
????Unlock();
????};

????virtual T Pop()
????{
????T lpData = NULL;
????Lock();
????if(m_Queue.size())
????{
????????lpData = m_Queue.front();
????????m_Queue.pop();
????}
????Unlock();
????return lpData;
????};
};
template<class ElementClass, int NumElement>
class CMemoryPool
{
????enum
????{
????chunk_size =NumElement
????};
????typedef unsigned char byte;
????list<byte *>??chunks;//內存指針
????queue<ElementClass *> free_list;//空閑塊隊列
????CLock guard;
public:
????CMemoryPool()
????{
????InitMemPool();
????}
????~CMemoryPool()
????{
????DestroyMemPool();
????}
????//初時化內存池
????void InitMemPool()
????{
????guard.Lock();
????used = 0;
????free = 0;
????//byte *memory;
????//memory = new byte[chunk_size*sizeof(ElementClass)];
????//if (!memory)
????//{
????//????cout << "內存分配出錯....."<< endl;
????//????return ;
????//}
????//chunks.push_front(memory);
????//for(int i =0;i< chunk_size;i++,free++)
????//{
????//????ElementClass *newnode = (ElementClass *)(memory + i*sizeof(ElementClass));
????//????free_list.push(newnode);

????//}

????guard.Unlock();
????}
????//銷毀內存池
????void DestroyMemPool()
????{
????while(!free_list.empty())
????{
????????free_list.pop();
????}
????for(std::list<byte *>::iterator all_iter = chunks.begin();all_iter!=chunks.end();++all_iter)
????{
????????delete [](*all_iter);
????}
????chunks.clear();
????used = 0;
????free= 0;
????}
????//從內存池里分配一塊內存
????ElementClass* MemPoolAlloc()
????{
????guard.Lock();
????byte *memory;
????if(free_list.empty())
????{
????????memory = new byte[chunk_size*sizeof(ElementClass)];
????????if (!memory)
????????{
????????cout << "內存分配出錯....."<< endl;
????????return NULL;
????????}
????????chunks.push_front(memory);
????????for(int i =0;i< chunk_size;i++,free++)
????????{
????????ElementClass *newnode = (ElementClass *)(memory + i*sizeof(ElementClass));
????????free_list.push(newnode);

????????}
????}
????ElementClass??*redata = NULL;
????redata = free_list.front();
????ZeroMemory(redata,sizeof(ElementClass));
????free_list.pop();
????++used;
????--free;
????guard.Unlock();
????return redata;
????}
????//還原內存池
????void MemPoolFree(ElementClass??*p)
????{
????guard.Lock();
????ZeroMemory(p,sizeof(ElementClass));
????free_list.push(p);
????-- used;
????++ free;
????guard.Unlock();
????}
????//內存池信息統計.
????void GetMemPoolInfo()
????{
????cout << "used is "
????????<< used
????????<< " ;??free is "
????????<< free << endl;
????}
private:
????UINT used;
????UINT free;
};
/********************************************************************************************************************/
代碼都非常簡單.相信有點c++基礎的朋友都能看懂.我只是簡單說一下內存池代碼.內存池里ElementClass類型一定要結構形的.也就是那種POD

類型還是什么類型我剛剛聽說這個名詞,所以記不住.內存池第一次分派的時候,分派NumElement>個ElementClass類型節點,這些節點構成一個隊

列free_list,這些都是通過模版參數傳入的.每次用的時候就就從頭找出一個節點,釋放的時候又在尾部加入一個節點,很好理解.用完的時候再

分配NumElement>個ElementClass類型節點,節點到最后統一釋放.

接下來我們看看程序用到的頭文件
//iocpsever.cpp
#include <winsock2.h>
#include "memorypool.h"
#include <Mswsock.h>
#include <process.h>

using namespace std;

#pragma comment(lib, "ws2_32.lib")
#pragma comment(lib, "Mswsock.lib")

using namespace std;
1.頭文件有一個郁悶的問題就是winsock2.h與windows互相包含問題.不知道有什么完美解決辦法,我再項目中用afxsock.h替代winsock2.h.
需要用到庫兩個庫文件我們都通過編譯器指令連接進來.
2.完成端口模型用一句話來講就是,把所有的socket都綁定到完成端口上,通過完成端口統一來處理.綁定以后這些socket上有任何的情況,都通

過完成端口反映出來,達到簡化處理邏輯的作用.完成端口確實不會讓你失望,邏輯結構非常簡單.
3.我們首先為socket進行分類.服務器的socket可以分為三類,一類就是用于監聽的socket,監聽socket綁定時機是它成功綁定到某一個端口上的

時候我們就可以把它綁定到完成端口上.所以我把它放進初始化間斷.第二類客戶連接進入的socket,這個只能放進當連接進入完成端口,accept

事件觸發的時候我們把它綁定到完成端口上.第三類就是服務器連接出去的socket,我把它放進線程里,在線程內部進行綁定,因為這類socket相

對比較少.
4.接下來我們考慮連接問題.由于完成端口是一種異步模型,它工作機制和我們傳統的同步socket連接模型不一樣,我們先投遞許多連接消息,當

連接成功的時候,對應的完成端口上發生accept事件.這里的accept是我們自己定義的,并且在AcceptEx的時候通過重疊結構傳入的,你也可以任

意定義,但是我想你不會把accept定義成send,而send定義成 receive吧.由于我們在AcceptEx的時候傳入單io數據結構,所以一定要關閉完成端

口的時候把它釋放掉.關于資源的釋放我們放在下邊來考慮.話題不要扯遠,回來連接上來,我們到底該怎么樣投遞呢.如果我們一下投遞的過多,

就會造成資源浪費,過少如果用完該怎么辦.這樣客戶就不能那個連接了.通用的做法就是每次投遞10個,如果用完繼續投機10個.怎么樣才能知道

用完呢?需要在監聽socket上注冊一個FD_ACCEPT事件,當投遞用完的時候事件觸發.這就就可以繼續投機,我也把它放進線程里邊去做這件事.
????g_hAcceptExOverEvent = CreateEvent(NULL, FALSE, FALSE, NULL);
????if(!g_hAcceptExOverEvent)
????{
????return false;
????}
????//幫定事件，由于我們用AcceptEx是一個異步的投遞，這樣幫定之后，如果投遞的AcceptEx事件全部完成
????//則g_hAcceptExOverEvent事件得到通知，進而同步AcceptEx調用
????if(WSAEventSelect(Listen, g_hAcceptExOverEvent, FD_ACCEPT) == SOCKET_ERROR)
????{
????return false;
????}
????//由于開始是復位，變成置位
????SetEvent(g_hAcceptExOverEvent);
5.最后討論資源釋放問題,分析如下代碼
????closesocket(Listen);
????g_session._guard.Lock();
????for(CList<LPPER_HANDLE_DATA>::iterator lpNode = g_session.begin(); lpNode != g_session.end();lpNode++)
????{
????closesocket((*lpNode)->Socket);
????printf(" close socket??= %d\n",(*lpNode)->Socket);
????}
????g_session._guard.Unlock();
????Sleep(1000);
????//向IOCP發送結束線程信號

????for(DWORD i = 0; i < threadcnt; i++)
????PostQueuedCompletionStatus(CompletionPort, 0, NULL, NULL);

????//等待工作線程結束，等待時間10秒

????if(WaitForMultipleObjects(threadcnt, m_arrayIOCPThreadHandle, TRUE, 10000) != WAIT_OBJECT_0)
????{
????//如果10秒內沒有結束所有線程，就強制結束
????for(DWORD i = 0; i < threadcnt; i++)
????????TerminateThread(m_arrayIOCPThreadHandle[i], 0);
????}
????//關閉所有工作線程句柄
????for(DWORD i = 0; i < threadcnt; i++)
????CloseHandle(m_arrayIOCPThreadHandle[i]);
????//釋放線程句柄數組
????delete [] m_arrayIOCPThreadHandle;
????m_arrayIOCPThreadHandle = NULL;
????CloseHandle(CompletionPort);
????g_per_io_data.DestroyMemPool();
????g_per_handle_data.DestroyMemPool();
????g_session.Clear();
我們也分三類來考慮
首先第一類監聽socket,監聽socket關閉之后,由于連接事件與監聽socket有關系.所以所有未完成的連接都會從完成端口返回來.這些連接里我

們分配了單io數據結構一定要釋放.
釋放邏輯如下

bRet = GetQueuedCompletionStatus(CompletionPort,
????????&BytesTransferred,
????????(PULONG_PTR)
????????&lpCompletionKey,
????????(LPOVERLAPPED*)
????????&lpPerIoData,
????????INFINITE);
????if(!bRet)//沒有取到任何東西
????{
????????if(!lpPerIoData)
????????continue;
????}
????//收到線程結束信號
????if(!lpCompletionKey)
????????return 0;

????LPPER_HANDLE_DATA lpSession = (LPPER_HANDLE_DATA)lpCompletionKey;

????if(!bRet)
????{
????????if(BytesTransferred == 0 )//斷開連接
????????{
????????if (lpPerIoData->OperationType == OP_READ)
????????{
????????????printf("client socket %d disconnect\n",lpPerIoData->Socket);
????????????closesocket(lpPerIoData->Socket); //關閉socket
????????????g_session.Erase(lpSession->node); //從Session鏈表中刪除
????????????g_per_handle_data.MemPoolFree(lpSession); //將Session歸還到池
????????????g_per_io_data.MemPoolFree(lpPerIoData);
????????????continue;
????????}

????????}
????}
????switch (lpPerIoData->OperationType)
????{
????case OP_ACCEPT://有連接進來
????????{
????????if (!bRet)
????????{
????????????closesocket(lpPerIoData->Socket);?? //關閉socket
????????????g_per_io_data.MemPoolFree(lpPerIoData);????//歸還結構體到內存池
????????????printf("關閉已經投遞空閑連接\n");
????????????break;
????????}
.........................后邊代碼省略掉當關閉監聽socket的時候GetQueuedCompletionStatus會返回false并且BytesTransferred等于零,

并且lpPerIoData,不為空lpPerIoData->OperationType操作符號就是我們AcceptEx傳入的自定義操作類型accept,這樣就釋放掉了與連接相關的

資源.所以我們在關掉監聽socket之后要sleep一下讓資源能充分釋放.
第二類就是客戶socket.
分兩種情況考慮.(1)服務器主動關閉,這又可以分為兩種情況,第一,當你在工作線程里關掉socket的時候GetQueuedCompletionStatus是收不到

任何消息的,所以直接釋放資源就是了.這樣產生的一個問題就是如果你把拆包邏輯放進完成端口工作線程的時候,當數據包發生錯誤的時候,你

想關閉socket,一定要記住連釋放掉與這個socket相關的資源.
if(WSARecv(lpPerIoData->Socket,
????????????&(lpPerIoData->DataBuf),
????????????1,
????????????&lpPerIoData->RecvBytes,
????????????&lpPerIoData->Flags,
????????????&(lpPerIoData->Overlapped),
????????????NULL)== SOCKET_ERROR)
????????????{
????????????if(WSAGetLastError() != ERROR_IO_PENDING)//讀操作失敗
????????????{
????????????????closesocket(lpPerIoData->Socket); //關閉socket
????????????????g_session.Erase(lpSession->node); //從Session鏈表中刪除
????????????????g_per_handle_data.MemPoolFree(lpSession); //將Session歸還到池
????????????????g_per_io_data.MemPoolFree(lpPerIoData);
????????????????
????????????}
????????????}
第二,就是當在其他線程里關閉的時候,GetQueuedCompletionStatus就會收到消息.由于我們每次收到自定義recv事件的時候,接著就投遞了一個
recv事件.所以這個事件失敗消息肯定會從完成端口返回.處理如下
bRet = GetQueuedCompletionStatus(CompletionPort,
????????&BytesTransferred,
????????(PULONG_PTR)
????????&lpCompletionKey,
????????(LPOVERLAPPED*)
????????&lpPerIoData,
????????INFINITE);
????if(!bRet)//沒有取到任何東西
????{
????????if(!lpPerIoData)
????????continue;
????}
????//收到線程結束信號
????if(!lpCompletionKey)
????????return 0;

????LPPER_HANDLE_DATA lpSession = (LPPER_HANDLE_DATA)lpCompletionKey;

????if(!bRet)
????{
????????if(BytesTransferred == 0 )//斷開連接
????????{
????????if (lpPerIoData->OperationType == OP_READ)
????????{
????????????printf("client socket %d disconnect\n",lpPerIoData->Socket);
????????????closesocket(lpPerIoData->Socket); //關閉socket
????????????g_session.Erase(lpSession->node); //從Session鏈表中刪除
????????????g_per_handle_data.MemPoolFree(lpSession); //將Session歸還到池
????????????g_per_io_data.MemPoolFree(lpPerIoData);
????????????continue;
????????}

????????}
????}
返回代碼為false并且BytesTransferred為零且lpPerIoData->OperationType == OP_READ.
(2)用戶斷開的處理邏輯與服務器斷開第二種情況是一樣的.有些資料講服務器主動斷開的時候GetQueuedCompletionStatus會返回true是錯誤的

.
6.ERROR_IO_PENDING消息處理這個消息說明現在這個重疊操作還沒有完成,我們必須進行等待,所以必須在錯誤判斷的時候忽略掉這個消息.
????????????if(WSAGetLastError() != ERROR_IO_PENDING)//讀操作失敗
????????????{
????????????????closesocket(lpPerIoData->Socket); //關閉socket
????????????????g_session.Erase(lpSession->node); //從Session鏈表中刪除
????????????????g_per_handle_data.MemPoolFree(lpSession); //將Session歸還到池
????????????????g_per_io_data.MemPoolFree(lpPerIoData);
????????????????
????????????}
7.GetQueuedCompletionStatus返回之后的邏輯判斷按下邊的代碼進行,我們必須知道為什么這樣做,理由就是上邊六條再加上msdn里

GetQueuedCompletionStatus這個函數的說明.
DWORD WINAPI ServerWorkerThread(LPVOID lpParam)
{
????DWORD BytesTransferred;
????LPPER_IO_DATA lpPerIoData = NULL;
????LPVOID lpCompletionKey = NULL;
????BOOL bRet = FALSE;

????while (1)
????{
????bRet = GetQueuedCompletionStatus(CompletionPort,
????????&BytesTransferred,
????????(PULONG_PTR)
????????&lpCompletionKey,
????????(LPOVERLAPPED*)
????????&lpPerIoData,
????????INFINITE);
????if(!bRet)//沒有取到任何東西
????{
????????if(!lpPerIoData)
????????continue;
????}
????//收到線程結束信號
????if(!lpCompletionKey)
????????return 0;

????LPPER_HANDLE_DATA lpSession = (LPPER_HANDLE_DATA)lpCompletionKey;

????if(!bRet)
????{
????????if(BytesTransferred == 0 )//斷開連接
????????{
????????if (lpPerIoData->OperationType == OP_READ)
????????{
????????????printf("client socket %d disconnect\n",lpPerIoData->Socket);
????????????closesocket(lpPerIoData->Socket); //關閉socket
????????????g_session.Erase(lpSession->node); //從Session鏈表中刪除
????????????g_per_handle_data.MemPoolFree(lpSession); //將Session歸還到池
????????????g_per_io_data.MemPoolFree(lpPerIoData);
????????????continue;
????????}

????????}
????}
????switch (lpPerIoData->OperationType)
????{
????case OP_ACCEPT://有連接進來
????????{
????????if (!bRet)
????????{
????????????closesocket(lpPerIoData->Socket);?? //關閉socket
????????????g_per_io_data.MemPoolFree(lpPerIoData);????//歸還結構體到內存池
????????????printf("關閉已經投遞空閑連接\n");
????????????break;
????????}
????????LPPER_HANDLE_DATA lpCurSession;
????????lpCurSession=g_per_handle_data.MemPoolAlloc();
????????printf(" connect socket??= %d\n",lpPerIoData->Socket);
????????lpCurSession->Socket = lpPerIoData->Socket;//就是我們AcceptEx時傳入的socket
????????lpCurSession->node = g_session.Push_Back(lpCurSession);
????????lpPerIoData->lpSession = lpCurSession;
????????if(!CreateIoCompletionPort(
????????????(HANDLE)lpPerIoData->Socket,
????????????CompletionPort,
????????????(ULONG_PTR)lpCurSession,
????????????0))
????????{
????????????closesocket(lpPerIoData->Socket); //關閉socket
????????????g_per_handle_data.MemPoolFree(lpSession);
????????}
????????else
????????{
????????????//g_per_io_data與每次讀寫相關聯
????????????lpPerIoData->OperationType = OP_READ;
????????????ZeroMemory(&(lpPerIoData->Overlapped), sizeof(OVERLAPPED));
????????????lpPerIoData->Flags = 0;
????????????lpPerIoData->DataBuf.len = 1024;
????????????lpPerIoData->DataBuf.buf = lpPerIoData->buffer;
????????????lpPerIoData->RecvBytes =0;
????????????lpPerIoData->lpSession = lpSession;
????????????ZeroMemory(lpPerIoData->buffer,1024);

????????????if(WSARecv(lpPerIoData->Socket,
????????????&(lpPerIoData->DataBuf),
????????????1,
????????????&lpPerIoData->RecvBytes,
????????????&lpPerIoData->Flags,
????????????&(lpPerIoData->Overlapped),
????????????NULL)== SOCKET_ERROR)
????????????{
????????????if(WSAGetLastError() != ERROR_IO_PENDING)//讀操作失敗
????????????{
????????????????closesocket(lpPerIoData->Socket); //關閉socket
????????????????g_session.Erase(lpSession->node); //從Session鏈表中刪除
????????????????g_per_handle_data.MemPoolFree(lpSession); //將Session歸還到池
????????????????g_per_io_data.MemPoolFree(lpPerIoData);
????????????????
????????????}
????????????}

????????}

????????}
????????break;
????case OP_READ:
????????{
????????//cout << lpPerIoData->DataBuf.buf << endl;
????????send(lpPerIoData->Socket,"9876543210",lstrlen("9876543210")+1,0);
????????lpPerIoData->OperationType = OP_READ;
????????ZeroMemory(&(lpPerIoData->Overlapped), sizeof(OVERLAPPED));
????????lpPerIoData->Flags = 0;
????????lpPerIoData->DataBuf.len = 1024;
????????lpPerIoData->DataBuf.buf = lpPerIoData->buffer;
????????lpPerIoData->RecvBytes =0;
????????lpPerIoData->lpSession = lpSession;
????????ZeroMemory(lpPerIoData->buffer,1024);

????????if(WSARecv(lpPerIoData->Socket,
????????????&(lpPerIoData->DataBuf),
????????????1,
????????????&lpPerIoData->RecvBytes,
????????????&lpPerIoData->Flags,
????????????&(lpPerIoData->Overlapped),
????????????NULL)== SOCKET_ERROR)
????????{
????????????if(WSAGetLastError() != ERROR_IO_PENDING)
????????????{
????????????closesocket(lpPerIoData->Socket); //關閉socket
????????????g_session.Erase(lpSession->node); //從Session鏈表中刪除
????????????g_per_handle_data.MemPoolFree(lpSession); //將Session歸還到池
????????????g_per_io_data.MemPoolFree(lpPerIoData);
????????????}
????????}
????????}
????????break;
????case OP_WRITE:
????????break;
????default:
????????break;
????}
????}
}
下邊就是整個控制臺代碼.
/********************************以下代碼要與前邊的容器頭文件一起進行編譯***********************************************/
//iocpsever.cpp
#include <winsock2.h>
#include "memorypool.h"
#include <Mswsock.h>
#include <process.h>

using namespace std;

#pragma comment(lib, "ws2_32.lib")
#pragma comment(lib, "Mswsock.lib")

using namespace std;


// 單句柄數據，每個連接（客戶端）對應一個這樣的結構。
//只有連接斷開，或者服務器關閉的時候才釋放
struct tagPER_HANDLE_DATA;

typedef unsigned (WINAPI *PBEGINTHREADX_THREADFUN)(LPVOID lpThreadParameter); //線程函數原型

typedef struct tagPER_HANDLE_DATA *LPPER_HANDLE_DATA;
typedef CList<LPPER_HANDLE_DATA>::iterator LISTSESSIONNODE;

typedef struct tagPER_HANDLE_DATA
{
????SOCKET Socket;
????LISTSESSIONNODE node;
????// 將和這個句柄關聯的其他有用信息，盡管放在這里面吧
}PER_HANDLE_DATA;

// 單I/O操作數據。每次收發數據的時候
//收發數據操作完成數之后釋放。
//需要注意的是OVERLAPPED Overlapped一定要放在最前

typedef struct tagPER_IO_DATA
{
????OVERLAPPED Overlapped;
????WSABUF DataBuf;
????char buffer[1024];
????DWORD RecvBytes;
????DWORD Flags;
????int OperationType;
????SOCKET Socket;
????LPPER_HANDLE_DATA lpSession;
}PER_IO_DATA,*LPPER_IO_DATA;
//操作標志
#define OP_READ 0
#define OP_WRITE 1
#define OP_ACCEPT 2

//連接數據內存池
CMemoryPool<PER_IO_DATA,1024> g_per_io_data;
//數據收發內存池
CMemoryPool<PER_HANDLE_DATA,1024> g_per_handle_data;
//完成線程

//保存當前連接數據指針
CList<LPPER_HANDLE_DATA> g_session;


SOCKET Listen;
HANDLE CompletionPort;
HANDLE *m_arrayIOCPThreadHandle;
DWORD threadcnt;
DWORD WINAPI ServerWorkerThread(LPVOID lpParam);


BOOL b;
HANDLE g_hAcceptExOverEvent;
HANDLE g_hAcceptExThread;
DWORD WINAPI WinSockAcceptEXThread(LPVOID lpParam);
bool StartUpWinSockAcceptEXThread();
void CloseWinSockAcceptEXThread();
void InitCompletionPort();
void CloseCompletionPort();
int main(void)
{
????InitCompletionPort();
????StartUpWinSockAcceptEXThread();

????{
????HANDLE hWaitEvent;
????hWaitEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
????WaitForSingleObject(hWaitEvent, 30000/*INFINITE*/);
????CloseHandle(hWaitEvent);

????}
????CloseCompletionPort();
????return 0;
}
DWORD WINAPI ServerWorkerThread(LPVOID lpParam)
{
????DWORD BytesTransferred;
????LPPER_IO_DATA lpPerIoData = NULL;
????LPVOID lpCompletionKey = NULL;
????BOOL bRet = FALSE;

????while (1)
????{
????bRet = GetQueuedCompletionStatus(CompletionPort,
????????&BytesTransferred,
????????(PULONG_PTR)
????????&lpCompletionKey,
????????(LPOVERLAPPED*)
????????&lpPerIoData,
????????INFINITE);
????if(!bRet)//沒有取到任何東西
????{
????????if(!lpPerIoData)
????????continue;
????}
????//收到線程結束信號
????if(!lpCompletionKey)
????????return 0;

????LPPER_HANDLE_DATA lpSession = (LPPER_HANDLE_DATA)lpCompletionKey;

????if(!bRet)
????{
????????if(BytesTransferred == 0 )//斷開連接
????????{
????????if (lpPerIoData->OperationType == OP_READ)
????????{
????????????printf("client socket %d disconnect\n",lpPerIoData->Socket);
????????????closesocket(lpPerIoData->Socket); //關閉socket
????????????g_session.Erase(lpSession->node); //從Session鏈表中刪除
????????????g_per_handle_data.MemPoolFree(lpSession); //將Session歸還到池
????????????g_per_io_data.MemPoolFree(lpPerIoData);
????????????continue;
????????}

????????}
????}
????switch (lpPerIoData->OperationType)
????{
????case OP_ACCEPT://有連接進來
????????{
????????if (!bRet)
????????{
????????????closesocket(lpPerIoData->Socket);?? //關閉socket
????????????g_per_io_data.MemPoolFree(lpPerIoData);????//歸還結構體到內存池
????????????printf("關閉已經投遞空閑連接\n");
????????????break;
????????}
????????LPPER_HANDLE_DATA lpCurSession;
????????lpCurSession=g_per_handle_data.MemPoolAlloc();
????????printf(" connect socket??= %d\n",lpPerIoData->Socket);
????????lpCurSession->Socket = lpPerIoData->Socket;//就是我們AcceptEx時傳入的socket
????????lpCurSession->node = g_session.Push_Back(lpCurSession);
????????lpPerIoData->lpSession = lpCurSession;
????????if(!CreateIoCompletionPort(
????????????(HANDLE)lpPerIoData->Socket,
????????????CompletionPort,
????????????(ULONG_PTR)lpCurSession,
????????????0))
????????{
????????????closesocket(lpPerIoData->Socket); //關閉socket
????????????g_per_handle_data.MemPoolFree(lpSession);
????????}
????????else
????????{
????????????//g_per_io_data與每次讀寫相關聯
????????????lpPerIoData->OperationType = OP_READ;
????????????ZeroMemory(&(lpPerIoData->Overlapped), sizeof(OVERLAPPED));
????????????lpPerIoData->Flags = 0;
????????????lpPerIoData->DataBuf.len = 1024;
????????????lpPerIoData->DataBuf.buf = lpPerIoData->buffer;
????????????lpPerIoData->RecvBytes =0;
????????????lpPerIoData->lpSession = lpSession;
????????????ZeroMemory(lpPerIoData->buffer,1024);

????????????if(WSARecv(lpPerIoData->Socket,
????????????&(lpPerIoData->DataBuf),
????????????1,
????????????&lpPerIoData->RecvBytes,
????????????&lpPerIoData->Flags,
????????????&(lpPerIoData->Overlapped),
????????????NULL)== SOCKET_ERROR)
????????????{
????????????if(WSAGetLastError() != ERROR_IO_PENDING)//讀操作失敗
????????????{
????????????????closesocket(lpPerIoData->Socket); //關閉socket
????????????????g_session.Erase(lpSession->node); //從Session鏈表中刪除
????????????????g_per_handle_data.MemPoolFree(lpSession); //將Session歸還到池
????????????????g_per_io_data.MemPoolFree(lpPerIoData);
????????????????
????????????}
????????????}

????????}

????????}
????????break;
????case OP_READ:
????????{
????????//cout << lpPerIoData->DataBuf.buf << endl;
????????send(lpPerIoData->Socket,"9876543210",lstrlen("9876543210")+1,0);
????????lpPerIoData->OperationType = OP_READ;
????????ZeroMemory(&(lpPerIoData->Overlapped), sizeof(OVERLAPPED));
????????lpPerIoData->Flags = 0;
????????lpPerIoData->DataBuf.len = 1024;
????????lpPerIoData->DataBuf.buf = lpPerIoData->buffer;
????????lpPerIoData->RecvBytes =0;
????????lpPerIoData->lpSession = lpSession;
????????ZeroMemory(lpPerIoData->buffer,1024);

????????if(WSARecv(lpPerIoData->Socket,
????????????&(lpPerIoData->DataBuf),
????????????1,
????????????&lpPerIoData->RecvBytes,
????????????&lpPerIoData->Flags,
????????????&(lpPerIoData->Overlapped),
????????????NULL)== SOCKET_ERROR)
????????{
????????????if(WSAGetLastError() != ERROR_IO_PENDING)
????????????{
????????????closesocket(lpPerIoData->Socket); //關閉socket
????????????g_session.Erase(lpSession->node); //從Session鏈表中刪除
????????????g_per_handle_data.MemPoolFree(lpSession); //將Session歸還到池
????????????g_per_io_data.MemPoolFree(lpPerIoData);
????????????}
????????}
????????}
????????break;
????case OP_WRITE:
????????break;
????default:
????????break;
????}
????}
}
DWORD WINAPI WinSockAcceptEXThread(LPVOID lpParam)
{
????LINGER lingerStruct = { 0x01, 0x00 };
????BOOL bNodelay = TRUE;
????//創建事件
????while (b)
????{
????//每次投遞10次，進入等待狀態，當AcceptEx全部完成之后，繼續投遞
????if(WaitForSingleObject(g_hAcceptExOverEvent, INFINITE) == WAIT_FAILED)
????????continue;
????for(int i =0;i<10 && b;i++)
????{
????????int zero =0;
????????PER_IO_DATA * pper_io_data = NULL;
????????DWORD dwAddrLen = sizeof(sockaddr_in)+16;
????????pper_io_data = g_per_io_data.MemPoolAlloc();
????????pper_io_data ->Socket = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED);
????????if(pper_io_data->Socket == INVALID_SOCKET)
????????{
????????g_per_io_data.MemPoolFree(pper_io_data);
????????continue;
????????}
????????pper_io_data->OperationType = OP_ACCEPT;
????????pper_io_data->lpSession=NULL;
????????ZeroMemory(&pper_io_data->Overlapped,sizeof(OVERLAPPED));//一定要注意清零
????????pper_io_data->RecvBytes =0;
????????pper_io_data->Flags =0;
????????ZeroMemory(pper_io_data->buffer,1024);
????????setsockopt(pper_io_data->Socket, IPPROTO_TCP, TCP_NODELAY, (char*)&bNodelay, sizeof(BOOL));
????????setsockopt(pper_io_data->Socket, SOL_SOCKET, SO_LINGER, (char*)&lingerStruct, sizeof(LINGER));

????????if(!AcceptEx(Listen,pper_io_data ->Socket,pper_io_data ->buffer,0,dwAddrLen,dwAddrLen,&pper_io_data -

>RecvBytes,&pper_io_data->Overlapped))
????????{
????????if(WSAGetLastError() != ERROR_IO_PENDING)//對于AcceptEx，WSARecv,WSASend一定要有這樣的判斷，因為是異步的所以

不會立即完成
????????{
????????????closesocket(pper_io_data->Socket);
????????????g_per_io_data.MemPoolFree(pper_io_data); //歸還結構體到內存池
????????????continue;
????????}

????????}
????}
????}
????return 0;
}
bool StartUpWinSockAcceptEXThread()
{
????g_hAcceptExOverEvent = CreateEvent(NULL, FALSE, FALSE, NULL);
????if(!g_hAcceptExOverEvent)
????{
????return false;
????}
????//幫定事件，由于我們用AcceptEx是一個異步的投遞，這樣幫定之后，如果投遞的AcceptEx事件全部完成
????//則g_hAcceptExOverEvent事件得到通知，進而同步AcceptEx調用
????if(WSAEventSelect(Listen, g_hAcceptExOverEvent, FD_ACCEPT) == SOCKET_ERROR)
????{
????return false;
????}
????//由于開始是復位，變成置位
????SetEvent(g_hAcceptExOverEvent);
????b = TRUE;
????g_hAcceptExThread = (HANDLE)_beginthreadex(NULL, 0, (PBEGINTHREADX_THREADFUN)WinSockAcceptEXThread, NULL, 0, NULL);
????if(!g_hAcceptExThread)
????return false;
????return true;

}
void CloseWinSockAcceptEXThread()
{
????b=false;
????SetEvent(g_hAcceptExOverEvent);
????if(WaitForSingleObject(g_hAcceptExThread,10000)!= WAIT_OBJECT_0)
????TerminateThread(g_hAcceptExThread, 0);
????CloseHandle(g_hAcceptExThread);
????CloseHandle(g_hAcceptExOverEvent);
};
void CloseCompletionPort()
{
????
????closesocket(Listen);
????g_session._guard.Lock();
????for(CList<LPPER_HANDLE_DATA>::iterator lpNode = g_session.begin(); lpNode != g_session.end();lpNode++)
????{
????closesocket((*lpNode)->Socket);
????printf(" close socket??= %d\n",(*lpNode)->Socket);
????}
????g_session._guard.Unlock();
????Sleep(1000);
????//向IOCP發送結束線程信號

????for(DWORD i = 0; i < threadcnt; i++)
????PostQueuedCompletionStatus(CompletionPort, 0, NULL, NULL);

????//等待工作線程結束，等待時間10秒

????if(WaitForMultipleObjects(threadcnt, m_arrayIOCPThreadHandle, TRUE, 10000) != WAIT_OBJECT_0)
????{
????//如果10秒內沒有結束所有線程，就強制結束
????for(DWORD i = 0; i < threadcnt; i++)
????????TerminateThread(m_arrayIOCPThreadHandle[i], 0);
????}
????//關閉所有工作線程句柄
????for(DWORD i = 0; i < threadcnt; i++)
????CloseHandle(m_arrayIOCPThreadHandle[i]);
????//釋放線程句柄數組
????delete [] m_arrayIOCPThreadHandle;
????m_arrayIOCPThreadHandle = NULL;
????CloseHandle(CompletionPort);
????g_per_io_data.DestroyMemPool();
????g_per_handle_data.DestroyMemPool();
????g_session.Clear();
}
void InitCompletionPort()
{
????WSADATA wsd;
????SYSTEM_INFO SystemInfo;
????SOCKADDR_IN InternetAddr;
????WSAStartup(MAKEWORD(2, 2), &wsd);

????//創建完成端口
????CompletionPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE,
????NULL,
????0,
????threadcnt);

????//得到處理器數量
????GetSystemInfo(&SystemInfo);
????//經驗公式：一般按公式創建工作線程

????threadcnt = 2*SystemInfo.dwNumberOfProcessors+2;
????m_arrayIOCPThreadHandle = new HANDLE[threadcnt];
????for(DWORD i = 0; i < threadcnt; i++)
????{
????m_arrayIOCPThreadHandle[i] = (HANDLE)_beginthreadex(NULL, 0, (PBEGINTHREADX_THREADFUN)ServerWorkerThread, NULL, 0,

NULL);
????if(!m_arrayIOCPThreadHandle[i])
????????return ;
????}
????//創建監聽socket

????Listen = WSASocket(AF_INET,
????SOCK_STREAM,
????0,
????NULL,
????0,
????WSA_FLAG_OVERLAPPED);

????InternetAddr.sin_family = PF_INET;
????InternetAddr.sin_port = htons(20000);
????InternetAddr.sin_addr.s_addr = htonl(INADDR_ANY);
????//幫定到指定端口

????bind(Listen, (SOCKADDR*)&InternetAddr, sizeof(InternetAddr));

????//開始監聽
????listen(Listen, SOMAXCONN);
????//完成端口幫定到監聽socket
????printf(" listen socket??= %d\n",Listen);
????if (CreateIoCompletionPort((HANDLE) Listen, CompletionPort, (ULONG_PTR)&Listen, threadcnt) == NULL)
????{
????printf("CreateIoCompletionPort failed with error %d\n", GetLastError());
????return ;
????}
}




FROM:http://www.vchelp.net/cndevforum/subject_view.asp?subject_id=176818&forum_id=55
作者:lustskyboy溝通無限