socket多路复用

目录

多路复用的相关内容：

多路复用定义：

同步/异步：

阻塞/非阻塞：

常见的IO模型有五种：

多路复用实现的三种方式：

1、select多路复用：

2、poll多路复用：

3、epoll多路复用：

三种多路复用的代码实现方式：

1、select()多路复用实现网络socket服务器多路并发的流程图及代码：

流程图：

代码实现：

2、poll()多路复用实现网络socket服务器多路并发的代码：

3、epoll()多路复用实现网络socket服务器多路并发的代码：

代码：

多路复用的相关内容：

多路复用定义：

• 多路复用是一种同步I/O模型，实现一个线程可以监视多个文件句柄；
• 一旦某个文件句柄就绪，就能够通知应用程序进行相应的读写操作；
• 没有文件句柄就绪就会阻塞应用程序，交出CPU。

同步是在Linux下进行网络编程时，同步(Sync)/异步(Async)，阻塞(Block)/非阻塞(Unblock)四种调用方式之一。

同步/异步：

同步（sync）和异步（async）的概念描述的是用户线程与内核的交互方式。
同步是指用户线程发起IO请求后需要等待或者轮询内核IO操作完成后才能继续执行；
异步是指用户线程发起IO请求后仍继续执行，当内核IO操作完成后会通知用户线程，或者调用用户线程注册的回调函数。

阻塞/非阻塞：

阻塞和非阻塞的概念描述的是用户线程调用内核操作的方式。
阻塞是指IO操作在没有接收完数据或者没有得到结果之前不会返回，需要彻底完成后才返回到用户空间；
非阻塞是指IO操作后立即返回给用户一个状态值，无需等到IO操作彻底完成。

常见的IO模型有五种：

在Linux下进行网络编程是，服务器端编程需要构造高性能的IO模型，常见的IO模型有五种：

（1）同步阻塞IO（Blocking IO）：即传统的IO模型，在linux中默认情况下所有的socket都是阻塞模式。发 送方发送请求之后一直等待响应，接收方处理请求时进行的的IO操作如果不能马上等到返回结果，就一直等到返回结果后，才响应发送方，期间不能进行其它工作。即传统的IO模型。 几乎所有的程序员第一次接触到的网络编程都是从listen()、read()、write() 等接口开始的，这些接口都是阻塞型的，一个简单的改进方案是在服务器端使用多线程（或多进程）。多线程（或多进程）的目的是让每个连接都拥有独立的线程（或进程），这样任何一个连接的阻塞都不会影响其他的连接。 （2）同步非阻塞IO（Non-blocking IO）：默认创建的socket都是阻塞的，同步非阻塞IO是在同步阻塞IO的基础上，将socket设置为NONBLOCK，这个可以使用ioctl()系统调用设置。这样做发送方可以在发起IO请求后可以立即返回，如果该次读操作并未读取到任何数据，需要不断地发起IO请求，直到数据到达后，才真正读取到数据，继续执行。整个IO请求的过程中，虽然用户线程每次发起IO请求后可以立即返回，但是为了等到数据，仍需要不断地轮询、重复请求，消耗了大量的CPU的资源。一般很少直接使用这种模型，而是在其他IO模型中使用非阻塞IO这一特性。 （3）IO多路复用（IO Multiplexing）：IO多路复用模型是建立在内核提供的多路分离函数select基础之上的，使用select函数可以避免同步非阻塞IO模型中轮询等待的问题，此外poll、epoll都是这种模型。在该种模式下，用户首先将需要进行IO操作的socket添加到select中，然后阻塞等待select系统调用返回。当数据到达时，socket被激活，select函数返回。用户线程正式发起read请求，读取数据并继续执行。从流程上来看，使用select函数进行IO请求和同步阻塞模型没有太大的区别，甚至还多了添加监视socket，以及调用select函数的额外操作，效率更差。但是，使用select以后最大的优势是用户可以在一个线程内同时处理多个socket的IO请求。用户可以注册多个socket，然后不断地调用select读取被激活的socket，即可达到在同一个线程内同时处理多个IO请求的目的。而在同步阻塞模型中，必须通过多线程的方式才能达到这个目的。 (4) 信号驱动IO（signal driven IO）：调用sigaltion系统调用，当内核中IO数据就绪时以SIGIO信号通知请求进程，请求进程再把数据从内核读入到用户空间，这一步是阻塞的。 （5）异步IO（Asynchronous IO）：即经典的Proactor设计模式，也称为异步非阻塞IO。“真正”的异步IO需要操作系统更强的支持。在IO多路复用模型中，事件循环将文件句柄的状态事件通知给用户线程，由用户线程自行读取数据、处理数据。而在异步IO模型中，当用户线程收到通知时，数据已经被内核读取完毕，并放在了用户线程指定的缓冲区内，内核在IO完成后通知用户线程直接使用即可。     相比于IO多路复用模型，信号驱动IO和异步IO并不十分常用，不少高性能并发服务程序使用IO多路复用模型+多线程任务处理的架构基本可以满足需求。况且目前操作系统对异步IO的支持并非特别完善，更多的是采用IO多路复用模型模拟异步IO的方式。

多路复用实现的三种方式：

1、select多路复用：

函数原型及相关说明：

#include 
#include 
struct timeval
{long tv_sec; //secondslong tv_usec; //microseconds
};
FD_ZERO(fd_set* fds) //清空集合
FD_SET(int fd, fd_set* fds) //将给定的描述符加入集合
FD_ISSET(int fd, fd_set* fds) //判断指定描述符是否在集合中
FD_CLR(int fd, fd_set* fds) //将给定的描述符从文件中删除 
int select(int max_fd, fd_set *readset, fd_set *writeset, fd_set *exceptset, struct timeval *timeout);

1. select 函数的 返回值 是就绪描述符的 数目 ，超时时返回 0 ，出错返回 -1 ； 2. 第一个参数 max_fd 指待测试的 fd 的总个数，它的值是待测试的 最大文件描述符加 1 。 Linux 内核从 0 开始到 max_fd-1 扫描文件描述符，如果有数据出现事件( 读、写、异常 ) 将会返回；假设需要监测的文件描述符是 8,9,10 ，那么 Linux 内核实际也要监测 0~7 ，此时真正带测试的文件描述符是0~10 总共 11 个，即 max(8,9,10)+1 ，所以第一个参数是所有要监听的文件描述符中最大的 +1 。 3. 中间三个参数 readset 、 writeset 和 exceptset 指定要让内核测试 读、写和异常 条件的 fd 集合，如果不需要测试的可以设置为NULL； 4. 最后一个参数 是设置 select 的 超时时间 ，如果设置为 NULL 则永不超时； 说明： 1、 select监视并等待多个文件描述符的属性发生变化，它监视的属性分 3 类，分别是 readfds ( 文件描述符有数据到来可读 ) 、 writefds ( 文件描述符可写 ) 、和 exceptfds ( 文件描述符异常 ) 。 调用后 select 函数会阻塞 ，直到有描述符就绪（ 有数据可读、可写、或者有错误异常 ），或者 超时 （ timeout 指定等待时间）发生 函数才返回 。当 select() 函数返回后，可以通过遍历 fdset ，来找到究竟是 哪些 文件描述符就绪。 2、待测试的描述集总是从0， 1 ， 2 ， ... 开始的。 所以， 假如你要检测的描述符为 8 ， 9 ， 10 ， 那么系统实际也要 监测 0 ， 1 ， 2 ， 3 ， 4 ， 5 ， 6, 7 ， 此时真正待测试的描述符的个数为 11 个， 也就是 max （ 8 ， 9 ， 10 ） + 1 3、在Linux 内核有个参数 __FD_SETSIZE 定义了每个 FD_SET 的句柄个数中，这也意味着 select 所用到的 FD_SET 是有限的，也正是这个原因select() 默认只能同时处理 1024 个客户端的连接请求：              /linux/posix_types.h:              #define __FD_SETSIZE 1024 4、基于select的I/O复用模型的是单进程执行可以为多个客户端服务，这样可以减少创建线程或进程所需要的CPU时间片或内存资源的开销；此外几乎所有的平台上都支持select()，其良好跨平台支持是它的另一个优点。当然它也有两个主要的缺点：     1. 每次调用 select()都需要把fd集合从用户态拷贝到内核态，之后内核需要遍历所有传递进来的fd，这时如果客户端fd很多时会导致系统开销很大；     2. 单个进程能够监视的文件描述符的数量存在最大限制，在Linux上一般为1024，可以通过setrlimit()、修改宏定义甚至重新编译内核等方式来提升这一限制，但是这样也会造成效率的降低；

2、poll多路复用：

select()和poll()系统调用在本质上是没有太大区别的，poll()的机制与select()类似，管理多个描述符也是进行轮询，然后根据描述符的状态进行处理，但是poll()函数没有最大文件描述符的数量限制（不过数量过多之后还是会下降）。poll()和select()的一个共同之处就是：包含大量文件描述符的数据会被整体复制于用户态和内核的地址空间之间，而不论这些文件描述符是否就绪，而文件描述符的数据增加之后，系统的开销也就增大。

函数原型及说明：

#include 
struct pollfd
{int fd; /* 文件描述符 */short events; /* 等待的事件 */short revents; /* 实际发生了的事件 */
} ;
int poll(struct pollfd *fds, nfds_t nfds, int timeout);

第一个参数用来指向该结构体类型的数组，每一个pollfd结构体指定了一个被监视的文件描述符，指示poll()监视多个文件描述符，每个结构体的events域是监视该文件描述符的事件掩码，由用户来设置这个域。revents域是文件描述符的操作结果事件掩码，events域中请求的任何事件都可能在revents域中返回。下表列出指定 events 标志以及测试 revents 标志的一些常值：

第二个参数nfds指定数组中监听的元素个数

第三个参数timeout指定等待的毫秒数，无论IO是否准备好，poll都会返回。timeoout指定为负数值表示无线超时，使poll()一直挂起直到一个指定事件发生；timeout为0表示poll调用立即返回并列出准备好I/O的文件描述符，但并不等待其它的事件。也即是，该文件描述符的IO一旦准备好就直接选举出来。

3、epoll多路复用：

epoll是Linux内核为处理
   大批量文件描述符而对poll做了改进的，相当于是Linux下多路复用IO接口select/poll的增强版本，它能显著提高程序在大量并发连接中只有少量活跃的情况下的系统CPU利用率。有一点原因是在获取事件的时候，它不需要遍历整个被监听的描述符集，只要遍历那些被内核IO事件异步唤醒而加入ready队列的描述符集合就行了。
   此外还有一点就是epoll处理提供select/poll那种IO事件的水平触发外，还提供了边缘触发，这一点使得用户空间程序有可能缓存IO状态，减少epoll_wait/epoll_pwait的调用，提高应用程序效率。

函数原型及相关说明：

#include 
int epoll_create(int size)

系统调用epoll_create()创建了一个新的epoll实例，其对应的兴趣列表初始化为空。若成功创建返回文件描述符，若创建出错返回-1，参数size指定了我们想要通过epoll实例来检查的文件描述符的个数，该参数并不是一个上限，而是告诉内核应该如何为内部数据结构划分初始大小。从Linux2.6.8依赖，size参数被忽略不用。

struct epoll_event
{uint32_t    events;        //epoll eventsepoll_data_t data;        //user data
};
typedef union epoll_data
{void         *ptr;int            fd;uint32_t    u32;uint64_t    u64;
}eppoll_data_t;int epoll_ctl(int epfd,int op,int fd,struct epoll_event  *ev)

epoll_ctl该函数能够修改由文件描述符epfd所代表的epoll实例中的兴趣列表，若成功就返回0，若失败就返回-1；
参数说明：
  第一个参数epfd是epoll_create()的返回值；
  第二个参数op是用来指定需要执行的操作，可以是EPOLL_CTL_ADD(将描述符fd添加到epoll实例中的兴趣列表中去)，EPOLL_CTL_MOD(修改描述符上设定的事件)，EPOLL_CTL_DEL(将文件描述符从兴趣列表中移除)
  第三个参数指明了要修改兴趣列表中的哪一个文件描述符的设定
  第四个参数ev是指向结构体epoll_event的指针

    该结构体参数ev为文件描述符fd所作的设置

      - events字段是一个位掩码，它指定了我们为待检查的描述符fd上所感兴趣的事件集合；
      - data字段是一个联合体，当描述符fd稍后称为就绪态时，联合成员可用来指定传回给调用进程的信息

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

 系统调用epoll_wait()返回实例中处于就绪态的文件描述符的信息，数组evlist中的元素个数，如果在timeout超时间间隔内没有任何文件描述符处于就绪态的话就返回0，出错返回-1.

参数说明：
     第一个参数是epoll_create()的返回值
     第二个参数所指向的结构体中返回的是有关就绪态文件描述符的信息，数组evlist的空间由调度者负责申请；

    第三个参数maxevents指定所evlist数组里包含的元素个数；
    第四个参数是timeout用来确定epoll_wait()的阻塞行为
        1. -1：调用将一直阻塞，直到兴趣列表中的文件描述有事件产生或捕捉到一个信号为止
        2.  0：执行一次非阻塞式的检查，看兴趣列表中的描述符产生了哪个事件
        3.  >0:调用将阻塞至多timeout毫秒，直到文件描述符上有事件发生，或者直到捕捉到 一个信号为止

三种多路复用的代码实现方式：

1、select()多路复用实现网络socket服务器多路并发的流程图及代码：

流程图：

代码实现：

 1 #include2 #include3 #include4 #include5 #include6 #include7 #include8 #include9 #include 10 #include 11 #include 12 #include13 #include14 #include15 #include16 17 #define ARRAY_SIZE(x)         (sizeof(x)/sizeof(x[0]))18 19 static inline void msleep(unsigned long ms);20 static inline void print_usage(char *progname);21 int socket_server_init(char *listen_ip, int listen_port);22 23 int main(int argc, char **argv)24 {25     int                 ch;26     char                *progname = NULL;27     int                 ser_port = 0;28     int                 i,j;29     int                 maxfd = 0;30     int                 rv;31     int                 fds_array[1024];32     int                 daemon_run = 0;33     int                 listenfd,connfd;34     fd_set              rdset;35     int                 found;36     char                buf[1024];37         struct option       opts[] = {38         {"daemon",no_argument, NULL, 0},39         {"port", required_argument, NULL, 0},40         {"help", no_argument, NULL, 0},41         {NULL, 0, NULL, 0}42     };43 44     progname = basename(argv[0]);45 46     while( (ch = getopt_long(argc, argv, "bp:h", opts, NULL)) != -1)47     {48         switch( ch )49         {50             case 'b':51                 daemon_run = 1;52                 break;53             case 'p':54                 ser_port = atoi(optarg);55                 break;56             case 'h':57                 print_usage(progname);58                 return EXIT_SUCCESS;59             default:60                 break;61         }
62     }63     if( !ser_port )64     {65         print_usage(progname);66         return -1;67     }68 69     if((listenfd = socket_server_init(NULL, ser_port)) < 0)//初始化socket70     {71         printf("ERROR: %s server listen on port %d failure\n",argv[0],ser_port);72         return -2;73     }74     printf("%s server start to listen on port %d \n",argv[0],ser_port);75 76     if(daemon_run)77     {78         daemon(0, 0);79     }80 81     for(i=0; imaxfd ? fds_array[i] : maxfd;95                 FD_SET(fds_array[i],&rdset);//将数组中的文件描述符加入rdset集合96         }97 98     rv =select(maxfd+1, &rdset, NULL, NULL, NULL);// select监视并等待多个文件描述符的属性发生变化，99     if( rv < 0 )
100     {
101         printf("select failure:%s\n",strerror(errno));
102                 break;
103     }
104     else if( 0 == rv )
105     {
106         printf("select get timeout\n");
107         continue;
108     }
109 
110     if(FD_ISSET(listenfd,&rdset))/*判断监听描述符是否在集合中，且是否有触发，如果在且收到客户端的请求则accept客户端的连接请求，且在数组仍有容量时将accept的fd加入数组中，若数组已经没有容量，则不接受该客户端的连接，并close该fd，如果不在则将该文件描述符且数组仍有容量加入数组中，但如果数组中已有的文件描述符有事件发生，则收发数据*/
111     {
112         if((connfd = accept(listenfd,(struct sockaddr *)NULL,NULL)) < 0)
113         {
114             printf("accept new client failure:%s\n",strerror(errno));
115             continue;
116         }
117         found = 0;
118         for(i=0; i

1 #include2 #include3 #include4 #include5 #include6 #include 7 #include 8 #include 9 #include 10 #include 11 #include 12 #include 13 #include 14 #include 15 16 17 #define  MAX_EVENTS       51218 #define  ADDAY_SIZE(x)   (sizeof(x)/sizeof(x[0]))19 20 int socket_server_init(char *listen_ip, int listen_port);21 static inline void print_usage(char *progname);22 void set_socket_rlimit(void);23 24 int main(int argc , char **argv)25 {26     int                  listenfd,connfd;27     int                  ser_port = 0;28     int                  daemon_run = 0;29     char                 *progname = NULL;30     int                  opt;31     int                  rv;32     int                  i, j;33     int                  found;34     char                 buf[1024];35     int                  epollfd;36     struct epoll_event   event;37     struct epoll_event   event_array[MAX_EVENTS];38     int                  events;39     struct option       opts[] = {40         {"daemon", no_argument, NULL, 'b'},41         {"port", required_argument, NULL, 'p'},42         {"help", no_argument, NULL, 'h'},43         {NULL, 0, NULL, 0}44     };45 46          progname = basename(argv[0]);47 48     while((opt = getopt_long(argc, argv, "bp:h", opts, NULL)) != -1 )49     {50         switch( opt )51         {52             case 'b':53                 daemon_run = 1;54                 break;55             case 'p':56                 ser_port = atoi( optarg );57                 break;58             case 'h':59                 print_usage(progname);60                 return EXIT_SUCCESS;61             default:
62                 break;63         }64     }65     if( !ser_port )66     {67         print_usage(progname);68         return -5;69     }70 71     set_socket_rlimit();//修改socket监听的最大文件描述符的数量72 73     if((listenfd = socket_server_init(NULL, ser_port)) < 0)74     {75         printf("ERROR:%s server listen on port %d failure\n",argv[0],ser_port);76         return -6;77     }78     printf("%s server start to listen on port %d.\n",argv[0],ser_port);79 80     if( daemon_run )//将程序放在后台运行81     {82         daemon(0,0);83     }84 85     if((epollfd = epoll_create(MAX_EVENTS)) < 0)//)创建了一个新的epoll实例86     {87         printf("epoll_create() failure:%s\n",strerror(errno));88         return -7;89     }90     event.events = EPOLLIN;91     event.data.fd = listenfd;//将listenfd加入结构体体epoll_event92     if((epoll_ctl(epollfd, EPOLL_CTL_ADD,listenfd,&event)) < 0)//修改由文件描述符epfd所代表的epoll实例中的兴趣列表，即将listenfd添加到epoll实例中的兴趣列表中去93     {94         printf("epoll add listen socket failure:%s\n",strerror(errno));95         return -8;96     }97     for( ; ; )98     {99         events = epoll_wait(epollfd, event_array, MAX_EVENTS, -1);//等待事件发生
100         if( events < 0 )
101         {
102             printf("epoll failure:%s\n",strerror(errno));
103             break;
104         }
105         else if( 0 == events)
106         {
107             printf("epoll get timeout\n");
108             continue;
109         }
110         for(i=0; i0，事件响应*/
111         {
112             if((event_array[i].events & EPOLLERR) || (event_array[i].events & EPOLLHUP))
113             {
114                 printf("epoll_wait get error on fd[%d]:%s\n",event_array[i].data.fd,strerror(errno));
115                 epoll_ctl(epollfd, EPOLL_CTL_DEL, event_array[i].data.fd, NULL);
116                 close(event_array[i].data.fd);
117             }
118             if(event_array[i].data.fd == listenfd)/*监听到套接字get事件意味着新客户端现在开始连接*/
119             {
120                 if((connfd = accept(listenfd, (struct sockaddr *)NULL, NULL)) <0 )
121                 {
122                     printf("accept new client failurn :%s\n",strerror(errno));
123                     continue;
124                 }
125                 event.data.fd = connfd;
126                 event.events = EPOLLIN;
127                 if(epoll_ctl(epollfd, EPOLL_CTL_ADD, connfd, &event) < 0 )
128                 {
129                     printf("epoll add client socket failure:%s\n",strerror(errno));
130                     close(event_array[i].data.fd);
131                     continue;
132                 }
133                 printf("epoll add new client socket[%d] ok\n",connfd);
134             }
135             else/*已连接到客户端，读取客户端写入的数据*/
136             {
137                 if( (rv = read(event_array[i].data.fd, buf,sizeof(buf))) <= 0)
138                 {
139                     printf("socket[%d] read failure or get disconnect and will be removed.\n",event_array[i].data.fd);
140                     epoll_ctl(epollfd, EPOLL_CTL_DEL, event_array[i].data.fd, NULL);
141                     close(event_array[i].data.fd);
142                     continue;
143                 }
144                 else/*将数据进行处理，并写回客户端*/
145                 {
146                     printf("socket[%d] read get %d bytes data\n",event_array[i].data.fd, rv);
147                     for(j=0; j

本文来自互联网用户投稿，文章观点仅代表作者本人，不代表本站立场，不承担相关法律责任。如若转载，请注明出处。 如若内容造成侵权/违法违规/事实不符，请点击【内容举报】进行投诉反馈！