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

【高级IO】多路转接之Epoll

news 来源：原创 2025/5/11 2:14:49

多路转接之Epoll

一.Epoll实现原理
二.Epoll接口
三.封装Epoller
四.实现Epoll服务器
- 1.初始化
- 2.启动EpollServer
- 3.派发就绪事件
- 4. listen套接字就绪
- 5.普通套接字就绪
五.代码
六.Epoll优点

一.Epoll实现原理

Epoll模型由三部分构成，一个是红黑树，一个是就绪队列，一个是回调方法。
回调方法是一旦有事件就绪了，就会回调该方法，检测红黑树中是否有对应的文件描述符和事件，如果有就将就绪事件放入到就绪队列中。
红黑树存放的是用户关心的文件描述符及其事件。
就绪队列中存放的是已经就绪的事件。

在这里插入图片描述

epoll在内核中就是一个数据结构，把它也当做文件因为我们的linux当中一切文件。所以呢，epoll返回值也是个文件描述符，最终epoll模型也统一被接入到了struct file文件里。

这个回调函数callback是干什么的呢?
[问题]操作系统在硬件层面，怎么知道网卡上有数据了呢?—>硬件中断
比如说，一旦数据链路层底层有数据就绪了那么数据链路层就回发生中断，自动回调用我们对应的callback
1.callback中会将数据向上交付，交付到tcp的接收缓队列中。并且会查找红黑树中的结点。
2.查找红黑树，是为了确认这个的接收队列和哪一个我们对应的文件描述符是关联的并且看看这个fd是否关心什么事件
如果确认这个接收队列是和fd是关联的，并且有关心的事件，就插入到就绪队列中

如果底层有事件一旦就绪，它就会自动执行上面的callback方法，使用epoll的时候，操作系统它就会把这个回调函数注册到底层。
【原理】
所以最终一旦我们有底层有数据就绪，从硬件中断，再到交给操作系统再到接收队列再到检测它在红黑树中，就会将所有就绪的事件放入就绪对列当中。
所以未来用户，他只需要从就绪队列当中，拿已经就绪的事件就可以了。

二.Epoll接口

#include <sys/epoll.h>
int epoll_create(int size);int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);

1.int epoll_create(int size);
作用：epoll_create，它就是创建一个epoll模型，返回值是对应的文件描述符。也就是把空的红黑树建立起来空的就绪队列建立起来，回调机制callback在底层建立好。

2.int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
作用：epoll_crt它就是在对红黑树进行修改，修改要关心的文件描述符或事件。
参数1：epfd就是要修改的epoll模型是哪个，epoll_create创建时会返回。
参数2：op就是要对红黑树执行什么操作。有三种方法：ADD，MOD，DEL。
参数3，4：fd，event就是要具体关心哪个文件描述符的什么事件。要关心什么就插入到红黑树中。

3.int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);
作用：epoll_wait它本质就是在获取就绪队列中的就绪事件。一旦有事件就绪了，那么就会返回，它的返回值表示有几个fd就绪了。后面就不再需要检测过虑一下哪些是没有就绪的，直接遍历n次对应的events数组，里面全是就绪事件。

参数1：epfd，要等待哪个epoll模型
参数2：struct epoll_event*events，events是一个struct epoll_event类型的数组，每个成员代表的是就绪的事件是什么，哪个文件描述符就绪了。

events可以是以下几个宏的集合：
EPOLLIN : 表示对应的文件描述符可以读 (包括对端SOCKET正常关闭);
EPOLLOUT : 表示对应的文件描述符可以写;
EPOLLPRI : 表示对应的文件描述符有紧急的数据可读 (这里应该表示有带外数据到来);
EPOLLERR : 表示对应的文件描述符发生错误;
EPOLLHUP : 表示对应的文件描述符被挂断;
EPOLLET : 将EPOLL设为边缘触发(Edge Triggered)模式, 这是相对于水平触发(Level Triggered)来说的.
EPOLLONESHOT：只监听一次事件, 当监听完这次事件之后, 如果还需要继续监听这个socket的话, 需要
再次把这个socket加入到EPOLL队列里.

三.封装Epoller

首先是构建出Epoll模型对象。
然后构建等待模块：需要外面传入一个输出型参数struct epoll_event revents数组，和一个num大小，revents数组就用来将已经就绪的事件捞出来。num是表示一次性捞出来多少个就绪事件。
最后构建修改模块：需要外面传入执行的方法op，关心什么套接字fd，关心什么事件event，三个参数。

细节：
epoll_control尼，它的作用呢，是向指定的ep模型当中进行添加修改或者，系统调用中epoll_control要传入的是struct_epoll对象，所以需要构建一个struct_epoll对象，将外部传入的事件event设置到struct_epoll对象中，注意这里还需要将外传的fd，也设置到struct_epoll对象中。这样做的目的是，后续用户从wait中读取到struct_epoll对象，就可以从里面知道是哪个文件描述符就绪了。

epoll_ctrl呢，本质上其实是通过系统调用然后向我们红黑树志当中新增修改节点
红黑树节点呢和我们的底层回调产生勾连所以最后呢?当一旦就绪时
它就会在我们的底层通知我然后放到就绪队列中。

#include "Log.hpp"
#include <sys/epoll.h>
#include <errno.h>
#include <cstring>
class Epoller
{static const int size = 128;public:Epoller(){// epoll对象创建成功就表明epoll模型创建_epfd = epoll_create(size);if (_epfd == -1) // 表示创建失败{lg(Error, "epoll create error,%s", strerror(errno));}else // epoll模型创建成功{lg(Info, "epoll create success, _epfd=%d", _epfd);}}int EpollerWait(struct epoll_event revents[], int num) // revents里面能获取到关心的所有已就绪的事件{                                                      // num一次表示可获取的最大数量int n = epoll_wait(_epfd, revents, num, _timeout);return n; // 返回值表示有多少个已经就绪的事件}int EpollerUpdate(int op, int sock, uint32_t event) // 具体要对哪个文件描述符的什么事件进行什么操作(添加到红黑树中，从红黑中删除更新等){// event只是用户传进来要关心的事件，是读还写// 而传入到内核红黑树中，要使用struct epoll_eventint n = 0;if (op == EPOLL_CTL_DEL){n = epoll_ctl(_epfd, op, sock, nullptr); // 可以将空传入到内核里表示对该文件描述符不关心了if (n != 0){lg(Error, "epoll_crt delete error");}}else // EPOLL_CTL_MOD || EPOLL_CTL_ADD{// 要对哪个文件描述符关心什么事件呢？struct epoll_event ev;ev.events = event;ev.data.fd = sock; // 这里后面可以知道是什么fd的事件就绪了n = epoll_ctl(_epfd, op, sock, &ev);if (n != 0){lg(Error, "epoll_crt error");}}return n;}~Epoller(){if(_epfd>=0)close(_epfd);}private:int _epfd; // epoll模型的文件描述符int _timeout{3000};
};

四.实现Epoll服务器

1.初始化

初始化时，创建套接字，绑定端口，设置监听。

    void Init(){_listsocket_ptr->Socket();_listsocket_ptr->Bind(_port);_listsocket_ptr->Listen();lg(Info, "create listen socket success: %d\n", _listsocket_ptr->Fd());}

2.启动EpollServer

EpollServer服务器启动时，注意不能直接accept获取，而是应该先等待连接事件就绪，一旦有事件就绪，就将就绪的事件派发下去执行。
(要注意一开始listen套接字我们还没有设置到内核关心，所以在等待之前，首先需要将listen套接字添加到内核的红黑树上，并且关心读事件)

一旦有事件就绪，那么 int n = _epoller_ptr->EpollerWait(revs, num);就会返回，返回值是就绪事件的个数。所有已经就绪的事件都在revs这个输出型参数里。
将revs和n传给 Dispatcher(revs, n);，让 Dispatcher派发执行就绪事件

void Start(){// 将listensock添加到epoll中 -> listensock和他关心的事件，添加到内核epoll模型中rb_tree._epoller_ptr->EpollerUpdate(EPOLL_CTL_ADD, _listsocket_ptr->Fd(), EVENT_IN);struct epoll_event revs[num];for (;;){int n = _epoller_ptr->EpollerWait(revs, num);if (n > 0){// 有事件就绪lg(Debug, "event happened, fd is : %d", revs[0].data.fd);Dispatcher(revs, n);}else if (n == 0){lg(Info, "time out ...");}else{lg(Error, "epll wait error");}}}

3.派发就绪事件

所有就绪的事件都在struct epoll_event revs[]这个数组中，并且个数为num个。
所以只需要循环num次就可以将本次获取的所有就绪事件执行完。

1.首先根据revs找到是哪个文件描述符就绪，就绪的是什么事件。

2.根据不同的就绪事件，分别对应不同的方法处理。如果就绪的是读事件，就执行读处理，如果是写事件就执行读处理…。
而读事件又分为读取连接和读取数据两种。可以根据就绪的套接字来判断：如果是listnen套接字就绪，则一定是连接事件就绪，那么就执行Accepter函数，如果是普通套接字，则一定是读取数据事件就绪，那么就执行Recver函数。

 void Dispatcher(struct epoll_event revs[], int num){for (int i = 0; i < num; i++){uint32_t events = revs[i].events;int fd = revs[i].data.fd;if (events & EVENT_IN){if (fd == _listsocket_ptr->Fd()){Accepter();}else{// 其他fd上面的普通读取事件就绪Recver(fd);}}else if (events & EVENT_OUT){}else{}}}

4. listen套接字就绪

listen套接字就绪，就代表有客户端发起连接，这时候就可以直接accept，不需要等待。
而如果获取连接成功了，那么就会获得一个新的套接字，我们还需要对这个普通套接字添加到红黑树中，让内核关心这个普通套接字的读/写事件。

 void Accepter(){// 获取了一个新连接std::string clientip;uint16_t clientport;int sock = _listsocket_ptr->Accept(&clientip, &clientport);if (sock > 0){// 我们能直接读取吗？不能_epoller_ptr->EpollerUpdate(EPOLL_CTL_ADD, sock, EVENT_IN);lg(Info, "get a new link, client info@ %s:%d", clientip.c_str(), clientport);}}

所以以后每获取一个新连接，就会将该新连接插入到内核红黑树中，让内核关心它的读/写事件。
那么在下次等待时，一旦读数据事件就绪，就会返回，派发执行该事件。

5.普通套接字就绪

如果是普通套接字就绪，表示的是有客户发送数据过来，对应的套接字的读事件就绪了，就绪后就会去执行Recver函数，从就绪的套接字中读取数据。
如果对端关闭连接，或者连接异常，就需要将该套接字从红黑树中删除掉。设置为不关心。

// for testvoid Recver(int fd){// demochar buffer[1024];ssize_t n = read(fd, buffer, sizeof(buffer) - 1); // bug?if (n > 0){buffer[n] = 0;std::cout << "get a messge: " << buffer << std::endl;// wrirtestd::string echo_str = "server echo $ ";echo_str += buffer;write(fd, echo_str.c_str(), echo_str.size());}else if (n == 0){lg(Info, "client quit, me too, close fd is : %d", fd);//细节3_epoller_ptr->EpollerUpdate(EPOLL_CTL_DEL, fd, 0);close(fd);}else{lg(Warning, "recv error: fd is : %d", fd);_epoller_ptr->EpollerUpdate(EPOLL_CTL_DEL, fd, 0);close(fd);}}

五.代码

#pragma once#include <iostream>
#include <memory>
#include <sys/epoll.h>
#include "Socket.hpp"
#include "Epoller.hpp"
#include "Log.hpp"
#include "nocopy.hpp"uint32_t EVENT_IN = (EPOLLIN);
uint32_t EVENT_OUT = (EPOLLOUT);class EpollServer : public nocopy
{static const int num = 64;public:EpollServer(uint16_t port): _port(port),_listsocket_ptr(new Sock()),_epoller_ptr(new Epoller()){}void Init(){_listsocket_ptr->Socket();_listsocket_ptr->Bind(_port);_listsocket_ptr->Listen();lg(Info, "create listen socket success: %d\n", _listsocket_ptr->Fd());}void Accepter(){// 获取了一个新连接std::string clientip;uint16_t clientport;int sock = _listsocket_ptr->Accept(&clientip, &clientport);if (sock > 0){// 我们能直接读取吗？不能_epoller_ptr->EpollerUpdate(EPOLL_CTL_ADD, sock, EVENT_IN);lg(Info, "get a new link, client info@ %s:%d", clientip.c_str(), clientport);}}// for testvoid Recver(int fd){// demochar buffer[1024];ssize_t n = read(fd, buffer, sizeof(buffer) - 1); // bug?if (n > 0){buffer[n] = 0;std::cout << "get a messge: " << buffer << std::endl;// wrirtestd::string echo_str = "server echo $ ";echo_str += buffer;write(fd, echo_str.c_str(), echo_str.size());}else if (n == 0){lg(Info, "client quit, me too, close fd is : %d", fd);//细节3_epoller_ptr->EpollerUpdate(EPOLL_CTL_DEL, fd, 0);close(fd);}else{lg(Warning, "recv error: fd is : %d", fd);_epoller_ptr->EpollerUpdate(EPOLL_CTL_DEL, fd, 0);close(fd);}}void Dispatcher(struct epoll_event revs[], int num){for (int i = 0; i < num; i++){uint32_t events = revs[i].events;int fd = revs[i].data.fd;if (events & EVENT_IN){if (fd == _listsocket_ptr->Fd()){Accepter();}else{// 其他fd上面的普通读取事件就绪Recver(fd);}}else if (events & EVENT_OUT){}else{}}}void Start(){// 将listensock添加到epoll中 -> listensock和他关心的事件，添加到内核epoll模型中rb_tree._epoller_ptr->EpollerUpdate(EPOLL_CTL_ADD, _listsocket_ptr->Fd(), EVENT_IN);struct epoll_event revs[num];for (;;){int n = _epoller_ptr->EpollerWait(revs, num);if (n > 0){// 有事件就绪lg(Debug, "event happened, fd is : %d", revs[0].data.fd);Dispatcher(revs, n);}else if (n == 0){lg(Info, "time out ...");}else{lg(Error, "epll wait error");}}}~EpollServer(){_listsocket_ptr->Close();}private:std::shared_ptr<Sock> _listsocket_ptr;std::shared_ptr<Epoller> _epoller_ptr;uint16_t _port;
};

#include <iostream>
#include <memory>
#include "EpollServer.hpp"int main()
{std::unique_ptr<EpollServer> svr(new EpollServer(8888));svr->Init();svr->Start();
}