轉(zhuǎn)載于hxzb7215191 的BLOG
當(dāng)然TCP方式的模型還有事件選擇模型。
就是把所有的網(wǎng)絡(luò)事件和我們的一個(gè)程序里定義的事件梆定。
這個(gè)有它的好處，可能可以讓我們更好的寫一個(gè)線程來管理
接收與發(fā)送。
現(xiàn)在來講一下一個(gè)完成端口模型。
? 完成端口
?
?一個(gè)完成端口其實(shí)就是一個(gè)通知隊(duì)列，由操作系統(tǒng)把已經(jīng)完成的重疊I/O請求的通知
?放入其中。當(dāng)某項(xiàng)I/O操作一旦完成，某個(gè)可以對該操作結(jié)果進(jìn)行處理的工作者線程
?就會(huì)收到一則通知。而套接字在被創(chuàng)建后，可以在任何時(shí)候與某個(gè)完成端口進(jìn)行關(guān)
?聯(lián)。
?
?步驟:
?1、創(chuàng)建一個(gè)空的完成端口；
?2、得到本地機(jī)器的CPU個(gè)數(shù)；
?3、開啟CPU*2個(gè)工作線程(又名線程池),全部都在等待完成端口的完成包;
?4、創(chuàng)建TCP的監(jiān)聽socket，使用事件邦定，創(chuàng)建監(jiān)聽線程;
?5、當(dāng)有人連接進(jìn)入的時(shí)候，將Client socket保存到一個(gè)我們自己定義的關(guān)鍵鍵，
??? 并把它與我們創(chuàng)建的完成端口關(guān)聯(lián)；
?6、使用WSARecv和WSASend函數(shù)投遞一些請求，這是使用重疊I/O的方式;
?7、重復(fù)5~6;
?注：1、重疊I/O的方式中，接收與發(fā)送數(shù)據(jù)包的時(shí)候，一定要進(jìn)行投遞請求這是
???它們這個(gè)體系結(jié)構(gòu)的特點(diǎn)
???當(dāng)然，在完成端口方式中，不是直接使用的WSARecv和WSASend函數(shù)進(jìn)行請求
???的投遞的。而是使用的ReadFile,Write的方式
??2、完成端口使用了系統(tǒng)內(nèi)部的一些模型，所以我們只要按照一定的順序調(diào)用就
???可以完成了。
??3、完成端口是使用在這樣的情況下，有成千上萬的用戶連接的時(shí)候，它能夠
???保證性能不會(huì)降低。
?
#include?
<
winsock2.h
>
#include?
<
windows.h
>
#include?
<
stdio.h
>
#define
?PORT?5150
#define
?DATA_BUFSIZE?8192
//
關(guān)鍵項(xiàng)
typedef?
struct
{
???OVERLAPPED?Overlapped;
???WSABUF?DataBuf;
???CHAR?Buffer[DATA_BUFSIZE];
???DWORD?BytesSEND;
???DWORD?BytesRECV;
}
?PER_IO_OPERATION_DATA,?
*
?LPPER_IO_OPERATION_DATA;
typedef?
struct
?
{
???SOCKET?Socket;
}
?PER_HANDLE_DATA,?
*
?LPPER_HANDLE_DATA;
DWORD?WINAPI?ServerWorkerThread(LPVOID?CompletionPortID);
void
?main(
void
)
{
???SOCKADDR_IN?InternetAddr;
???SOCKET?Listen;
???SOCKET?Accept;
???HANDLE?CompletionPort;
???SYSTEM_INFO?SystemInfo;
???LPPER_HANDLE_DATA?PerHandleData;
???LPPER_IO_OPERATION_DATA?PerIoData;
???
int
?i;
???DWORD?RecvBytes;
???DWORD?Flags;
???DWORD?ThreadID;
???WSADATA?wsaData;
???DWORD?Ret;
???
if
?((Ret?
=
?WSAStartup(
0x0202
,?
&
wsaData))?
!=
?
0
)
???
{
??????printf(
"
WSAStartup?failed?with?error?%d\n
"
,?Ret);
??????
return
;
???}
???
//
打開一個(gè)空的完成端口
???
if
?((CompletionPort?
=
?CreateIoCompletionPort(INVALID_HANDLE_VALUE,?NULL,?
0
,?
0
))?
==
?NULL)
???
{
??????printf(?
"
CreateIoCompletionPort?failed?with?error:?%d\n
"
,?GetLastError());
??????
return
;
???}
???
//
?Determine?how?many?processors?are?on?the?system.
???GetSystemInfo(
&
SystemInfo);
???
//
?開啟cpu個(gè)數(shù)的2倍個(gè)的線程
???
for
(i?
=
?
0
;?i?
<
?SystemInfo.dwNumberOfProcessors?
*
?
2
;?i
++
)
???
{
??????HANDLE?ThreadHandle;
??????
//
?Create?a?server?worker?thread?and?pass?the?completion?port?to?the?thread.
??????
if
?((ThreadHandle?
=
?CreateThread(NULL,?
0
,?ServerWorkerThread,?CompletionPort,
?????????
0
,?
&
ThreadID))?
==
?NULL)
??????
{
?????????printf(
"
CreateThread()?failed?with?error?%d\n
"
,?GetLastError());
?????????
return
;
??????}
??????
//
?Close?the?thread?handle
??????CloseHandle(ThreadHandle);
???}
???
//
打開一個(gè)服務(wù)器socket
???
if
?((Listen?
=
?WSASocket(AF_INET,?SOCK_STREAM,?
0
,?NULL,?
0
,
??????WSA_FLAG_OVERLAPPED))?
==
?INVALID_SOCKET)
???
{
??????printf(
"
WSASocket()?failed?with?error?%d\n
"
,?WSAGetLastError());
??????
return
;
???}
?
???InternetAddr.sin_family?
=
?AF_INET;
???InternetAddr.sin_addr.s_addr?
=
?htonl(INADDR_ANY);
???InternetAddr.sin_port?
=
?htons(PORT);
???
if
?(bind(Listen,?(PSOCKADDR)?
&
InternetAddr,?
sizeof
(InternetAddr))?
==
?SOCKET_ERROR)
???
{
??????printf(
"
bind()?failed?with?error?%d\n
"
,?WSAGetLastError());
??????
return
;
???}
???
if
?(listen(Listen,?
5
)?
==
?SOCKET_ERROR)
???
{
??????printf(
"
listen()?failed?with?error?%d\n
"
,?WSAGetLastError());
??????
return
;
???}
???
//
開始接收從客戶端來的連接
???
while
(TRUE)
???
{
??????
if
?((Accept?
=
?WSAAccept(Listen,?NULL,?NULL,?NULL,?
0
))?
==
?SOCKET_ERROR)
??????
{
?????????printf(
"
WSAAccept()?failed?with?error?%d\n
"
,?WSAGetLastError());
?????????
return
;
??????}
??????
//
?創(chuàng)建一個(gè)關(guān)鍵項(xiàng)用于保存這個(gè)客戶端的信息，用戶接收發(fā)送的重疊結(jié)構(gòu)，
??????
//
?還有使用到的緩沖區(qū)
??????
if
?((PerHandleData?
=
?(LPPER_HANDLE_DATA)?GlobalAlloc(GPTR,?
?????????
sizeof
(PER_HANDLE_DATA)))?
==
?NULL)
??????
{
?????????printf(
"
GlobalAlloc()?failed?with?error?%d\n
"
,?GetLastError());
?????????
return
;
??????}
??????
//
?Associate?the?accepted?socket?with?the?original?completion?port.
??????printf(
"
Socket?number?%d?connected\n
"
,?Accept);
??????PerHandleData
->
Socket?
=
?Accept;
??????
//
與我們的創(chuàng)建的那個(gè)完成端口關(guān)聯(lián)起來,將關(guān)鍵項(xiàng)也與指定的一個(gè)完成端口關(guān)聯(lián)
??????
if
?(CreateIoCompletionPort((HANDLE)?Accept,?CompletionPort,?(DWORD)?PerHandleData,
?????????
0
)?
==
?NULL)
??????
{
?????????printf(
"
CreateIoCompletionPort?failed?with?error?%d\n
"
,?GetLastError());
?????????
return
;
??????}
??????
//
?投遞一次接收，由于接收都需要使用這個(gè)函數(shù)來投遞一個(gè)接收的準(zhǔn)備
??????
if
?((PerIoData?
=
?(LPPER_IO_OPERATION_DATA)?GlobalAlloc(GPTR,??????????
sizeof
(PER_IO_OPERATION_DATA)))?
==
?NULL)
??????
{
?????????printf(
"
GlobalAlloc()?failed?with?error?%d\n
"
,?GetLastError());
?????????
return
;
??????}
??????ZeroMemory(
&
(PerIoData
->
Overlapped),?
sizeof
(OVERLAPPED));
??????PerIoData
->
BytesSEND?
=
?
0
;
??????PerIoData
->
BytesRECV?
=
?
0
;
??????PerIoData
->
DataBuf.len?
=
?DATA_BUFSIZE;
??????PerIoData
->
DataBuf.buf?
=
?PerIoData
->
Buffer;
??????Flags?
=
?
0
;
??????
if
?(WSARecv(Accept,?
&
(PerIoData
->
DataBuf),?
1
,?
&
RecvBytes,?
&
Flags,
?????????
&
(PerIoData
->
Overlapped),?NULL)?
==
?SOCKET_ERROR)
??????
{
?????????
if
?(WSAGetLastError()?
!=
?ERROR_IO_PENDING)
?????????
{
????????????printf(
"
WSARecv()?failed?with?error?%d\n
"
,?WSAGetLastError());
????????????
return
;
?????????}
??????}
???}
}
//
工作線程
DWORD?WINAPI?ServerWorkerThread(LPVOID?CompletionPortID)
{
???HANDLE?CompletionPort?
=
?(HANDLE)?CompletionPortID;
???DWORD?BytesTransferred;
???LPOVERLAPPED?Overlapped;
???LPPER_HANDLE_DATA?PerHandleData;
???LPPER_IO_OPERATION_DATA?PerIoData;
???DWORD?SendBytes,?RecvBytes;
???DWORD?Flags;
???
???
while
(TRUE)
???
{
??????
//
完成端口有消息來了
??????
if
?(GetQueuedCompletionStatus(CompletionPort,?
&
BytesTransferred,
?????????(LPDWORD)
&
PerHandleData,?(LPOVERLAPPED?
*
)?
&
PerIoData,?INFINITE)?
==
?
0
)
??????
{
?????????printf(
"
GetQueuedCompletionStatus?failed?with?error?%d\n
"
,?GetLastError());
?????????
return
?
0
;
??????}
??????
//
是不是有人退出了
??????
if
?(BytesTransferred?
==
?
0
)
??????
{
?????????printf(
"
Closing?socket?%d\n
"
,?PerHandleData
->
Socket);
?????????
if
?(closesocket(PerHandleData
->
Socket)?
==
?SOCKET_ERROR)
?????????
{
????????????printf(
"
closesocket()?failed?with?error?%d\n
"
,?WSAGetLastError());
????????????
return
?
0
;
?????????}
?????????GlobalFree(PerHandleData);
?????????GlobalFree(PerIoData);
?????????
continue
;
??????}
??????
//
??????
if
?(PerIoData
->
BytesRECV?
==
?
0
)
??????
{
?????????PerIoData
->
BytesRECV?
=
?BytesTransferred;
?????????PerIoData
->
BytesSEND?
=
?
0
;
??????}
??????
else
??????
{
?????????PerIoData
->
BytesSEND?
+=
?BytesTransferred;
??????}
??????
if
?(PerIoData
->
BytesRECV?
>
?PerIoData
->
BytesSEND)
??????
{
?????????
//
?Post?another?WSASend()?request.
?????????
//
?Since?WSASend()?is?not?gauranteed?to?send?all?of?the?bytes?requested,
?????????
//
?continue?posting?WSASend()?calls?until?all?received?bytes?are?sent.
?????????ZeroMemory(
&
(PerIoData
->
Overlapped),?
sizeof
(OVERLAPPED));
?????????PerIoData
->
DataBuf.buf?
=
?PerIoData
->
Buffer?
+
?PerIoData
->
BytesSEND;
?????????PerIoData
->
DataBuf.len?
=
?PerIoData
->
BytesRECV?
-
?PerIoData
->
BytesSEND;
?????????
if
?(WSASend(PerHandleData
->
Socket,?
&
(PerIoData
->
DataBuf),?
1
,?
&
SendBytes,?
0
,
????????????
&
(PerIoData
->
Overlapped),?NULL)?
==
?SOCKET_ERROR)
?????????
{
????????????
if
?(WSAGetLastError()?
!=
?ERROR_IO_PENDING)
????????????
{
???????????????printf(
"
WSASend()?failed?with?error?%d\n
"
,?WSAGetLastError());
???????????????
return
?
0
;
????????????}
?????????}
??????}
??????
else
??????
{
?????????PerIoData
->
BytesRECV?
=
?
0
;
?????????
//
?Now?that?there?are?no?more?bytes?to?send?post?another?WSARecv()?request.
?????????Flags?
=
?
0
;
?????????ZeroMemory(
&
(PerIoData
->
Overlapped),?
sizeof
(OVERLAPPED));
?????????PerIoData
->
DataBuf.len?
=
?DATA_BUFSIZE;
?????????PerIoData
->
DataBuf.buf?
=
?PerIoData
->
Buffer;
?????????
if
?(WSARecv(PerHandleData
->
Socket,?
&
(PerIoData
->
DataBuf),?
1
,?
&
RecvBytes,?
&
Flags,
????????????
&
(PerIoData
->
Overlapped),?NULL)?
==
?SOCKET_ERROR)
?????????
{
????????????
if
?(WSAGetLastError()?
!=
?ERROR_IO_PENDING)
????????????
{
???????????????printf(
"
WSARecv()?failed?with?error?%d\n
"
,?WSAGetLastError());
???????????????
return
?
0
;
????????????}
?????????}
??????}
???}
}