当前位置：首页 > java >正文

Windows下的异步IO通知模型

java 2025/8/27 18:03:28

异步通知IO模型

这种模型可以视为select函数模型的改进方式。

同步和异步

异步主要指不一致，在数据IO中非常有用。在Windows中的send和recv函数进行的是同步IO。函数返回的事件和数据被完整移动到输出、输入缓冲中的时间一致。

同步IO和异步IO函数的主要区别是返回的时刻与数据收发完成的时刻不一致。

通过使用异步IO可以更有效地使用CPU。在移动数据时可以去执行别的任务。

理解异步通知IO模型

顾名思义，通知IO是指发生了IO相关的特定情况，典型的通知IO模型是select函数。select函数是同步通知IO模型，因为该函数的返回时间与IO相关事件发生的时间是一致的。

异步通知IO模型中的函数返回时间与IO状态无关。在异步通知IO模型中，指定监视对象的函数和验证实际状态变化的函数是相互分离的。因此指定监视对象之后可以离开执行其它任务，最后再回来验证状态变化。

实现异步通知IO模型

WSAEventSelect函数用于指定某一套接字为事件监听对象。只要传入的套接字发生INetworkEvents中指定的事件之一，该函数便会将hEventObject所指向的内核对象改为signaled状态。因此该函数又称为连接事件对象和套接字的函数。

#include <winsock2.h>int WSAEventSelect(SOCKET s, WSAEVENT hEventObject, long lNetworkEvent);//成功返回0，失败时返回SOCKET_ERROR

无论事件发生与否，该函数调用后会立刻返回。

我们之前使用CreateEvent函数创建事件对象，在只需要创建manual-reset模式non-signaled状态的事件对象可以使用如下函数：

#include <winsock2.h>WSAEVENT WSACreateEvent(void);//成功时返回事件对象句柄，失败时返回WSA_INVALID_EVENT

通过WSACloseEvent函数销毁事件对象。

使用WSAWaitForMultipleEvent函数验证事件是否发生。

使用WSAEnumNetworkEvents函数区分事件类型。同时该函数将manuak-reset模式的事件对象改为non-signaled状态。

代码示例

下面这份代码展示了如何使用异步IO通知模型实现回声服务器端：

/*
使用异步通知IO模型的服务器端，实现回声服务器端大致实现思路：1. 创建接收客户端连接亲求的套接字hServSock，给该套接字分配地址，使该套接字变为监听状态2. 使用WSAEventSelect()监听hServSock的FD_ACCEPT事件，并将hServSock放入待监视的套接字数组中3. 在循环中使用WSAWaitForMultipleEvents()验证待监视的套接字数组中是否发生了事件对象的状态改变1. 得到第一个发生转变为signaled状态的事件对象句柄的对应下标，从该下标开始逐个验证1. 若验证事件对象发生转变，使用WSAEnumNetworkEvents()区分事件对象状态发生转变的原因2. 分别对对应事件进行处理
*/#include <stdio.h>
#include <string.h>
#include <WinSock2.h>#define BUF_SIZE 100void CompressSockets(SOCKET hSockArr[], int idx, int total); //断开连接后，该函数用于整理套接字
void CompressEvents(WSAEVENT hEventArr[], int idx, int total); //断开连接后，该函数用于整理事件对象句柄
void ErrorHandling(const char* msg);int main(int argc, char* argv[])
{//-------------以下是一些基本的准备工作if (argc != 2)ErrorHandling("argc error");WSADATA wsaData;if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0)ErrorHandling("WSAStartup() error");SOCKET hServSock, hClntSock;if ((hServSock = socket(AF_INET, SOCK_STREAM, 0)) == SOCKET_ERROR)ErrorHandling("socket() error");sockaddr_in servAdr, clntAdr;memset(&servAdr, 0, sizeof(servAdr));servAdr.sin_family = PF_INET;servAdr.sin_addr.s_addr = htonl(INADDR_ANY);servAdr.sin_port = htons(atoi(argv[1]));if (bind(hServSock, (sockaddr*)&servAdr, sizeof(sockaddr)) == SOCKET_ERROR)ErrorHandling("bind() error");if (listen(hServSock, 5) == SOCKET_ERROR)ErrorHandling("listen() error");//----------------------准备工作结束SOCKET hSockArr[WSA_MAXIMUM_WAIT_EVENTS]; //存储客户端套接字WSAEVENT hEventArr[WSA_MAXIMUM_WAIT_EVENTS]; //存储对应客户端发生的事件WSAEVENT newEvent;WSANETWORKEVENTS netEvent;int numOfClntSock = 0; //客户端套接字数量int strLen;int posInfo, //用于接收WSAWaitForMultipleEvent函数的返回值startIdx; //转变为signaled状态的事件对象的句柄的下标int clntAdrLen;char msg[BUF_SIZE];//该函数用于创建manual-reset模式下的non-signaled状态事件对象newEvent = WSACreateEvent();//该函数指定hServSock为监听对象，监听FD_ACCEPT事件，立即返回//只要发生第三个参数指定的事件之一，该函数就将newEvent指向的内核对象改为signaled状态if (WSAEventSelect(hServSock, //希望监听的套接字newEvent, //传递事件对象句柄以验证事件发生与否FD_ACCEPT //希望监听的事件：是否有新的连接请求) == SOCKET_ERROR)ErrorHandling("WSAEventSelect() error");//应该维护套接字和事件对象句柄之间的对应关系，可以通过一个下标在两个数组中找到相关联的套接字和事件对象//所以下列三行代码是一个公式，旨在将hServSock和其它客户端套接字一同进行监视hSockArr[numOfClntSock] = hServSock; //把接收客户端请求的套接字句柄存入hEventArr[numOfClntSock] = newEvent; //把与hServSock关联的事件对象存入numOfClntSock++;while (true){//该函数用于验证是否发生事件，有事件状态转为signaled时才返回posInfo = WSAWaitForMultipleEvents(numOfClntSock, //需要验证是否转为signaled状态的事件对象的个数hEventArr, //事件对象句柄数组首地址FALSE, //有一个事件对象转为signaled状态便返回WSA_INFINITE,FALSE //传递TRUE时进入可警告可等待状态);//使用返回索引值减去宏得到转变为signaled状态事件对象句柄对应的索引startIdx = posInfo - WSA_WAIT_EVENT_0;//从特定位置开始逐个验证事件是否发生for (int i = startIdx; i < numOfClntSock; i++){//由于先前已经对一组事件对象进行了验证，所以此处不进行等待，立即返回//发生了转换则处理，没有则验证下一个事件对象是否转换int sigEventIdx = WSAWaitForMultipleEvents(1, &hEventArr[i],TRUE, 0, FALSE);if (sigEventIdx == WSA_WAIT_FAILED || sigEventIdx == WSA_WAIT_TIMEOUT)continue;else{//这行代码用于sigEventIdx = i;//该函数用于区分事件类型WSAEnumNetworkEvents(hSockArr[sigEventIdx], //发生事件的套接字句柄hEventArr[sigEventIdx], //与套接字相关联的事件对象句柄&netEvent //保存事件发生的类型信息);if (netEvent.lNetworkEvents & FD_ACCEPT)//请求连接时{if (netEvent.iErrorCode[FD_ACCEPT_BIT] != 0){puts("accpet error");break;}clntAdrLen = sizeof(clntAdr);hClntSock = accept(hSockArr[sigEventIdx], (sockaddr*)&clntAdr, &clntAdrLen);newEvent = WSACreateEvent();WSAEventSelect(hClntSock, newEvent, FD_READ | FD_CLOSE);hEventArr[numOfClntSock] = newEvent;hSockArr[numOfClntSock] = hClntSock;numOfClntSock++;puts("connected new client...");}if (netEvent.lNetworkEvents & FD_READ) //接收数据时{if (netEvent.iErrorCode[FD_READ_BIT] != 0){puts("read error");break;}strLen = recv(hSockArr[sigEventIdx], msg, sizeof(msg), 0);send(hSockArr[sigEventIdx], msg, strLen, 0);}if (netEvent.lNetworkEvents & FD_CLOSE) //断开连接时{if (netEvent.iErrorCode[FD_CLOSE_BIT] != 0){puts("close error");break;}WSACloseEvent(hEventArr[sigEventIdx]);closesocket(hSockArr[sigEventIdx]);numOfClntSock--;CompressSockets(hSockArr, sigEventIdx, numOfClntSock);CompressEvents(hEventArr, sigEventIdx, numOfClntSock);}}}}WSACleanup();return 0;
}void CompressSockets(SOCKET hSockArr[], int idx, int total)
{for (int i = 0; i < total; i++){hSockArr[i] = hSockArr[i + 1];}
}void CompressEvents(WSAEVENT hEventArr[], int idx, int total)
{for (int i = 0; i < total; i++){hEventArr[i] = hEventArr[i + 1];}
}void ErrorHandling(const char* msg)
{perror(msg);exit(1);
}