Linux多路IO复用技术——epoll详解与一对多服务器实现

在正式阅读本篇博客之前，建议大家先按顺序把下面这两篇博客看一下，否则直接来看这篇博客的话估计很难搞懂

多路IO复用技术①——select详解&如何使用select模型在本地主机实现简易的一对多服务器多路IO复用技术②——poll详解&如何使用poll模型在本地主机实现简易的一对多服务器

在了解以上两篇博客讲解的内容后，我们正式开始本篇博客的相关内容讲解，第三种多路套接字监听技术——epoll模型

目录

EPOLL模型优化的第一部分

1.对拷贝开销的优化

2.对挂载开销的优化

3.对查找开销的优化

EPOLL模型的相关接口

相关结构体 struct epoll_event

相关函数

EPOLL模型优化的第二部分

4.监听方面的优化

EPOLL模型的优缺点

使用epoll模型实现简易一对多服务器的程序实现

程序构成           结果图示

EPOLL的两种监听模式

epoll模型是最优秀的套接字监听技术，他几乎解决了select和poll模型中的所有缺点，并在其优点上进行了进一步的优化，接下来，我们就来了解一下epoll模型的进行多路套接字监听的原理吧

EPOLL模型优化的第一部分

我们知道，select和poll模型实现的监听原理其实都是按照以下这些步骤：

1. 系统会将这个用户层的监听集合拷贝到内核层

2. 系统会将内核层的该监听集合中监听的套接字放入IO设备等待队列，由IO设备等待队列来进行一次又一次的遍历来判断那些套接字中有数据需要处理

3. 当IO设备等待队列发现有套接字处于就绪状态时，会传出就绪集合到内核层

4. 系统通过select模型将该就绪集合由内核层拷贝到用户层，供用户使用

想要了解下面的内容，这里需要为大家简单讲解一种数据结构——红黑树  ，并希望大家能够记住红黑树的优点

红黑树的优势就在于非常适合用来存储有序的数据，并且增删查改的时间复杂度很低，只有O(log2n)

1.对拷贝开销的优化

在之前的博客中，我们讲过，随着select/poll模型的持续使用，会产生大量的拷贝开销和挂载开销

这些拷贝开销大致由两部分组成：

1. 系统从用户层到内核层，内核层到用户层翻来覆去的层级转换

2. 一旦监听集合更新，就要重新拷贝整个就绪集合，只要服务器不停，就会面临无止尽的拷贝

该如何减少第一部分的拷贝开销呢？

其实很简单粗暴，epoll模型的开发人员直接将监听集合定义在内核层。所以epoll模型的监听集合都是建立在内核层的

PS：这种监听集合实际上是红黑树，每一个叶子节点对应一个监听套接字

至于如何减少第二部分的拷贝开销，就和红黑树树这种数据结构息息相关了。

我们知道，select和poll模型都是采用数组作为作为监听集合，数组这种结构非常不便于查找其中哪些数据产生了变化，就导致当有新的套接字要放入监听集合中时，系统会将整个新的监听集合全部拷贝到内核层。

而红黑树就完美的避开了这一点，每当有新的套接字要放入监听树中时，我们只需要为这棵监听红黑树添加一个叶子节点就够了，而不是重建这棵树，这就导致了这方面的开销大大减少了

2.对挂载开销的优化

在对拷贝开销的优化方面，epoll模型的开发人员依旧采用了红黑树这种数据结构，我们来为大家简单的介绍一下原因：

我们知道，在select与poll模型中，数组这种结构非常不便于查找其中哪些数据产生了变化，所以每当有新的套接字放入监听集合中时，系统会将监听的套接字全部重新挂载到IO设备等待队列，这就导致挂载开销方面的开销非常大。

而红黑树则完美的解决了这个问题，由于红黑树的特性，每当有对应新套接字的叶子节点被放入监听树中时，系统能够很轻松的就能找到对应新套接字的叶子节点是哪一个，然后只需要将对应的新套接字放入IO设备等待队列就可以了，在挂载开销方面的消耗大大减少

3.对查找开销的优化

我们知道，虽然数组的下标是有序的，方便我们进行遍历，但是数组中的内容并不一定是有序的，这就导致我们无法根据我们目标内容来直接找到其在数组中的哪个位置，而是需要去通过循环一次次的查找，但红黑树就完美地解决了这个问题

我们知道，当我们调用accept函数让服务器与多个客户端建立TCP链接时，其返回的int类型的套接字文件描述符，其实都是有序递增的，建立一个链接返回值就+1。由于红黑树在有序数据存储方面的优势，使得服务器在对于监听就绪和取消监听方面的效率非常之高。

其实epoll模型不止在这些方面进行了优化，其他部分需要大家了解了EPOLL模型的相关接口后，才能为大家进行讲解，所以别着急，我们先来了解下EPOLL模型的相关接口吧

EPOLL模型的相关接口

以下接口的头文件都是 #include <sys/epoll.h>

相关结构体 struct epoll_event

struct epoll_event 
{
    uint32_t events;  // epoll 事件类型，包括可读，可写等
    epoll_data_t data; // 用户数据，可以是一个指针或文件描述符等
    callback();  //这个函数与用户没有关系，我们是看不见的，只有当发现对应套接字就绪时才会调用该函数进行回调操作
};

typedef union epoll_data 
{
    void *ptr;   //指向任何类型的用户数据
    int fd;   //套接字文件描述符
    uint32_t u32;   //32位的无符号整数
    uint64_t u64;   //64位的无符号整数
} epoll_data_t;

其中，events字段表示要监听的事件类型，可以是以下值之一：

• EPOLLIN：表示对应的文件描述符上有数据可读

• EPOLLOUT：表示对应的文件描述符上可以写入数据

• EPOLLRDHUP：表示对端已经关闭连接，或者关闭了写操作端的写入

• EPOLLPRI：表示有紧急数据可读

• EPOLLERR：表示发生错误

• EPOLLHUP：表示文件描述符被挂起

• EPOLLET：表示将epoll设置为边缘触发模式

• EPOLLONESHOT：表示将事件设置为一次性事件

这个回调的原理是什么呢？其实很简单

其实负责监听这些套接字事件的设备就是网卡，所以最先知道套接字就绪的就是网卡设备，这时候开发人员就想了，那我能不能实现一个功能，创建一个结构体队列，直接将该队列与网卡进行绑定，形成一种回调关系。这样的话，当网卡监听到某套接字处于就绪状态时，就通过异步通知的方式来告诉服务器是哪个套接字就绪了，不需要再一次次的遍历来判断哪个套接字就绪

所以epoll模型监听套接字的过程就如下所示：

1. 建立结构体(ep_item)队列，队列中的每一个结构体和一个套接字对应，将该队列与网卡进行绑定

2. 建立一个就绪链表，存放对应就绪套接字的结构体epoll_events

3. 监听到某一套接字就绪时，调用结构体epoll_events中的callback函数，进行拆包，将结构体ep_item中的结构体epoll_events拆分出来，并弹出到就绪链表中（双向链表）

4. 将这个就绪链表中的内容拷贝到用户在用户层定义的结构体(epoll_event)数组，以便用户使用

不理解的话可以看下面的这个图

相关函数

先来介绍一下一会会用到的参数：

• int epfd; // 返回的监听红黑树文件描述符

• int epoll_max; //监听套接字的最大数量

• int option; //需要执行的操作——添加、修改、删除

• int sockfd; //需要添加，修改，删除的socket文件描述符

• struct epoll_event * node; //挂在红黑树上的节点

• int array_size; //存放对应就绪套接字的结构体的数组大小

• struct epoll_event ready_array[array_size]; //存放对应就绪套接字的结构体的数组

• int ready_max; //允许同时监听的最大套接字数量

• int timeout; //工作模式

PS：一般情况下，epoll_max = ready_max = array_size

我们来分别介绍一下option和timeout参数的有效值：

option参数的有效值为：

• EPOLL_CTL_ADD：添加与目标套接字对应的结构体(epoll_event)到epoll监听树上，开始监听该套接字

• EPOLL_CTL_MOD：修改目标套接字的监听事件

• EPOLL_CTL_DEL：将与目标套接字对应的结构体(epoll_event)从epoll监听树上摘除，也就是不再监听该套接字

timeout参数的有效值：

• timeout = 0：表示非阻塞监听套接字相关事件

• timeout = -1：表示阻塞监听套接字相关事件

• timeout > 0：表示阻塞监听时间为timout(单位：毫秒)，如果这段时间内没有监听到套接字相关事件，后续变为非阻塞监听

epoll_create()的错误码如下所示：

• EINVAL：epoll_max大小不为正。

• EMFILE：遇到了每个用户对/ proc / sys / fs / epoll / max_user_instances施加的epoll实例数量的限制。

• ENFILE：超过系统设定的进程最大创建的文件描述符数量。

• ENOMEM：没有足够的内存来创建内核对象。

epoll_ctl()的错误码如下所示：

• EBADF：epfd或fd不是有效的文件描述符。

• EEXIST：option为EPOLL_CTL_ADD，并且提供的文件描述符fd已在该epoll实例中注册。

• EINVAL：epfd不是epoll文件描述符，或者fd与epfd相同，或者此接口不支持请求的操作option。

• ENOENT：option是EPOLL_CTL_MOD或EPOLL_CTL_DEL，并且fd未在该epoll实例中注册。

• ENOMEM：没有足够的内存来处理请求的操作控制操作。

• ENOSPC：尝试在主机上注册（EPOLL_CTL_ADD）新文件描述符时遇到了/ proc / sys / fs / epoll / max_user_watches施加的限制。

• EPERM：目标文件fd不支持epoll。

epoll_wait()的错误码如下所示：

• EBADF：epfd不是有效的文件描述符。

• EFAULT：具有写许可权不能访问事件指向的存储区。

• EINTR：在任何请求的事件发生或超时到期之前，信号处理程序中断了该调用；参见signal（7）。

• EINVAL：epfd不是epoll文件描述符，或者epoll_max小于或等于零。

EPOLL模型优化的第二部分

4.监听方面的优化

由于epoll模型的自定义结构体队列、异步通知和回调函数的使用，导致用户不再需要遍历去查找就绪的套接字是哪些，直接对自己在用户层自定义的结构体数组进行遍历处理就可以了，大大减少了时间片方面的消耗，使得服务器的处理能力实现质的飞跃

基本了解了epoll在哪些方面做出了优化后，我们就可以来聊一聊epoll模型的优缺点了

EPOLL模型的优缺点

优点：

1. 不存在重复的拷贝开销与挂载开销

2. 自定义实现监听队列，采用异步回调方式，无需轮询，队列体积更小

3. 不止返回就绪的套接字数量，还返回就绪的套接字是哪些

4. 监听的事件种类丰富

5. 可以为不同的套接字设置不同的监听事件，不像select模型只能批量设置监听事件

6. 可以监听的socket数量不受1024的硬限制

缺点：

1. 仅linux系统支持

2. 红黑树的缺点

使用epoll模型实现简易一对多服务器的程序实现

程序构成

该服务器与客户端由以下几个程序共同组成：

• func_2th_parcel.h：定义二次包裹的函数名

• func_2th_parcel.c：对网络初始化相关的函数进行二次包裹

• epoll_server.c：使用poll模型的服务器程序

• client.c：客户端程序

/*************************************************************************
        > File Name: func_2th_parcel.h
        > Author: Nan
        > Mail: **@qq.com
        > Created Time: 2023年10月18日 星期三 18时32分22秒
 ************************************************************************/
 
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <sys/stat.h>
#include <signal.h>
#include <pthread.h>
#include <fcntl.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <sys/mman.h>
#include <time.h>
#include <ctype.h>
#include <sys/select.h>
#include <sys/socket.h>
#include <poll.h>
#include <sys/epoll.h>
 
//socket函数的二次包裹
int SOCKET(int domain , int type , int protocol);
 
//bind函数的二次包裹
int BIND(int sockfd , const struct sockaddr* addr , socklen_t  addrlen);
 
//listen函数的二次包裹
int LISTEN(int sockfd , int backlog);
 
//send函数的二次包裹
ssize_t SEND(int sockfd , const void* buf , size_t len , int flags);
 
//recv函数的二次包裹
ssize_t RECV(int sockfd , void* buf , size_t len , int flags);
 
//connect函数的二次包裹
int CONNECT(int sockfd , const struct sockaddr* addr , socklen_t addrlen);
 
//accept函数的二次包裹
int ACCEPT(int sockfd , struct sockaddr* addr , socklen_t addrlen);
 
//网络初始化函数
int SOCKET_NET_CREATE(const char* ip , int port);
 
//服务端与客户端建立连接并返回客户端套接字文件描述符
int SERVER_ACCEPTING(int server_fd);

/*************************************************************************
        > File Name: func_2th_parcel.c
        > Author: Nan
        > Mail: **@qq.com
        > Created Time: 2023年10月18日 星期三 18时32分42秒
 ************************************************************************/
 
#include <func_2th_parcel.h>
 
int SOCKET(int domain , int type , int protocol){
    int return_value;
    if((return_value = socket(domain , type , protocol)) == -1){
        perror("socket call failed!\n");
        return return_value;
    }
    return return_value;
}
 
int BIND(int sockfd , const struct sockaddr* addr , socklen_t addrlen){
    int return_value;   
    if((return_value = bind(sockfd , addr , addrlen)) == -1){
        perror("bind call failed!\n");
        return return_value;
    }                      
    return return_value;   
}
 
int LISTEN(int sockfd , int backlog){
    int return_value;   
    if((return_value = listen(sockfd , backlog)) == -1){
        perror("listen call failed!\n");
        return return_value;
    }                      
    return return_value;   
}
 
ssize_t SEND(int sockfd , const void* buf , size_t len , int flags){
    ssize_t return_value;
    if((return_value = send(sockfd , buf , len , flags)) == -1){
        perror("send call failed!\n");
        return return_value;
    }
    return return_value;
}
 
ssize_t RECV(int sockfd , void* buf , size_t len , int flags){
    ssize_t return_value;   
    if((return_value = recv(sockfd , buf , len , flags)) == -1){
        perror("recv call failed!\n");
        return return_value;
    }                      
    return return_value;   
}
 
int CONNECT(int sockfd , const struct sockaddr* addr , socklen_t addrlen){
    int return_value;   
    if((return_value = connect(sockfd , addr , addrlen)) == -1){
        perror("connect call failed!\n");
        return return_value;
    }                      
    return return_value;   
}
 
int ACCEPT(int sockfd , struct sockaddr* addr , socklen_t addrlen){
    int return_value;   
    if((return_value = accept(sockfd , addr , &addrlen)) == -1){
        perror("accept call failed!\n");
        return return_value;
    }                      
    return return_value;   
}
 
int SOCKET_NET_CREATE(const char* ip , int port){
    int sockfd;
    struct sockaddr_in addr;
    addr.sin_family = AF_INET;
    addr.sin_port = htons(port);
    inet_pton(AF_INET , ip , &addr.sin_addr.s_addr);
    sockfd = SOCKET(AF_INET , SOCK_STREAM , 0);
    BIND(sockfd , (struct sockaddr*)&addr , sizeof(addr));
    LISTEN(sockfd , 128);
    return sockfd;
}
 
int SERVER_ACCEPTING(int server_fd)
{
    int client_sockfd;
    struct sockaddr_in client_addr;
    char client_ip[16];
    char buffer[1500];
    bzero(buffer , sizeof(buffer));
    bzero(&client_addr , sizeof(client_addr));
    socklen_t addrlen = sizeof(client_addr);
    client_sockfd = ACCEPT(server_fd , (struct sockaddr*)&client_addr , addrlen);
    bzero(client_ip , 16);
    //将客户端的IP地址转成CPU可以识别的序列并存储到client_ip数组中
    inet_ntop(AF_INET , &client_addr.sin_addr.s_addr , client_ip , 16);
    sprintf(buffer , "Hi , %s welcome tcp test server service..\n" , client_ip);
    printf("client %s , %d , connection success , client sockfd is %d\n" , client_ip , ntohs(client_addr.sin_port) , client_sockfd);
    SEND(client_sockfd , buffer , strlen(buffer) , MSG_NOSIGNAL);
    return client_sockfd;
}

/*************************************************************************
        > File Name: epoll_server.c
        > Author: Nan
        > Mail: **@qq.com
        > Created Time: 2023年10月25日 星期三 18时53分30秒
 ************************************************************************/

#include <func_2th_parcel.h>

#define MAX_EPOLL 100000

int main(void)
{
    int server_sockfd;//服务器套接字文件描述符
    int epfd;//监听树文件描述符
    struct epoll_event ready_array[MAX_EPOLL];//由于存放epoll返回的就绪结构体
    struct epoll_event node;//向监听树中放入的节点
    int ready_num;//处于就绪状态的套接字的数量
    int client_sockfd;//客户端套接字文件描述符
    char rw_buffer[1500];//读写缓冲区
    int flag;
    int recv_len = 0;//接受到的数据的长度
    int i;
    bzero(rw_buffer , sizeof(rw_buffer));//清空读写缓冲区
    server_sockfd = SOCKET_NET_CREATE("192.168.79.128" , 6060);//进行网络初始化
    //初始化epoll结构体
    node.data.fd = server_sockfd;
    node.events = EPOLLIN;//监听读事件
    //初始化监听树，并做错误处理
    if((epfd = epoll_create(MAX_EPOLL)) == -1)
    {
    	perror("epoll_create call failed\n");
    	exit(-1);
    }
    //向监听树中添加服务器套接字结构体，并做错误处理
    if((epoll_ctl(epfd , EPOLL_CTL_ADD , server_sockfd , &node)) == -1)
    {
    	perror("epoll_ctl call failed\n");
    	exit(-1);
    }
    printf("epoll_server wait TCP connect\n");
    while(1)
    {
    	//获取处于就绪状态的套接字数量
        if((ready_num = epoll_wait(epfd , ready_array , MAX_EPOLL , -1)) == -1)
        {
            perror("epoll_wait call failed\n");
            exit(0);
        }
        i = 0;
        while(ready_num)
        {
            //辨别就绪，如果是服务端套接字就绪
            if(ready_array[i].data.fd == server_sockfd)
            {
                client_sockfd = SERVER_ACCEPTING(ready_array[i].data.fd);//与客户端建立TCP链接
                node.data.fd = client_sockfd;
                if(epoll_ctl(epfd , EPOLL_CTL_ADD , client_sockfd , &node) == -1)
                {
                	perror("epoll_ctl failed\n");
                	exit(-1);
                }
            }
            //如果是客户端套接字就绪
            else
            {  
                recv_len = RECV(ready_array[i].data.fd , rw_buffer , sizeof(rw_buffer) , 0);
                flag = 0;
                //如果recv_len = 0,就说明与客户端套接字对应的客户端退出了，将对应客户端套接字移出套接字存储数组与监听集合
                if(recv_len == 0)
                {
                    perror("某一客户端与本服务器断开链接\n");
                    printf("客户端%d 与本服务器断开链接，关闭其对应的套接字并停止监听\n" , ready_array[i].data.fd);
                    //关闭该套接字
                    close(ready_array[i].data.fd);
                    //将其对应的节点从监听树上摘下来
                    epoll_ctl(epfd , EPOLL_CTL_DEL , ready_array[i].data.fd , NULL);
                    //注意，这个两个步骤不能搞反
                }
                //进行业务处理：小写字母转大写字母
                printf("服务器已接收到客户端%d 发来的信息 : %s，现在对其进行处理\n" , ready_array[i].data.fd , rw_buffer);
                while(recv_len > flag)
                {
                    rw_buffer[flag] = toupper(rw_buffer[flag]);
                    flag++;
                }
                printf("服务器已对客户端%d 发来的信息完成处理，处理后的数据为 : %s\n" , ready_array[i].data.fd , rw_buffer);
                SEND(ready_array[i].data.fd , rw_buffer , recv_len , MSG_NOSIGNAL);
                bzero(rw_buffer , sizeof(rw_buffer));
                recv_len = 0;
            }
            ready_num--;
            i++;
        }
    }
    close(server_sockfd);
    printf("server shutdown\n");
    return 0;
}

/*************************************************************************
        > File Name: client.c
        > Author: Nan
        > Mail: **@qq.com
        > Created Time: 2023年10月19日 星期四 18时29分12秒
 ************************************************************************/
 
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <sys/stat.h>
#include <signal.h>
#include <pthread.h>
#include <fcntl.h>
#include <sys/socket.h>
#include <netdb.h>
#include <arpa/inet.h>
#include <time.h>
 
//服务器实现大小写转换业务
 
int main()
{
    //1.定义网络信息结构体与读写缓冲区并初始化
    struct sockaddr_in dest_addr;
    char buffer[1500];
    bzero(&dest_addr , sizeof(dest_addr));
    bzero(buffer , sizeof(buffer));
    dest_addr.sin_family = AF_INET;
    dest_addr.sin_port = htons(6060);
    //字符串ip转大端序列
    inet_pton(AF_INET , "192.168.79.128" , &dest_addr.sin_addr.s_addr);
    int sockfd = socket(AF_INET , SOCK_STREAM , 0);
    int i;
    //2.判断连接是否成功
    if((connect(sockfd , (struct sockaddr*) &dest_addr , sizeof(dest_addr))) == -1)
    {
        perror("connect failed!\n");
        exit(0);
    }
    recv(sockfd , buffer , sizeof(buffer) , 0);
    printf("%s" , buffer);
    bzero(buffer , sizeof(buffer));
    //3.循环读取终端输入的数据
    while( (fgets(buffer , sizeof(buffer) , stdin) ) != NULL)
    {
        i = strlen(buffer);
        buffer[i-1] = '\0';
        //向服务端发送消息
        send(sockfd , buffer , strlen(buffer) , MSG_NOSIGNAL);
        //接收服务端发来的消息
        recv(sockfd , buffer , sizeof(buffer) , 0);
        //打印服务端发来的信息
        printf("response : %s\n" , buffer);
        //清空读写缓冲区，以便下一次放入数据
        bzero(buffer , sizeof(buffer));
    }
    //4.关闭套接字，断开连接
    close(sockfd);
    return 0;
}

结果图示

EPOLL的两种监听模式

1.EPOLLLT，水平触发模式，也被称之为负责模式，如果不进行特殊设置，默认情况下对套接字都是以此模式监听。

在此模式下，当epoll监听到某一套接字就绪时，就会先发送处理通知给上层。之后每隔一段时间就来查看该套接字缓冲区中的数据有没有被处理完。

如果没有处理完，就会继续发送处理通知给上层，一天不处理完就发一天，一周处理不完就发一周，直到缓冲区中的数据被全部处理完毕。这就是所说的负责模式

PS：在此模式下，如果缓冲区中的数据没有被处理完，epoll_wait()无法开启下一轮监听。意思就是如果数据没处理完，就去调用epoll_wait()函数的话，会立即返回，返回值为未处理的就绪事件数量

优点：能够保证套接字缓冲区中的所有数据被处理完毕

缺点：开销比较大

适用场景：该模式适合处理有效期较短、或需要被紧急处理的数据

2.EPOLLET，边缘触发模式，也被称之为不负责模式，使用该模式需要进行手动设置

在此模式下，当epoll监听到某一套接字就绪时，指挥发送一次处理通知给上层。之后无论该套接字缓冲区中的数据有没有被处理完，都与epoll无关，epoll可以立即进入新一轮的监听

优点：开销比较小

缺点：存在隐患，不能够保证套接字缓冲区中的所有数据被处理(完毕)，用户要自行保证数据读取完毕

PS：边缘触发模式一般结合非阻塞读取套接字缓冲区的数据

想要设置边缘触发模式也很简单，按照下面的步骤来就可以了

int sockfd;//套接字文件描述符

struct epoll_event node;
node.data.fd = sockfd;
node.events = EPOLLIN|EPOLLET;//监听读事件，监听模式切换为边缘触发模式

以上就是本篇博客的全部内容了，大家有什么地方没有看懂的话，可以在评论区留言给我，咱要力所能及的话就帮大家解答解答

今天的学习记录到此结束啦，咱们下篇文章见，ByeBye！

文章来源地址https://www.toymoban.com/news/detail-744045.html

到了这里，关于Linux多路IO复用技术——epoll详解与一对多服务器实现的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！