http://blog.csdn.net/yahle
#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; 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); // 把一个socket与一个自定义的结构体关联到完成端口(相当于把socket与一个结构体变量进行绑定), // 这样当发送上面3种网络事件的时候,该结构体变量会再传回给程序 // 这样就可以区分当前网络事件是那个socket发出的 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 实现文件 #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) { // 客户端断开连接 pthis->OnClose(re); OVERLAPPEDPLUS *olp = (OVERLAPPEDPLUS *)pOverlap; closesocket(olp->s); delete olp; // 释放 与socket绑定的结构体变量 continue; } if (re == NULL) return 0; OVERLAPPEDPLUS *olp = (OVERLAPPEDPLUS *)pOverlap; switch(olp->OpCode) { case OP_READ: pthis->OnRead(re, olp->wbuf.buf, berByte); // 调用 OnRead() 通知应用程序,服务器收到来自客户端的网络数据 WSARecv(olp->s, &olp->wbuf, 1, &olp->dwBytes, &olp->dwFlags, &olp->ol, NULL); // 继续调用一个接收的 I/O 异步请求 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) { // 错误处理 } pthis->OnAccept(client); // 调用 OnAccept()通知应用程序有新客户端连接 } 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) // 创建完成端口的句柄 return false; this->m_acceptThread = true; g_haThread = CreateThread(NULL, 0, AcceptThread, (LPVOID)this, 0, &m_athreadID); // 创建连接线程,用来接收客户端的连接 this->m_workThread = true; g_hwThread = CreateThread(NULL, 0, WorkThread, (LPVOID)this, 0, &m_wthreadID); // 创建工作线程,用来处理完成端口消息的 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) { // 这里可以使用用户自定义的缓冲区地址,如果用户不想设置,也可以采用默认分配的缓冲区 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) { // 连接服务器 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; } // 填写服务器地址信息 // 端口为1982 // IP地址为INADDR_ANY,注意使用htonl将IP地址转换为网络格式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); // 绑定监听端口 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) // 创建完成端口的句柄 return false; this->m_workThread = true; g_hwThread = CreateThread(NULL, 0, WorkThread, (LPVOID)this, 0, &m_wthreadID); // 创建工作线程,用来处理完成端口消息的 this->SetIoCompletionPort(m_socket, &m_socket); // 设置完成端口监听的socket return true; } void CIOCPClient::Send(char *buf, int len) { send(m_socket, buf, len, 0); }
#include "stdafx.h"
(yahle)#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; } |