广州做网站推广的公司,网站托管服务 重庆,行业门户网站建设方案,排名软件下载之前的内容我们基本掌握了基础IO#xff0c;如套接字#xff0c;文件描述符#xff0c;重定向#xff0c;缓冲区等知识都是文的基本认识#xff0c;而高级IO则是指更加高效的IO。
对于应用层#xff0c;在读写的时候#xff0c;本质就是把数据写给OS#xff0c;若一方…之前的内容我们基本掌握了基础IO如套接字文件描述符重定向缓冲区等知识都是文的基本认识而高级IO则是指更加高效的IO。
对于应用层在读写的时候本质就是把数据写给OS若一方为空就会进行阻塞式等待读写的本质也就是数据的拷贝。
所以高效的 IO就需要在多线程的情况下单位时间的的拷贝的数据更多等待的比重越小。
目录
五种IO模型
非阻塞式IO
多路复用多路转接
IO多路转接之select
初识select select函数原型
IO多路转接之epoll
1.认识有关epoll的接口
2.epoll原理
3.编写epoll
4.epoll的工作模式
注意
读写的改进 reactor 五种IO模型 当前有五种IO模型
我们以钓鱼佬钓鱼为例鱼竿文件描述符 上钩读写就绪鱼数据
1.钓鱼佬张三在完成打窝之后选择好地点就开始钓鱼在等待的过程中雷打不动任何人都干扰不到他他死死的盯着鱼漂一有动静钩被咬了就提竿上鱼。这种等待的方式我们称为阻塞式IO这也是大部分常用的读写接口的方式。
2.钓鱼佬李四是一个资深的钓鱼佬在完成准备工作时李四就先观察了鱼漂发现没动静就拿起了手机刷视频看一会儿之后再检查鱼漂发现上钩了于是提杆上鱼。李四每隔一段时间进行检查如果满足条件就读写反之就干自己的事。这种方式称为非阻塞式IO--非阻塞轮询。
3.钓鱼佬王五拥有较为先进的鱼竿完成准备工作后直接就把鱼竿插入地上一心一意的干其它事鱼竿上有报警装置一旦上鱼立刻发送警报告知王五所以王五只需要看有没有报警信号。这种方式被称为信号驱动式IO。
4.赵六身家万贯的热爱钓鱼的钓鱼佬秉持着杆多鱼多的准则直接准备了一卡车鱼竿依次完成准备工作入水后准备开钓由于鱼竿数量多赵六在岸上来回检查看哪只鱼竿上货了。这种方式称为多路复用多路转接。
5.田七以吃鱼为爱好的世界五百强上市公司CEO司机小王拉着田七来钓鱼刚准备钓鱼由于公司业务繁忙对钓鱼不是很感冒田七又回到了公司处理且告知小王车上装备齐全你在这里钓鱼掉满了之后再给我打电话我来接你。对于田七来说这种方式称为异步IO。
首先我们来看看阻塞式与非阻塞式 IO
两者在效率上基本一样有人觉得非阻塞式效率会高一点这指的是非阻塞可以去干别的事但是对于IO效率等拷贝是一样的区别是等的方式不同一个一直等一个间隔等一会儿。
再看同步与异步
同步参与等或者参数与了拷贝都可以是同步IO。异步两者都没参与只拿数据。
那么综上哪一种效率更高呢?
实际上就是田七多路复用IO。我们学习的重点也就是多路复用. 非阻塞式IO fcntl 一个文件描述符 , 默认都是阻塞 IO,通过函数fcntl可以修改文件描述符的属性。 函数原型如下 #include unistd.h
#include fcntl.h
int fcntl(int fd, int cmd, ... /* arg */ ); 传入的 cmd 的值不同 , 后面追加的参数也不相同 . fcntl 函数有 5 种功能 : 复制一个现有的描述符 cmdF_DUPFD . 获得 / 设置文件描述符标记 (cmdF_GETFD 或 F_SETFD). 获得 / 设置文件状态标记 (cmdF_GETFL 或 F_SETFL). 获得 / 设置异步 I/O 所有权 (cmdF_GETOWN 或 F_SETOWN). 获得 / 设置记录锁 (cmdF_GETLK,F_SETLK 或 F_SETLKW). 我们此处只是用第三种功能 , 获取 / 设置文件状态标记 , 就可以将一个文件描述符设置为非阻塞 实现函数 SetNoBlock 基于 fcntl, 我们实现一个 SetNoBlock 函数 , 将文件描述符设置为非阻塞 . void SetNoBlock(int fd) { int fl fcntl(fd, F_GETFL); if (fl 0) { perror(fcntl);return; }fcntl(fd, F_SETFL, fl | O_NONBLOCK);
} 使用 F_GETFL 将当前的文件描述符的属性取出来 ( 这是一个位图 ). 然后再使用 F_SETFL 将文件描述符设置回去 . 设置回去的同时 , 加上一个 O_NONBLOCK 参数 了解到这两个主要的接口我们就可以简单的实现一下非阻塞
#includeiostream
#includestring.h
#includeunistd.h
#includefcntl.h
#includecstdio
using namespace std;//默认阻塞我们设置为非阻塞
void setNonBlock(int fd)
{int f1fcntl(fd,F_GETFL);if(f10){cerrget failedendl;}//设置标记位,实际上是在原基础上新增一个标记位fcntl(fd,F_SETFL,f1 | O_NONBLOCK);//以非阻塞式打开coutset O_NONBLOCK succedendl;
}int main()
{char buffer[1024];//缓冲区while(true){printf(please enter#);fflush(stdout);setNonBlock(0);sleep(1);ssize_t sizeread(0,buffer,sizeof(buffer)-1);//从标准输入中读取if(size0){buffer[size-1]0;cout读取到数据 :bufferendl;}else if(size0){coutread doneendl;break;}else{cerrerrno:strerror(errno)endl;}}return 0;
}一般阻塞式系统就会等待着我们输入数据后才进行下一步工作这里是打印输入的字符。
我们设置为非阻塞式有几点需要注意
1.设置为非阻塞系统不会等待着我们输出才打印而是以读取到数据0打印错误信息。
2,出错非两种情况第一个就是fd真的异常 第二个就是底层还没就绪你就返回了。这里是第二种。
3.非阻塞也能获取到数据并打印不过给你就绪的时间非常短。
对于第二点如何区分是还没就绪还是fd异常可以通过打印错误码而知晓。
多路复用多路转接
实际上对于阻塞式非阻塞式非阻塞式轮询这些等待和拷贝一个函数就会搞定了但是对于多路转接因为要提高IO效率因此等待与拷贝是分开的的多路复用的本质就是非阻塞轮询因此我们需要将非阻塞和轮询逐一实现。
IO多路转接之select
初识select 系统提供 select 函数来实现多路复用输入 / 输出模型 . select 系统调用是用来让我们的程序监视多个文件描述符的状态变化的 ; 程序会停在 select 这里等待直到被监视的文件描述符有一个或多个发生了状态改变 select函数原型 select的函数原型如下: #include sys/select.h
int select(int nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout);
其中有三个参数类型为fd_set位图类型为输入输出型参数。
为输入型参数时用户告诉内核我给一些fd你帮我注意观察一些fd的读事件如果读事件就绪你就告诉我
为输出型参数时内核告诉用户你给我传来的fd我已经帮你知晓了有哪些已经就绪进行位图修改你可以读了。
至于为什么用位图因为位图本质就是设置某个数的第几位为0/1,举例如果此时内核发现0号和1号文件被描述符已经就绪就会把第0位和第1位比特位置为1 比特位的位置代表对应文件描述符从右向左。比特位的内容就可以表示内核是否要关心。 作为输入型位图 比特位的位置代表文件描述符 比特位的内容代表是否内核需要关心 作为输出型位图 比特位的位置还是文件描述符 比特位的内容代表用户关心的fd的读事件就绪了 于是就提供了一系列位图的接口用来修改为位图。 其中timeout为一个结构体里面有两个成员一个代表秒一个代表微秒。、
如果设置了timeout在该时间内如果有已经就绪的立即返回之后的timeout表示剩下的时间。
如果没设置只要有一个就绪了就返回。
select的返回值: 返回值 正常情况下返回已准备好的文件描述符的个数。经过超时时长后仍无设备准备好返回值为 0。如果 select 被某个信号中断返回 -1 并设置 errno 为 EINTR。如果出错返回 -1 并设置相应的 errno。 对于服务端来说使用多路复用可以提高IO效率当没有用访问服务端使此时select的个数为零但一有客户端访问就会有对应的文件描述符接收缓冲区被创建并且添加进行状态检测。
select.hpp
#include Socket.hpp
#include sys/select.h
#include sys/time.h
#include sys/types.h
#include unistd.h
#include iostreamconst int max_fd (sizeof(fd_set)) * 8; // 8个位图最多存储的bit位
static const uint32_t defaultport 8080;
class SelectServer
{
public:SelectServer(const uint32_t Port defaultport) : port(Port){// 对辅助数组初始化为-1for (int i 0; i 1024; i){fd_arry[i] defaultfd;}}~SelectServer(){_listenfd.Close();}bool Init(){_listenfd.Createsockfd();_listenfd.Bind(port);_listenfd.Listen();return true;}void Accept(){// 连接就绪,获取新连接可以通信了std::string clientip;uint16_t clientport;int sockfd _listenfd.Accept(clientip, clientport); // 就绪了不在阻塞if (sockfd 0)return;std::cout 获取连接成功 描述符: sockfd std::endl;// 获取之后不能进行读取而是设置select不然还是会阻塞在select中int pos 1;for (; pos max_fd; pos){if (fd_arry[pos] ! defaultfd){continue;}else{break;}}if (pos max_fd){std::cout can not handle std::endl;close(sockfd); // 无法处理就关掉接收的文件描述符}else{// 找到了空闲位置此时放到fd_arry里面管理fd_arry[pos] sockfd;}}void recver(int fd,int i){// 不在位图中代表的是接收后新的fdchar buffer[1024];ssize_t n read(fd, buffer, sizeof(buffer) - 1);if (n 0){std::cout read meesage: buffer std::endl;}else if(n0){std::coutclient closed std::endl;//如果为空说明要关闭连接了close(fd);fd_arry[i]defaultfd;}else{std::coutread warningstd::endl;close(fd);fd_arry[i]defaultfd;}}void Disapacher(fd_set fds){// 为了确定就绪的文件描述符。需要遍历整个数组for (int i 0; i max_fd; i){int fd fd_arry[i];if (fd defaultfd){continue;}// 先判断文件描述符是否在位图中if (fd _listenfd.Fd() FD_ISSET(fd, fds)) // 判断是否是位图中的是就代表就绪{Accept();}else{recver(fd,i);}}}void Start(){int listensock _listenfd.Fd(); // 获取套接字的文件描述符fd_arry[0] listensock;fd_set rfds; // 读的文件描述符位图FD_ZERO(rfds); // 先进行清空struct timeval timeout {10, 0}; // 设置时间戳每隔五秒检验一次int fdmax fd_arry[0];while (true){for (int i 0; i max_fd; i){if (fd_arry[i] defaultfd){continue;}else{// 如果不为-1说明该位置有要检测的文件描述符FD_SET(fd_arry[i], rfds); // 将指定的文件描述符添加进来(交给select进行检测)// 同时找出最大的文件描述符if (fd_arry[i] fdmax){fdmax fd_arry[i];std::cout max has updated: fdmax std::endl;}}}// 这里不能直接进行acceptaccept的本质就是检测并获取listenfd上面的事件accept只能阻塞等一个// 因为我们要实现多路复用所以这里是要去等待多个套接字并检测他们的状态使用selectint n select(fdmax 1, rfds, nullptr, nullptr, timeout); // 通过内核的select帮我进行检测刚开始启动只有一个listenfdswitch (n){case 0:std::cout time out std::endl;break;case -1:std::cout select error std::endl;default:// 有链接了如果对链接不处理select会一直通知你,需要处理并设置位图// 新连接到来我们认为是读事件,Disapacher(rfds);//名为事件派发器break;}}}private:Sock _listenfd;uint32_t port;int fd_arry[max_fd]; // 辅助数组统计所有描述符
};main.cc
#includeSelectSever.hpp
#includememory
#includeiostreamvoid usehelp(char*s)
{printf(use correct code:%s port[1024]\n,s);
}
int main(int argc,char*argv[])
{std::unique_ptrSelectServer server(new(SelectServer));server-Init();server-Start();return 0;
}
编写select的思路 1.首先创建绑定监听套接字之后创建文件描述符数组对所有文件描述符管理。 2.插入最开始的listenfd到数组里遍历数组找出最大的文件描述符设置位图用来标识就绪的文件描述符之后select数组最大fd1文。 2.select之后判断是否有新链接有新的连接这里我们默认是读事件就会有新的文件描述符对事件进行处理。 3.处理时数组里的文件描述符分两种一种使位图中的就绪的fd一种是就绪后进行accept的新的fd因此对整个数组进行判断是位图中的就行新接受新连接是accept的进行读事件反之是-1continue。 但是select是有很明显的缺陷 1.等待的fd是有上限的--1024 2.输入输出参数较多数据拷贝频率高 3.输入输出参数多每一次都对需要关心的fd重置以达到复用 4.管理数组fd,有太多次进行遍历 为了解决种种情况我们使用了下一种接口poll: poll 函数接口 说明 fds 是一个 poll 函数监听的结构列表 . 每一个元素中 , 包含了三部分内容 : 文件描述符 , 监听的事件集合 , 返 回的事件集合. nfds 表示 fds 数组的长度 . timeout 表示 poll 函数的超时时间 , 单位是毫秒 (ms). 返回结果 返回值小于 0, 表示出错 ; 返回值等于 0, 表示 poll 函数等待超时 ; 返回值大于 0, 表示 poll 由于监听的文件描述符就绪而返回 这里我们介绍一下重点放在select_poll上。
IO多路转接之epoll
epoll的说明是为了处理大批量句柄而改进的poll
1.认识有关epoll的接口
epoll_create 创建一个epoll模型 epoll_wait 等待多长时间进行一次事件检查 返回就绪的fd与event 返回值位已经准备就绪的个数。对于event的宏表示的就绪时间类型如下 EPOLLIN : 表示对应的文件描述符可以读 ( 包括对端 SOCKET 正常关闭 ); EPOLLOUT : 表示对应的文件描述符可以写 ; EPOLLPRI : 表示对应的文件描述符有紧急的数据可读 ( 这里应该表示有带外数据到来 ); EPOLLERR : 表示对应的文件描述符发生错误 ; EPOLLHUP : 表示对应的文件描述符被挂断 ; EPOLLET : 将 EPOLL 设为边缘触发 (Edge Triggered) 模式 , 这是相对于水平触发 (Level Triggered) 来说的 . EPOLLONESHOT 只监听一次事件 , 当监听完这次事件之后 , 如果还需要继续监听这个 socket 的话 , 需要 再次把这个 socket 加入到 EPOLL 队列里 . epoll_ctl 作用是对系统的文件描述符的事件增加删除修改。即对文件描述符做管理 第一个参数就是epoll_create的返回值op代表三个选项增加修改删除第三个代表那个fd对应的哪个事件。
对于epoll的使用时要使用这三个系统接口的。
2.epoll原理 当某一进程调用 epoll_create 方法时 Linux 内核会创建一个 eventpoll 结构体这个结构体中有两个成 员与epoll 的使用方式密切相关 每一个 epoll 对象都有一个独立的 eventpoll 结构体用于存放通过 epoll_ctl 方法向 epoll 对象中添加进来 的事件. 这些事件都会挂载在红黑树中如此重复添加的事件就可以通过红黑树而高效的识别出来 ( 红黑树的插 入时间效率是lgn 其中 n 为树的高度 ). 而所有添加到 epoll 中的事件都会与设备 ( 网卡 ) 驱动程序建立回调关系也就是说当响应的事件发生时 会调用这个回调方法. 这个回调方法在内核中叫 ep_poll_callback, 它会将发生的事件添加到 rdlist 双链表中 . 在 epoll 中对于每一个事件都会建立一个 epitem结构体 实际上epoll和poll是两个不同的模型没太多关系epoll是专门为用户设计的一个底层事件调度的模型。
3.编写epoll
对于select来说我们使用一个系统调用接口加上文件描述符数组来进行非阻塞式的等待。
那么对于epoll来说就是通过三个系统调用接口1.创建模型3.进行非阻塞式等待时间自己设置2.使用epoll_ctl管理事件本质上时使用红黑树的节点绑定事件进行高效的管理。
每一个事件和她的描述符都被封装在一个结构体epoll_event当中之后通过一个数组管理这里epol_event。 第一个不太完美的版本的eopp如下
epoler.hpp
//封装epoll接口
#pragma once
#include sys/epoll.h
#include unistd.h
#includeiostream
#includestring.h
#includeNocopy.hppuint32_t EVENT_IN(EPOLLIN); //读事件
uint32_t EVENT_OUT(EPOLLOUT);//写事件//因为我们不想该类被拷贝所以继承不被拷贝的类
class Epoll :public nocopy
{public:static const int size128;Epoll():_timeout(3000) //设置检测时间3s一次,若为零就是非阻塞等待{//创建epoll模型_epfdepoll_create(size);if(_epfd-1){std::couterrno: errnoerror message:strerror(errno)std::endl;}else{std::coutcreate succed ,epfd:_epfdstd::endl;}}//使用epoll接口进行等待int Epollwait(struct epoll_event events[],int num) //输入参数events表示就序的事件 num表示个数{int retepoll_wait(_epfd,events,num,_timeout);//就绪了就从这里拿}int Epollctl(int oper,int sock,uint32_t event){//向指定模型根据oper添加修改删除事件if(operEPOLL_CTL_DEL){//这里对删除事件进行单独处理int retepoll_ctl(_epfd,oper,sock,nullptr);}else{//注册事件struct epoll_event ev; //创建事件结构体并设置ev.eventsevent;//设置事件ev.data.fdsock;//设置fdint retepoll_ctl(_epfd,oper,sock,ev);//进行什么操作呢 本质就是向红黑树新增节点节点绑定事件if(ret-1){std::couterrno: errnoerror message:strerror(errno)std::endl;}else{std::coutadd succed ,epfd:_epfdstd::endl;}return ret;}}~Epoll(){if(_epfd-1){close(_epfd);std::coutclose succed ,epfd:_epfdstd::endl;}}private:int _epfd;int _timeout;
};
epollserver.hpp
#pragma once
#includeSocket.hpp
#includeEpoller.hpp
#includeNocopy.hpp
#includeiostream
#includememory
#includesys/epoll.h//设置你要关心的时事件const uint16_t defalutport8080;
class EpollServer :public nocopy
{public:static const int num64;//每次从就绪队列中取64个就绪事件EpollServer(uint16_t portdefalutport):_port(port),_listensock(new Sock),_epoller(new Epoll){}bool Init(){_listensock-Createsockfd();_listensock-Bind(_port);_listensock-Listen();return true;}void Start(){//将listensock添加到epoll中检测你关心的读写事件,这里就是添加_epoller-Epollctl(EPOLL_CTL_ADD,_listensock-Fd(),EVENT_IN);struct epoll_event evns[num];//num代表最多就绪的个数若果等待的个数太多就会分配批次取waitwhile(true){int n_epoller-Epollwait(evns,num); //返回值代表就绪的个数if(n0){//在事件插入之后有事件就绪了std::cout检测对应的fd的事件,fd:evns[0].data.fdstd::endl; //既然就绪那就开始处理事件HandlerEvent(evns,n);}else if(n0){std::couttime outstd::endl;}else{std::coutwait errorstd::endl;}}}void Accepter(){std::string clientip;uint16_t clientport;int sock _listensock-Accept(clientip,clientport);if(sock0){//获取到新连接再插入到红黑树当中_epoller-Epollctl(EPOLL_CTL_ADD,sock,EVENT_IN);}else{std::coutaccept errorstd::endl;}}void Dispatcher(int fd){char buffer[1024];// 已经获取新连接开始readint n read(fd, buffer, sizeof(buffer) - 1);if (n 0){buffer[n] 0;std::cout 读取到数据: buffer std::endl;//读完之后写回给客户端std::string echo_stringserver get message;std::string contentecho_stringstd::string(buffer);write(fd,content.c_str(),content.size());}else if(n0){std::cout client close: std::endl;//删除对应的文件描述符 _epoller-Epollctl(EPOLL_CTL_DEL,fd,0); //从红黑树中拿出去再关闭fdclose(fd);}else{std::cout read error!! std::endl;_epoller-Epollctl(EPOLL_CTL_DEL,fd,0); //从红黑树中拿出去再关闭fdclose(fd);}}//这里的事件假设就两种读 写void HandlerEvent( struct epoll_event evns[],int num){for(int i0;inum;i){uint32_t eventevns[i].events;int fdevns[i].data.fd; //判断是哪一种事件if(eventEVENT_IN){//判断fd是等待的fd还是已经获取新连接的fdif(fd_listensock-Fd()){Accepter();}else{Dispatcher(fd);}}else if(event EVENT_OUT){//写事件就绪}}}~EpollServer(){_listensock-Close();}private:std::shared_ptrSock _listensock;std::shared_ptrEpoll _epoller;uint16_t _port;};
4.epoll的工作模式
epoll的工作模式有利各种ET 和LT
LTlevel triggered工作模式--水平工作模式 epoll 默认状态下就是 LT 工作模式 . 1.当 epoll 检测到 socket 上事件就绪的时候 , 可以不立刻进行处理 . 或者只处理一部分 . 如上面的例子 , 由于只读了 1K 数据 , 缓冲区中还剩 1K 数据 , 在第二次调用 epoll_wait 时 , epoll_wait 仍然会立刻返回并通知socket 读事件就绪 . 直到缓冲区上所有的数据都被处理完 , epoll_wait 才不 会立刻返回 . 2.支持阻塞读写和非阻塞读写 通俗点说如果上层有新的数据你不拿走上层就一直通知你让你取数据只要有就拿走。
ETEdge Triggered工作模式 --边缘工作模式
如果我们在第1步将socket添加到epoll描述符的时候使用了EPOLLET标志, epoll进入ET工作模式. 当 epoll 检测到 socket 上事件就绪时 , 必须立刻处理 . 如上面的例子 , 虽然只读了 1K 的数据 , 缓冲区还剩 1K 的数据 , 在第二次调用 epoll_wait 的时候 , epoll_wait 不会再返回了 . 也就是说 , ET 模式下 , 文件描述符上的事件就绪后 , 只有一次处理机会 . ET 的性能比 LT 性能更高 ( epoll_wait 返回的次数少了很多 ). Nginx 默认采用 ET 模式使用 epoll. 只支持非阻塞的读写 通俗点说只有发生变化从没链接到有新连接从一个变多个从多个变没就会进行通知取走数据只有变化才会拿走。
由于通知的次数少且通知是有限的所以一般而言ET的工作效率更高一些但也因为之中特性程序员必须每次把数据都取完循环读取直到读取出错否则就会遗漏。
但是读完数据我们此时的read是阻塞的我们不能让她阻塞这会导致服务器挂起这是不好的所以我们需要设置读是非阻塞的。直到非阻塞状态下还读完了就会返回错误码。
且由于我们把缓冲区的数据都把走了这一位则TCP通信时窗口就更大了一次拿去的数据也多了。
注意
即使如此一般情况ET比LT高效但是你是一直比LT高效吗而且既然你ET都能一次性非阻塞的读取完我LT不行吗只是我们一般去使用ET如果可以你也可以去使用LT修改。
读写的改进 reactor
所以我们的读写是不合理的首先没有协议的定制其次读取并不是非阻塞的。
这里就还是以编写计算器的客户端与服务端为例
重要的读写部分在TcpServer.hpp中
TcpServer.hpp
#pragma once
#includestring
#includeiostream
#includememory
#includefunctional
#includeunordered_map
#include arpa/inet.h
#includeNocopy.hpp
#includeSocket.hpp
#includeEpoller.hpp
#includeComen.hpp
#includeProtocol.hpp
#includeCalculator.hppclass Connection;
class TcpServer;
using functstd::functionvoid (std::shared_ptrConnection); //定义了返回值为void参数为td::shared_ptrConnection指针的包装器
const static int buffersize1024;
const uint16_t defaultport8080;
const int num64;
class Connection
{public:void SetHandle(funct recv_cb,funct send_cb,funct except_cb){_recver_callbackrecv_cb;_send_callbacksend_cb;_except_callbackexcept_cb;}void Appendinbuffer(std::string buffer){_inbufferbuffer;//输入缓冲区}void Appendoubuffer(std::string buffer){_outbufferbuffer;}const std::string inbuffer(){return _inbuffer;}Connection(int sock,std::shared_ptrTcpServer tcpserver_ptr):_sock(sock),_tcpserver_callback(tcpserver_ptr)//回值指针用来传递TCPserver{}int getsock(){return _sock;}std::string getinbuf(){return _inbuffer;}std::string getoutbuf(){return _outbuffer;}~Connection(){}private:int _sock;private:std::string _inbuffer;//输入缓冲区std::string _outbuffer;//输出缓冲区public:funct _recver_callback; //接收回调funct _send_callback; //发送回调funct _except_callback; //异常回调//tcpserver回值指针std::shared_ptrTcpServer _tcpserver_callback;
};class TcpServer
{public:TcpServer(funct OnMessage,uint16_t portdefaultport):_epoll_server(new Epoll),_listensocket(new Sock),_port(port),_quit(true)//回调函数用来设置处理事件{_OnMessageOnMessage;_listensocket-Createsockfd();_listensocket-Bind(_port);_listensocket-Listen();SetNonBlock(_listensocket-Fd()); //将文件设置为非阻塞等待AddConnection(_listensocket-Fd(),EVENT_IN,std::bind(TcpServer::Accepter,this,std::placeholders ::_1),nullptr,nullptr);}void Init(){}void Accepter(std::shared_ptrConnection connection){//连接到来开始接受while(true){struct sockaddr_in peer;socklen_t lensizeof(peer);int fd::accept(connection-getsock(),(sockaddr*)peer,len); //获取新链接if(fd0){uint16_t clientportntohs(peer.sin_port);char ipbuffer[10];inet_ntop(AF_INET,peer.sin_addr,ipbuffer,sizeof(ipbuffer));std::string ipstd::string(ipbuffer);std::coutget new link,ip:ip,fd :fd,portclientportstd::endl;//获取的新连接设置非阻塞现在是ET模式SetNonBlock(fd);//之后再添加到哈希表中epoll_event中AddConnection(fd,EVENT_IN,\std::bind(TcpServer::Recver,this,std::placeholders::_1), //根据函数地址与参数bind成一个包装器对象std::bind(TcpServer::Sender,this,std::placeholders::_1),\std::bind(TcpServer::Exepter,this,std::placeholders::_1));}else{//直到非阻塞被读到没有新的连接了if(errnoEWOULDBLOCK){break;}else if(errnoEINTR){continue;}else{//读出错break;}}}}void AddConnection(int sock,uint32_t event,funct recv_cb,funct send_cb,funct except_cb){// 将listensock添加到epoll中检测你关心的读写事件,这里就是添加 并且设置为边缘工作模式_epoll_server-Epollctl(EPOLL_CTL_ADD, sock, event);//c除此之外每一个fd都要构建成Connetcion对象通过connect管理事件的处理std::shared_ptrConnection new_connectionstd::make_sharedConnection(sock,std::shared_ptrTcpServer(this));new_connection-SetHandle(recv_cb,send_cb,except_cb);//第三步就是添加到unordered_map_connection_map[sock]new_connection;}//事件管理进行事件的处理void Recver(std::shared_ptrConnection connection){int sockconnection-getsock();//ET模式读while(true){char buffer[buffersize];memset(buffer,0,sizeof(buffer));ssize_t nrecv(sock,buffer,sizeof(buffer)-1,0); //这里的数据默认为字符流如果还有二进制需要换为vector if(n0){//读取成功connection-Appendinbuffer(buffer);}else if(n0){//连接关闭std::coutlink closed !std::endl;connection-_except_callback(connection);}else{//读取错误if(errnoEWOULDBLOCK){break;}else if(errnoEINTR){continue;}else{std::coutread error!std::endl;connection-_except_callback(connection);break;}}}//读完之后将数据交给上层_OnMessage(connection); //协议的定制}void Sender(std::shared_ptrConnection connection){while(true){std::string bufferconnection-getoutbuf();int sockfdconnection-getsock();ssize_t nsend(sockfd,buffer.c_str(),buffer.size(),0);if(n0){//清空缓冲区buffer.erase(0,n);if(buffer.empty())break;}else if(n0){return ;}else {if(errnoEWOULDBLOCK){break;}else if(errnoEINTR){continue;//信号中断跳过本次}else {//发失败了进入异常处理std::coutsend errorstd::endl;connection-_except_callback(connection);return;}}if(!buffer.empty()){//对写事件的关心EnableEvent(connection-getsock(),true,true);}else{//关闭对写事件的处理EnableEvent(connection-getsock(),true,false);}}}void EnableEvent(int fd,bool readable,bool writeable ){uint32_t event0;event |((readable ? EPOLLIN:0)|(writeable ? EPOLLOUT : 0)|EPOLLET);_epoll_server-Epollctl(EPOLL_CTL_MOD,fd,event);//修改事件的读写,如果没发完就继续发通过修改事件为写}void Exepter(std::shared_ptrConnection connection){//处理链接异常或着断开std::cout事件异常断开连接std::endl;//首先移除链接//EnableEvent(connection-getsock(),false,false);//读写都设置为False_epoll_server-Epollctl(EPOLL_CTL_DEL,connection-getsock(),0);//移出事件 //关闭异常的文件描述符close(connection-getsock());std::cout已移除connection,已关闭文件描述符:connection-getsock()std::endl;//把map中的也删了}bool isConnectionsafe(int fd){auto it_connection_map.find(fd);if(it_connection_map.end()){return false;}return true;}void Disptcher(int timeout){//进行等待就绪时间int n_epoll_server-Epollwait(events,num,timeout);for(int i0;in;i){uint32_t eventevents[i].events;int sockevents[i].data.fd;//进行事件判断并转化成读写事件if((event EVENT_IN) isConnectionsafe(sock)){//读就绪if(_connection_map[sock]-_recver_callback)//函数存在{_connection_map[sock]-_recver_callback(_connection_map[sock]);//fd对应的connetcion信息其中调用对应的方法参数为connection对象}}if((event EPOLLOUT) isConnectionsafe(sock)){//写就绪if(_connection_map[sock]-_recver_callback){_connection_map[sock]-_send_callback(_connection_map[sock]);}}if(event EPOLLHUP){//读关闭event |(EPOLLIN EPOLLOUT);}if(event EPOLLERR){//错误event |(EVENT_IN EVENT_OUT);}}}void Loop(){_quitfalse;while (!_quit){Disptcher(3000);//3s}_quittrue;}~TcpServer(){}private:std::shared_ptrEpoll _epoll_server; //epoll接口std::unordered_mapint,std::shared_ptrConnection _connection_map; //连接池std::shared_ptrSock _listensocket; //套接字接口struct epoll_event events[num];uint32_t _port;bool _quit; //是否退出funct _OnMessage;//上层回调处理信息}; 剩余代码的所在地reactor · 但成伟/编程学习 - 码云 - 开源中国 (gitee.com)
主要难点是对事件与链接的管理事件的执行读写非阻塞这几点。
