/////////////////////////////////////////////////////////////////////////////////////
// Iocp 頭文件
#pragma once
#include <winsock2.h>
#pragma comment( lib, "ws2_32.lib" )
const int OP_READ = 0;
const int OP_WRITE = 1;
const int OP_ACCEPT = 2;
/*
OVERLAPPEDPLUS 結(jié)構(gòu)體設(shè)計(jì)思路
OVERLAPPED 是一個(gè)固定的用于處理網(wǎng)絡(luò)消息事件返回值的結(jié)構(gòu)體變量
在完成端口和重疊I/O模型里用于返回消息事件的結(jié)果
因?yàn)樵谔幚砭W(wǎng)絡(luò)消息的時(shí)候，發(fā)送的是一個(gè)返回值的結(jié)構(gòu)體指針，只要結(jié)構(gòu)體
的前面部分滿足系統(tǒng)的要求，在系統(tǒng)操作成功的時(shí)候也就會(huì)把這個(gè)結(jié)構(gòu)體指針
發(fā)回給用戶，我們只要在系統(tǒng)定義的結(jié)構(gòu)體后面擴(kuò)展一些自己的東西，就可以
很輕松的確定該消息是誰(shuí)發(fā)過(guò)來(lái)的。
不過(guò)好像完成端口在設(shè)計(jì)的時(shí)候也滿足了這樣的需求，所以在這里我只是放入
一些與系統(tǒng)連接有關(guān)的數(shù)據(jù)，用戶需要存放的數(shù)據(jù)這里就不在存放
這里存儲(chǔ)與系統(tǒng)相關(guān)的數(shù)據(jù)有：
socket
OpCode 本次消息的操作類(lèi)型（在完成端口的操作里面，是以消息通知系統(tǒng)，
讀數(shù)據(jù)/寫(xiě)數(shù)據(jù)，都是要發(fā)這樣的消息結(jié)構(gòu)體過(guò)去的，所以如果系統(tǒng)要同時(shí)
進(jìn)行讀寫(xiě)操作的話，就需要有一個(gè)變量來(lái)區(qū)分操作了）
WSABUF wbuf; // 讀寫(xiě)緩沖區(qū)結(jié)構(gòu)體變量
DWORD dwBytes, dwFlags; // 一些在讀寫(xiě)時(shí)用到的標(biāo)志性變量
char buf[4096]; // 自己的緩沖區(qū)
上面的4個(gè)變量存放的是一些與消息相關(guān)的數(shù)據(jù)，都是一些操作上用到的，
這些東西都是固定的，具體作用需要參考一下完成端口相關(guān)函數(shù)的參數(shù)接口
*/
struct OVERLAPPEDPLUS
{
OVERLAPPED ol;
SOCKET s;
int OpCode;
WSABUF wbuf;
DWORD dwBytes, dwFlags;
char buf[4096];
};
class CIOCP
{
protected:
HANDLE g_hwThread; // 工作線程句柄
DWORD m_wthreadID;
HANDLE g_haThread; // 連接線程句柄
DWORD m_athreadID;
public:
bool m_workThread;
bool m_acceptThread;
HANDLE m_hIocp; // 完成端口的句柄
SOCKET m_sSocket;
public:
CIOCP(void);
~CIOCP(void);
virtual void OnRead(void * p, char *buf, int len){};
virtual void OnAccept(SOCKET socket);
virtual void OnClose(void * p){};
bool SetIoCompletionPort(SOCKET socket, void *p, char *buf = NULL, int len = 0);
// 把一個(gè)socket與一個(gè)自定義的結(jié)構(gòu)體關(guān)聯(lián)到完成端口（相當(dāng)于把socket與一個(gè)結(jié)構(gòu)體變量進(jìn)行綁定），
// 這樣當(dāng)發(fā)送上面3種網(wǎng)絡(luò)事件的時(shí)候，該結(jié)構(gòu)體變量會(huì)再傳回給程序
// 這樣就可以區(qū)分當(dāng)前網(wǎng)絡(luò)事件是那個(gè)socket發(fā)出的
bool Init(void);
bool Listen(int port);
static DWORD __stdcall WorkThread(LPVOID Param);
static DWORD __stdcall AcceptThread(LPVOID Param);
};
class CIOCPClient: public CIOCP
{
protected:
SOCKET m_socket;
public:
bool Connect(char *ip, int port);
void Send(char *buf, int len);
};
//////////////////////////////////////////////////////////////////////////////////////////
// Iocp 實(shí)現(xiàn)文件
#include "StdAfx.h"
#include "iocp.h"
static bool bInit = false;
DWORD __stdcall CIOCP::WorkThread(LPVOID Param)
{
CIOCP * pthis = (CIOCP *)Param;
void * re;
OVERLAPPED * pOverlap;
DWORD berByte;
while(pthis->m_workThread)
{
int ret;
ret = GetQueuedCompletionStatus(pthis->m_hIocp, &berByte, (LPDWORD)&re, (LPOVERLAPPED *)&pOverlap, INFINITE);
if (ret == ERROR_SUCCESS)
{
}
if (berByte == 0)
{
// 客戶端斷開(kāi)連接
pthis->OnClose(re);
OVERLAPPEDPLUS *olp = (OVERLAPPEDPLUS *)pOverlap;
closesocket(olp->s);
delete olp; // 釋放 與socket綁定的結(jié)構(gòu)體變量
continue;
}
if (re == NULL) return 0;
OVERLAPPEDPLUS *olp = (OVERLAPPEDPLUS *)pOverlap;
switch(olp->OpCode)
{
case OP_READ:
pthis->OnRead(re, olp->wbuf.buf, berByte); // 調(diào)用 OnRead() 通知應(yīng)用程序，服務(wù)器收到來(lái)自客戶端的網(wǎng)絡(luò)數(shù)據(jù)
WSARecv(olp->s, &olp->wbuf, 1, &olp->dwBytes, &olp->dwFlags, &olp->ol, NULL); // 繼續(xù)調(diào)用一個(gè)接收的 I/O 異步請(qǐng)求
break;
default:
break;
}
}
return 0;
}
DWORD __stdcall CIOCP::AcceptThread(LPVOID Param)
{
CIOCP * pthis = (CIOCP *)Param;
while(pthis->m_acceptThread)
{
SOCKET client;
if ((client= accept(pthis->m_sSocket, NULL, NULL)) == INVALID_SOCKET)
{
// 錯(cuò)誤處理
}
pthis->OnAccept(client); // 調(diào)用 OnAccept()通知應(yīng)用程序有新客戶端連接
}
return 1;
}
CIOCP::CIOCP(void)
{
}
CIOCP::~CIOCP(void)
{
}
bool CIOCP::Init(void)
{
if (bInit)
return true;
WSADATA wsd;
if (WSAStartup(MAKEWORD(2,2), &wsd) != 0)
return false;
bInit = true;
return true;
}
bool CIOCP::Listen(int port)
{
if (!bInit)
if (!Init())
return false;
m_sSocket = socket(AF_INET, SOCK_STREAM, 0);
if (m_sSocket == INVALID_SOCKET)
return false;
//SOCKADDR_IN addr;
sockaddr_in addr;
addr.sin_family = AF_INET;
addr.sin_port = htons(port);
addr.sin_addr.S_un.S_addr = htonl(INADDR_ANY);
//addr.sin_addr.S_un.S_addr = inet_addr(ip);
if (bind(m_sSocket, (struct sockaddr *)&addr, sizeof(addr)) == SOCKET_ERROR)
return false;
if (listen(m_sSocket, 10) == SOCKET_ERROR)
return false;
if ((m_hIocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, NULL, 0)) == NULL) // 創(chuàng)建完成端口的句柄
return false;
this->m_acceptThread = true;
g_haThread = CreateThread(NULL, 0, AcceptThread, (LPVOID)this, 0, &m_athreadID); // 創(chuàng)建連接線程，用來(lái)接收客戶端的連接
this->m_workThread = true;
g_hwThread = CreateThread(NULL, 0, WorkThread, (LPVOID)this, 0, &m_wthreadID); // 創(chuàng)建工作線程，用來(lái)處理完成端口消息的
return true;
}
bool CIOCP::SetIoCompletionPort(SOCKET socket, void *p, char *buf, int len)
{
if (CreateIoCompletionPort((HANDLE)socket, m_hIocp, (ULONG_PTR)p, 0) == NULL)
return false;
OVERLAPPEDPLUS *olp = new OVERLAPPEDPLUS;
memset(olp, 0, sizeof(OVERLAPPEDPLUS));
olp->s = socket;
if (buf)
{
// 這里可以使用用戶自定義的緩沖區(qū)地址，如果用戶不想設(shè)置，也可以采用默認(rèn)分配的緩沖區(qū)
olp->wbuf.buf = buf;
olp->wbuf.len = len;
}
else
{
olp->wbuf.buf = olp->buf;
olp->wbuf.len = 4096;
}
olp->OpCode = OP_READ;
int ret = WSARecv(olp->s, &olp->wbuf, 1, &olp->dwBytes, &olp->dwFlags, &olp->ol, NULL);
if (ret == SOCKET_ERROR)
if (WSAGetLastError() != ERROR_IO_PENDING)
return false;
return true;
}
void CIOCP::OnAccept(SOCKET socket)
{
this->SetIoCompletionPort(socket, NULL);
}
//===================================================================================
bool CIOCPClient::Connect(char *ip, int port)
{
// 連接服務(wù)器
if (!bInit)
if (!Init())
return false;
// 初始化連接socket
m_socket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
if (m_socket == SOCKET_ERROR)
{
// printf("cocket Create fail");
return false;
}
// 填寫(xiě)服務(wù)器地址信息
// 端口為1982
// IP地址為INADDR_ANY，注意使用htonl將IP地址轉(zhuǎn)換為網(wǎng)絡(luò)格式ServerAddr.sin_family = AF_INET;
sockaddr_in ClientAddr;
ClientAddr.sin_family = AF_INET;
ClientAddr.sin_port = htons(port);
ClientAddr.sin_addr.s_addr = inet_addr(ip);
// 綁定監(jiān)聽(tīng)端口
bind(m_socket, (SOCKADDR *)&ClientAddr, sizeof(ClientAddr));
if (connect(m_socket, (SOCKADDR *)&ClientAddr, sizeof(ClientAddr)) == SOCKET_ERROR)
{
return false;
}
if ((m_hIocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, NULL, 0)) == NULL) // 創(chuàng)建完成端口的句柄
return false;
this->m_workThread = true;
g_hwThread = CreateThread(NULL, 0, WorkThread, (LPVOID)this, 0, &m_wthreadID); // 創(chuàng)建工作線程，用來(lái)處理完成端口消息的
this->SetIoCompletionPort(m_socket, &m_socket); // 設(shè)置完成端口監(jiān)聽(tīng)的socket
return true;
}
void CIOCPClient::Send(char *buf, int len)
{
send(m_socket, buf, len, 0);
}
///////////////////////////////////////////////////////////////////////////////////
// IOCPclient 應(yīng)用代碼
#include "stdafx.h"
#include "IOCP.h"
#include "TClientSocket.h"
class Iocp :public CIOCPClient
{
void OnRead(void * p, char *buf, int len)
{
printf(buf);
Sleep(1000);
this->Send(buf, len);
}
};
int _tmain(int argc, _TCHAR* argv[])
{
Iocp iocp;
iocp.Init();
iocp.Connect("127.0.0.1", 4311);
iocp.Send("test\0", 5);
gets(new char[1000]);
return 0;
}