iocp之简单实现

2023-10-26

一、步骤

这里写图片描述

二、代码

1、头文件

#include <WinSock2.h>
#include <Windows.h>

#ifdef _WINDOWS
#pragma comment(lib,"Ws2_32.lib")   //Socket编程需用的动态链接库
#pragma comment(lib,"Kernel32.lib") //IOCP需要用到的动态链接库
#endif

//LOG_VIEW->log(str, ...)	功能：输出日志到界面

2、加载套接字库

bool IocpServer::loadSockDll()
{
    WSAData wsa;
    int ret = WSAStartup(MAKEWORD(2, 2), &wsa);
    if (0 != ret)
    {
        LOG_VIEW->log("加载套接字库失败，err:%d", ret);
        return false;
    }

    if (LOBYTE(wsa.wVersion) != 2 || HIBYTE(wsa.wVersion) != 2)
    {
        WSACleanup();
        LOG_VIEW->log("没有找到2.2版本套接字库");
        return false;
    }

    return true;
}

3、创建IOCP+创建IOCP工作线程

bool IocpServer::initIocp()
{
    LOG_VIEW->log("创建IOCP端口...");
    m_iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);
    if (NULL == m_iocp)
    {
        LOG_VIEW->log("创建IOCP端口失败");
        return false;
    }

    LOG_VIEW->log("创建IOCP处理线程...");
    SYSTEM_INFO info;
    GetSystemInfo(&info);
    for (int i = 0; i < info.dwNumberOfProcessors*2; ++i)
    {
        LOG_VIEW->log("创建线程:%d", i+1);
        thread* th = new thread(std::bind(&IocpServer::workerProc, this, i+1));
        m_workerThread.push_back(th);
    }

    return true;
}

CreateIoCompletionPort参数说明：
param：有效的文件句柄或 INVALID_HANDLE_VALUE(一般用这个)
param：NULL(新建IOCP时)
param：0。
param：操作系统允许同时处理I/O完成端口数据包的最大线程数，如果此参数为零，则最大线程数等于cpu数量。
创建的工作线程数量为cpu数量*2，考虑到线程查询修改数据库时有等锁的情况，多余的线程可以替补。

4、创建socket、绑定、监听+创建接收客户端连接请求线程

bool IocpServer::initSocket(const string& ip, unsigned short port)
{
    LOG_VIEW->log("创建套接字");
    m_socket = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0, WSA_FLAG_OVERLAPPED);
    if (INVALID_SOCKET == m_socket)
    {
        LOG_VIEW->log("创建socket失败");
        return false;
    }

    LOG_VIEW->log("绑定");
    sockaddr_in addr;
    addr.sin_family = AF_INET;
    addr.sin_port = htons(port);
    addr.sin_addr.S_un.S_addr = inet_addr(ip.c_str());
    if(0 != ::bind(m_socket, (sockaddr*)&addr, sizeof(addr)))
    {
        LOG_VIEW->log("绑定socket失败");
        return false;
    }

    LOG_VIEW->log("监听");
    if (0 != listen(m_socket, 100))
    {
        LOG_VIEW->log("监听socket失败");
        return false;
    }

    LOG_VIEW->log("创建accept线程");
    m_accpetThread = new thread(std::bind(&IocpServer::acceptProc, this));

    return true;
}

WSASocket说明：socket的扩展版，最后一个参数WSA_FLAG_OVERLAPPED表示创建具有重叠属性的socket。只有具有重叠属性才够使用WSASend、WSARecv、WSAIoctl等函数。

5、接收客户端连接请求线程

void IocpServer::acceptProc()
{
    sockaddr_in addr;
    int addrLen = sizeof(addr);

    LOG_VIEW->log("accept线程工作中...");
    while (!m_stopThread)
    {
        memset(&addr, 0, sizeof(addr));
        SOCKET clientSocket = accept(m_socket, (sockaddr*)&addr, &addrLen);
        if (INVALID_SOCKET == clientSocket)
        {
            LOG_VIEW->log("接受连接失败");
            break;
        }

		//将IOCP端口与客户端SOCKET绑定
        CreateIoCompletionPort((HANDLE)clientSocket, m_iocp, (ULONG_PTR)clientSocket, 0);

        //通知新增连接，收到通知的模块处理为客户端创建会话对象
        string ip = inet_ntoa(addr.sin_addr);
        unsigned short port = htons(addr.sin_port);
        if (!m_notify->addConnect(clientSocket, ip, port))
        {
            LOG_VIEW->log("添加会话连接失败");
            continue;
        }

		//为客户端投递接收请求
        sIocpContent* ct = new sIocpContent(BUFF_LEN, eIocpType::IocpRecv);
        if (!deliveryRecvRequest(clientSocket, ct))
        {
            LOG_VIEW->log("投递IOCP接收请求失败");
            delete ct;
            break;
        }
    }

    LOG_VIEW->log("accept线程自然退出");
}

CreateIoCompletionPort参数说明：
param：有效的文件句柄或 INVALID_HANDLE_VALUE，这里用客户端SOCKET
param：现有I/O完成端口的句柄(与socket绑定时)
param：传送给处理函数的参数，即传给GetQueuedCompletionStatus第三个参数
param：操作系统允许同时处理I/O完成端口数据包的最大线程数，如果此参数为零，则最大线程数等于cpu数量。

6、IOCP工作线程

void IocpServer::workerProc(int tag)
{
    DWORD revLen = 0;
    SOCKET pKey = 0;
    LPOVERLAPPED pOl = NULL;

    LOG_VIEW->log("iocp线程[%d]工作中...", tag);
    while (!m_stopThread)
    {
        bool ret = GetQueuedCompletionStatus(m_iocp, &revLen, (PULONG_PTR)&pKey, &pOl, INFINITE);
        if (!ret)
        {
		    //错误处理
        }
		else
		{
			sIocpContent* content = (sIocpContent*)pOl;
			switch (content->type) {
	        case eIocpType::IocpSend:
	            //发送数据成功通知
	            //处理发送数据占的内存
	            //这部分内存使用内存池，可以重复利用
	            break;
	        case eIocpType::IocpRecv:
	            //收到客户端发来数据的通知
	            //将数据发送到这个客户端的会话对象，会话对象负责解包+处理
	            //为客户端投递接收请求，就可以继续收到客户端发来的数据了
	            memset(&content->ol, 0, sizeof(content->ol));
	            deliveryRecvRequest(pKey, content);
	            break;
	        default:
	            break;
	        }
		} 
    }

    LOG_VIEW->log("iocp线程[%d]自然退出", tag);
}

GetQueuedCompletionStatus参数说明
param：之前创建的IOCP端口
param：收到的数据长度
param：IOCP端口与SOCKET绑定时，CreateIoCompletionPort的第三个参数
param：投递请求时，CreateIoCompletionPort的第六个参数，是个指针
param：等待通知的时间，INFINITE即一直等

7、为客户端投递接收请求

bool IocpServer::deliveryRecvRequest(unsigned sock, sIocpContent *content)
{
    DWORD revLen = 0, lpFlag = 0;
    int ret = WSARecv(sock, &(content->data), 1, &revLen, &lpFlag, &(content->ol), NULL);
	if (ret == 0)
    {
		//收到客户端发来的数据
		//将数据发送到这个客户端的会话对象，会话对象负责解包+处理
		//为客户端投递接收请求，就可以继续收到客户端发来的数据了
    }
    else if (ret == SOCKET_ERROR && WSAGetLastError() != WSA_IO_PENDING)
    {
	    //错误处理
	    return false;
    }

    return true;
}

WSARecv参数说明
param：客户端套接字
param：指向WSABUF结构数组的指针
param：参数2数组的长度
param：接收操作立即完成时，此调用接收的字节数的指针
param：标志的指针，一般初始为0
param：指向WSAOVERLAPPED结构的指针，被底层加工后传给GetQueuedCompletionStatus的第四个参数
param：指向完成接收操作时调用的完成例程的指针，传NULL
返回值：若cpu空闲，可能马上就收到数据，返回0。若发生错误，返回SOCKET_ERROR。但若错误码是WSA_IO_PENDING表示，CPU现在忙，这个接收任务放一边，底层完成了再告诉你。

8、释放资源(愤怒地写在一起)

	//线程自然退出标志
	m_stopThread = true;
	
	//释放SOCKET
	if (INVALID_SOCKET != m_socket)
    {
        LOG_VIEW->log("释放套接字");
        closesocket(m_socket);
        m_socket = INVALID_SOCKET;
    }

    if (m_accpetThread)
    {
        LOG_VIEW->log("等待accept线程结束");
        if (m_accpetThread->joinable())
        {
            m_accpetThread->join();
        }

        delete m_accpetThread;
        m_accpetThread = NULL;
        LOG_VIEW->log("accept线程结束");
    }

	//释放IOCP相关
	if (m_iocp)
    {
        LOG_VIEW->log("关闭IOCP端口");
        CloseHandle(m_iocp);
        m_iocp = NULL;
    }

    for (size_t i = 0; i < m_workerThread.size(); ++i)
    {
        char buff[8] = {0};
        itoa(i+1, buff, 10);

        if (m_workerThread[i]->joinable())
        {
            LOG_VIEW->log("等待线程结束:%d", i+1);
            m_workerThread[i]->join();
        }

        LOG_VIEW->log("线程结束:", i+1);
        delete m_workerThread[i];
    }
    m_workerThread.clear();
	
	//卸载套接字库
	WSACleanup();

9、sIocpContent结构体

#include <Winsock2.h>

#define BUFF_LEN    (1024)

enum eIocpType
{
    IocpNone = 0,
    IocpRecv,
    IocpSend,
};

struct sIocpContent
{
    OVERLAPPED	ol;
    WSABUF		data;
    eIocpType	type;

    sIocpContent(int len, eIocpType tp)
    {
        memset(&ol, 0, sizeof(ol));
        data.buf = new char[len]();
        data.len = len;
        type = tp;
    }

    sIocpContent(char* buff, int len)
    {
        memset(&ol, 0, sizeof(ol));
        data.buf = buff;
        data.len = len;
        type = eIocpType::IocpSend;
    }

    ~sIocpContent()
    {
        delete[] data.buf;
    }
};

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)