C++后台开发之网络IO模型

为了解决网络IO中的问题，学者们提出了4种网络IO模型：①阻塞IO模型；②非阻塞IO模型；③多路IO复用模型；④异步IO模型。

1.阻塞IO模型

在Linux中，默认情况下所有的socket都是阻塞的，阻塞和非阻塞的概念描述的是用户线程调用内核IO操作的方式：阻塞是指IO操作需要彻底完成后才返回到用户空间；而非阻塞是指IO操作被调用后立即返回给用户一个状态值，不需要等到IO操作彻底完成。典型

2.非阻塞IO模型

当用户进程发出read操作时，如果内核中的数据还没有准备好，那么它并不会block用户进程，而是立刻返回一个错误。从用户进程角度讲，它发起一个read操作后，并不需要等待，而是马上就得到了一个结果。当用户进程判断结果是一个错误时，它就知道数据还没有准备好，于是它可以再次发送read操作。一旦内核中的数据准备好了，并且又再次收到了用户进程的系统调用，那么它马上就将数据复制到了用户内存中，然后返回正确的返回值。

使用如下函数可以将fd换成非阻塞状态：

fcntl( fd, F_SETFL, O_NONBLOCK );

3.多路复用型IO模型

多路IO复用，有时也称为事件驱动IO。它的基本原理就是有个函数（如select）会不断地轮询所负责的所有socket，当某个socket有数据到达了，就通知用户进程。

4.异步IO模型

用户进程发起read操作之后，立刻就可以开始去做其他的事；而另一方面，从内核的角度，当它收到一个异步的read请求操作之后，首先会立刻返回，所以不会对用户进程产生任何阻塞。然后，内核会等待数据准备完成

重点介绍多路复用IO模型中的三种函数，select，poll和epoll函数。

1.select函数

select函数原型如下：

int select(int maxfdp,fd_set *readfds,fd_set *writefds,fd_set *errorfds,struct timeval*timeout);

（1）参数maxfdp是一个整数值，是指集合中所有文件描述符的范围，其值为所有文件描述符的最大值加1。

（2）参数timeout是select的超时时间，这个参数至关重要，它可以使select处于3种状态：若传入参数timeout=NULL，那么select将处于阻塞态，直到监视到文件描述符集合中某个描述符变化为止；如果将参数timeout=0，则select将处于非阻塞状态，不管文件是否变化，有变化返回一个正值，无变化返回0；如果将参数timeout设置为大于0，即select在timeout时间内阻塞，超时时间之内有事件到来就返回了，否则在超时后不管怎样一定返回，超时返回0，有变化返回一个正值。

（3）readfds，writefds和errorfds表示分别指向fd_set 描述符集合的指针，分别监视描述符集合中描述符的读，写和异常变化。

fd_set 结构可以理解为一个fd的集合，其宏定义的控制函数如下所示：

fd_set set;

FD_ZERO(&set); /*将set清零*/

FD_SET(fd, &set); /*将fd加入set */

FD_CLR(fd, &set); /*将fd从set中清除*/

FD_ISSET;/*如果fd在set中则真，函数返回时，在set中的为变化描述符*/

例子程序：

使用select循环读取键盘输入：

#include "main.h"

int main(){

int keyboard;

int ret,i;

char c;

fd_set readfd;

struct timeval timeout;
//只读，非阻塞的方式打开
keyboard = open("/dev/tty",O_RDONLY | O_NONBLOCK);

assert(keyboard>0);

while(1){

timeout.tv_sec=1;

timeout.tv_usec=0;

FD_ZERO(&readfd);

FD_SET(keyboard,&readfd);

ret=select(keyboard+1,&readfd,NULL,NULL,&timeout);

if(FD_ISSET(keyboard,&readfd)) {

i=read(keyboard,&c,1);
if('n'==c)
continue;
printf("The input is %c\n",c);
if('q'==c)
break;
}
}
return 0;
}

使用select函数提高服务器处理能力：

#include <main.h>

#define DEFAULT_PORT 6666

int main(int argc,char **argv)
{
int serverfd; //监听fd
int acceptfd; //向客户端传递数据的fd

struct sockaddr_in my_addr;
struct sockaddr_in their_addr;

unsigned int sin_size;
unsigned int myport = 6666;
unsigned int lisnum = 10;

if((serverfd = socket(AF_INET,SOCK_STREAM,0)) == -1)
{
perror("socket");
return -1;
}

printf("socket OK!\n");
my_addr.sin_family = AF_INET;
my_addr.sin_port = htons(DEFAULT_PORT);
my_addr.sin_addr.s_addr = INADDR_ANY;

bzero(&(my_addr.sin_zero),0);

if(bind(serverfd,(struct sockaddr *)(&my_addr),sizeof(struct sockaddr)) == -1)
{
perror("bind");
return -2;
}
printf("bind OK!\n");

if(listen(serverfd,lisnum) == -1)
{
perror("listen");
return -3
}

fd_set client_fdset;
int maxsock;
struct timeval tv;
int client_sock[5]={0}; //需要监听的客户端fd
bzero((void *)client_sock,sizeof(client_sock));
int connect_cnt = 0;
maxsock = serverfd;

char buff[1024];
int ret;

while(1)
{
FD_ZERO(&client_fdset);
FD_SET(serverfd,&client_fdset);
tv.tv_sec = 30;
tv.tv_usec = 0;

for(int i=0;i<5;i++)
{
if(client_sock[i]!=0)
FD_SET(client_sock[i],&client_fdset);
}

ret = select(maxsock+1,client_fdset,NULL,NULL,tv);

if(ret<0)
{
printf("select error!\n");
break;
}
if(ret==0)
{
printf("time out!\n");
continue;
}

for(int i=0;i<connect_cnt;i++)
{
if(FD_ISSET(client_sock[i],&client_fdset))
{
ret = recv(client_sock[i],buff,1024,0);

if(ret<0) //如果ret<0表明该客户端已经关闭
{
close(client_sock[i]);
FD_CLR(client_sock[i],client_fdset);
client_sock[i] = 0;
}else{
printf("recv from client[%d]: %s\n",i,buff);
}

}
}

//serverfd描述符变化，表示新的客户端连接进来
if(FD_ISSET(serverfd,&client_fdset))
{
struct sockaddr_in client_addr;
unsigned int size = sizeof(sockaddr_in);
int sock_client = accept(serverfd,(struct sockaddr *)&client_addr,&size);

if(sock_client<0)
perror("accept error!\n");
continue;
}

if(connect_cnt<5)
{
client_sock[connect_cnt++] = sock_client;
bzero(buff,1024);
strcpy(buff,"welcome !\n");
send(sock_client,buff,1024,0);
bzero(buff,1024);

if(sock_client>maxsock)
maxsock = sock_client;

}else{
printf("to much connect_cnt!\n");
break;
}

}

for(int i=0;i<5;i++)
{
if(client_sock[i]!=0)
close(client_sock[i]);
}

close(serverfd);
return 0;

}

2.poll函数

poll函数原型如下：

#include <poll.h>

　　int poll(struct pollfd fd[], nfds_t nfds, int timeout);
//参数pollfd
　　struct pollfd
　　{
　　int fd;          //文件描述符
　　short events;    //请求的事件
　　short revents;   //返回的事件
　　};

fd：需要监控的fd结构体数组；

nfds：用于标记数组fds中的struct pollfd结构元素的总数量，结合数组长度来理解；

timeout：poll函数调用阻塞的时间，单位是ms。

poll的基本思想和select一致，服务器端poll多路复用代码如下：

#include "main.h"

#define IPADDRESS "127.0.0.1"
#define PORT 6666
#define MAXSIZE 1024
#define LISTENQ 5
#define OPEN_MAX 1000
#define INFTIM -1

int bind_and_listen()
{
struct sockaddr_in my_addr;
int serverfd = socket(AF_INET,SOCK_STREAM,0);
if(serverfd = -1)
{
perror("socket\n");
return -1;
}
my_addr.sin_family = AF_INET;
my_addr.sin_port = PORT;
my_addr.sin_addr.s_addr = INADDR_ANY;

int ret = bind(serverfd,(struct sockaddr *)(&my_addr),sizeof(struct sockaddr));

if(ret = -1)
{
perror("bind\n");
return -2;
}

if(listen(serverfd,LISTENQ) == -1)
{
perror("listen\n");
return -3;
}

printf("OK!\n");
return serverfd;

}

int main()
{
int serverfd = bind_and_listen();
int max = 1;
int ret = 0;

if(serverfd<0)
return -1;

struct pollfd clientfds[OPEN_MAX];

clientfds[0].fd = serverfd;
clientfds[0].events = POLLIN;

for(int i=1;i<OPEN_MAX;i++)
{
clientfds[i].fd = -1;
}

while(1)
{
ret = poll(clientfds,max+1;INFTIM);
if(ret<0)
{
perror("poll");
exit(1);
}
if(clientfds[0].revents & POLLIN) //判断是否有连接
{
socklen_t clientlen = sizeof(sockaddr_in);
int connfd = accept(serverfd,(struct sockaddr*)(&my_addr),&clientlen);
if(connfd == -1)
{
perror("accept error!\n");
exit(1);
}else{
for(int i=0;i<OPEN_MAX;i++)
{
if(clientfds[i].fd = -1)
{
clientfds[i].fd = connfd;
clientfds[i].events = POLLIN;
break;
}
}
if(i == OPEN_MAX)
{
printf("to much clients\n");
exit(0);
}
if(i > max-1)
max = i+1;
}
}

for(int i=1;i<OPEN_MAX;i++)
{
if(clientfds[i].fd = -1)
continue;
if(clientfds[i].revents & POLLIN)
{
int readlen = read(clientfds[i].fd,buf,MAXSIZE);
if(readlen == 0)//表明该fd已经关闭
{
close(clientfds[i].fd);
clientfds[i].fd = -1;
continue;
}

write(S
924a
TDOUT_FILENO,buf,readlen);
}
}

}

}

3.epoll函数

epoll的接口非常简单，一共就三个函数：

int epoll_create(int size);

创建一个epoll的句柄，size用来告诉内核这个监听的数目一共有多大。这个参数不同于select()中的第一个参数，给出最大监听的fd+1的值。需要注意的是，当创建好epoll句柄后，它就是会占用一个fd值，在linux下如果查看/proc/进程id/fd/，是能够看到这个fd的，所以在使用完epoll后，必须调用close()关闭，否则可能导致fd被耗尽。

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

epoll的事件注册函数，它不同与select()是在监听事件时告诉内核要监听什么类型的事件，而是在这里先注册要监听的事件类型。第一个参数是epoll_create()的返回值，第二个参数表示动作，用三个宏来表示：

EPOLL_CTL_ADD：注册新的fd到epfd中；

EPOLL_CTL_MOD：修改已经注册的fd的监听事件；

EPOLL_CTL_DEL：从epfd中删除一个fd；

第三个参数是需要监听的fd，第四个参数是告诉内核需要监听什么事，struct epoll_event结构如下：

typedef union epoll_data {
void *ptr;
int fd;
__uint32_t u32;
__uint64_t u64;
} epoll_data_t;

struct epoll_event {
__uint32_t events; /* Epoll events */
epoll_data_t data; /* User data variable */
};

events可以是以下几个宏的集合：

EPOLLIN ：表示对应的文件描述符可以读（包括对端SOCKET正常关闭）；

EPOLLOUT：表示对应的文件描述符可以写；

EPOLLPRI：表示对应的文件描述符有紧急的数据可读（这里应该表示有带外数据到来）；

EPOLLERR：表示对应的文件描述符发生错误；

EPOLLHUP：表示对应的文件描述符被挂断；

EPOLLET： 将EPOLL设为边缘触发(Edge Triggered)模式，这是相对于水平触发(Level Triggered)来说的。

EPOLLONESHOT：只监听一次事件，当监听完这次事件之后，如果还需要继续监听这个socket的话，需要再次把这个socket加入到EPOLL队列里

int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);

等待事件的产生，类似于select()调用。参数events用来从内核得到事件的集合，maxevents告之内核这个events有多大，这个 maxevents的值不能大于创建epoll_create()时的size，参数timeout是超时时间（毫秒，0会立即返回，-1将不确定，也有说法说是永久阻塞）。该函数返回需要处理的事件数目，如返回0表示已超时。