一个网站多个数据库,销售网站内容设计,seo入门教程网盘,WordPress使用sentcloud目录
理解五种IO模型非阻塞IO的设置多路转接之select 实现一个简易的select服务器select服务器的优缺点 多路转接之poll 实现一个简易的poll服务器poll服务器的优缺点 多路转接之epoll epoll原理epoll的优势用epoll实现一个简易的echo服务器 epoll的LT和ET工作模式 什么是LT和…目录
理解五种IO模型非阻塞IO的设置多路转接之select 实现一个简易的select服务器select服务器的优缺点 多路转接之poll 实现一个简易的poll服务器poll服务器的优缺点 多路转接之epoll epoll原理epoll的优势用epoll实现一个简易的echo服务器 epoll的LT和ET工作模式 什么是LT和ET 实现一个简易的reactor服务器
五种IO模型 如何理解五种IO模型 应用层调用read/recv和write/send的时候本质就是数据在用户层和操作系统的缓冲区的拷贝交互。要进行拷贝必须先判断条件成立该条件就是读写事件就绪。如果条件不成立这些函数就会被阻塞住等待资源就绪。 所以输入输出IO可以分为两个步骤等待 拷贝 给大家举个例子帮助大家理解五种IO模型 钓鱼 等待 钓动作 河边的张三在钓鱼张三的钓鱼动作一直盯着鱼漂直到鱼漂抖动就说明有鱼来了。 李四也在钓鱼李四的钓鱼动作一边做自己的事一边时不时观察鱼漂。 王五有个高级鱼竿鱼漂上带着铃铛王五的钓鱼动作一直做自己的事铃铛鱼漂响了就去钓鱼 赵六买了20个鱼竿钓鱼赵六的钓鱼动作一直检查每个鱼漂任何一个鱼漂动了就钓鱼。 小王开车载田七去公司的时候田七路过的时候看见别人钓鱼突然自己想吃野生鱼了就叫小王去钓鱼钓到了就给田七送过去。 人进程/线程。鱼数据。河内核空间。鱼漂数据就绪的事件。鱼竿文件描述符都是通过文件描述符进行操作。钓鱼recv/read 张三对应的是阻塞式IO。李四对应的是非阻塞IO王五对应的是信号阻塞式IO其中王五提前是知道铃铛响了该怎么办该钓鱼赵六对应的是多路转接/多路复用IO。田七对应的是异步IO小王操作系统有鱼了告诉就告诉田七田七直接用即可 在这几个人中你会觉得哪个人的钓鱼效率是最高的钓的鱼最多的很明显是赵六因为河中的鱼对每个钩子的咬钩概率是相同的而赵六的钩子占比是最大的所以必定单位时间内钓鱼的数量是最多的。 阻塞式IO在内核将数据准备好之前系统调用会一直等待所有的文件描述符默认都是阻塞方式。阻塞IO是最常见的IO模型因为简单易上手。 非阻塞IO如果内核还未将数据准备好系统调用仍然会直接返回并且返回EWOULDBLOCK错误码。非阻塞IO往往需要程序员循环的方式反复尝试读写该文件描述符这个过程称为轮询这对CPU来说是较大的浪费一般只有特定场景下才使用。 信号驱动IO内核将数据准备好的时候使用SIGIO信号通知应用程序进行IO操作 IO多路转接虽然从流程上看起来和阻塞IO类似实际上最核心在于IO多路转接能够同时等待多个文件描述符的就绪状态 异步IO由内核在数据拷贝完成时通知应用程序而信号驱动是告诉进程何时可以开始拷贝数据需要进程自己拷贝 什么叫做高效IO呢 任何IO过程都包含两个步骤等待 拷贝。而且在实际的应用场景中等待消耗的时间往往都远远高于拷贝的时间。让IO更高效最核心的办法就是让等待的时间尽量少。在单位时间内IO过程中等待对于拷贝的比重越小IO效率越高。 阻塞式IO和非阻塞IO有什么区别 区别等待时的状态不同。 阻塞调用是指调用结果返回之前当前线程会被挂起调用线程只有在到得到结果之后才会返回。 非阻塞调用指在不能立刻得到结果之前该调用不会阻塞当前线程。 同步IO和异步IO有什么区别 同步就是在发出一个调用时在没有得到该调用的结果之前该调用就不会返回。但是一旦调用返回就得到返回值了。换句话说就是有调用者主动等待这个调用结果。 异步调用在发出之后这个调用就直接返回所以没有返回结果。直到被动的通过状态/信号来通知来处理 非阻塞IO的设置 如何设置非阻塞IO 以前我们在写网络编程的时候recv的flag选项填的是0 如果我们想非阻塞等待则可以设flags为MSG_DONTWAIT。也可以用open非阻塞的方式打开方法有很多。 但提供一种更通用的做法文件描述符本质就是下标每一个下标指向的就是内核里的文件对象文件对象中是有文件描述符标志的。 传入的cmd值不同后面追加的参数也不相同 fcntl函数有5种功能 复制一个现有的描述符(cmd F_DUPFD) 获得/设置文件描述符标记(cmd F_GETFD或F_SETFD) 获得/设置文件状态标记(cmd F_GETFL或F_SETFL) 获得/设置异步I/O所有权(cmd F_GETOWN或F_SETOWN) 获得/设置记录锁(cmd F_GETLK, F_SETLK或F_SETLKW) 实现非阻塞轮询方式读取标准输入 #include iostream
#include unistd.h
#include fcntl.hvoid SetNoBlock(int fd)
{int flag fcntl(fd, F_GETFL);fcntl(fd, F_SETFL, flag | O_NONBLOCK);
}int main()
{SetNoBlock(0);char buffer[4096];while (1){std::cout Enter#;fflush(stdout);int n read(0, buffer, sizeof(buffer) - 1);if (n 0){buffer[n] 0;std::cout buffer std::endl;}else if (n 0){std::cout read done std::endl;break;}else{std::cout read error: strerror(errno) std::endl;break;}}return 0;
}为什么还没有输入就直接读出错了错误信息是资源暂时还没有就绪。 结论 设置成为非阻塞如果底层fd数据没有就绪recv/read/write/send返回值会以出错的形式返回。这样会有两种情况a、真的出错 b、底层资源没有就绪 我们怎么区分呢 – 通过errno区分 #include iostream
#include unistd.h
#include fcntl.h
#include cstringvoid SetNoBlock(int fd)
{int flag fcntl(fd, F_GETFL);fcntl(fd, F_SETFL, flag | O_NONBLOCK);
}int main()
{SetNoBlock(0);char buffer[4096];while (1){int n read(0, buffer, sizeof(buffer) - 1);if (n 0){buffer[n] 0;std::cout buffer std::endl;}else if (n 0){std::cout read done std::endl;break;}else{if (errno EWOULDBLOCK){//只是底层没有数据就绪这种错误是可以容忍的//do_other_thing(); //也可以做其他事情设置一些方法就不演示了continue;}std::cout read error: strerror(errno) std::endl;break;}}return 0;
}多路转接之select select是干什么的select负责多路转接IO中的等待一次可以等待多个文件描述符并不负责拷贝。 select系统调用是用来让我们的程序监视多个文件描述符的状态变化的程序会停在select这里等待直到被监视的文件描述符有一个或多个发生了状态变化。 nfds填写要监视的所有要监视的fd中的最大值 1。这与底层内核有关系内核会从0一直遍历到nfds。 timeout为输入输出型参数 如果设置为nullptr则为阻塞式。 若设置timeout {0, 0};为非阻塞会一直函数返回。 若设置timeout {5, 0};如果一直收不到消息阻塞5s每过5s就超时函数返回一次这样反复循环。如果过2s收到了消息则该参数会返回{3, 0}表示还有3s才超时。 return返回值大于0有几个fd就绪了。等于0超时/非阻塞返回了。小于0select调用失败了 readfds为输入输出参数 做输入参数时表示用户告诉内核你要帮我关心的文件描述符的读fd_set表示位图可以通过位图设置多个文件描述符。 做输出参数时表示内核告诉用户你要关心的文件描述中有哪些已经就绪了。 writefds和exceptfds同readfds参数 该位图既然是一种类型必然有大小所以能够想关心的fd的个数一定是有上限的。 int main() { std::cout sizeof(fd_set) std::endl; return 0; } 128 * 8 1024bit最多只能关心1024个fd。bit位的位置代表fd是多少为0则代表没设置该fd为1则代表设置了该fd。 实现一个简易的select服务器 为了节省篇幅后面再贴完整代码 #include Socket.hppclass SelectServer
{
public:SelectServer(uint16_t port):_port(port){}void Init(){_listensock.CreateSocket();_listensock.Setsockopt(); //设置端口复用_listensock.Bind(_port);_listensock.Listen();}void Start(){while (1){//能否直接调用accept不能直接accept因为会阻塞我们的目标就是写多路转接IO所以不能出现阻塞}}~SelectServer(){_listensock.Close();}
private:Socket _listensock;uint16_t _port;
};更新代码 #include Socket.hppclass SelectServer
{
public:SelectServer(uint16_t port):_port(port){}void Init(){_listensock.CreateSocket();_listensock.Setsockopt();//设置地址复用_listensock.Bind(_port);_listensock.Listen();}void HandleEvent(){}void Start(){while (1){//能否直接调用accept不能直接accept因为会阻塞fd_set readfd;FD_SET(_listensock.Fd(), readfd);struct timeval timeout {3, 0};int n select(_listensock.Fd() 1, readfd, nullptr, nullptr, timeout);//暂时这样设置if (n 0){std::cout select error: strerror(errno) std::endl;break;}if (n 0){lg(Debug, [%d: %d], timeout.tv_sec, timeout.tv_usec);}else{std::cout Get a New Link std::endl;sleep(2);HandleEvent();}}}~SelectServer(){_listensock.Close();}
private:Socket _listensock;uint16_t _port;
};运行结果如下 timeout设置为{3, 0}可以看到select每隔3s就超时一次同理如果timeout设置为{0, 0}则会一直超时如果设置为null则会阻塞。 客户端连接上后我们也看到了他一直在打印Get a New Link为什么一直打印因为上层没有把底层数据拿上去处理所以select就会一直提醒事件就绪了。 怎么处理 更新代码 #include Socket.hppclass SelectServer
{
public:SelectServer(uint16_t port):_port(port){}void Init(){_listensock.CreateSocket();_listensock.Setsockopt();_listensock.Bind(_port);_listensock.Listen();}void HandleEvent(fd_set* rfd){//代码走到这里就说明我们的连接事件就绪了if (FD_ISSET(_listensock.Fd(), rfd) true){string clientIp; uint16_t clientPort;int newSockfd _listensock.Accept(clientIp, clientPort); //会不会阻塞在这里不会因为走到这里说明资源已经就绪了if (newSockfd 0){return;}lg(Info, accept success: %s, %d\n, clientIp.c_str(), clientPort);}}void Start(){while (1){//能否直接调用accept不能直接accept因为会阻塞fd_set readfd;FD_SET(_listensock.Fd(), readfd);struct timeval timeout {3, 0};int n select(_listensock.Fd() 1, readfd, nullptr, nullptr, /*timeout*/0);if (n 0){std::cout select error: strerror(errno) std::endl;break;}if (n 0){lg(Debug, [%d: %d], timeout.tv_sec, timeout.tv_usec);}else{std::cout Get a New Link std::endl;sleep(1);HandleEvent(readfd);}}}~SelectServer(){_listensock.Close();}
private:Socket _listensock;uint16_t _port;
};代码运行结果 更新HandleEvent函数的代码 void HandleEvent(fd_set* rfd)
{//代码走到这里就说明我们的连接事件就绪了if (FD_ISSET(_listensock.Fd(), rfd) true){string clientIp; uint16_t clientPort;int newSockfd _listensock.Accept(clientIp, clientPort); //会不会阻塞在这里不会因为走到这里说明资源已经就绪了if (newSockfd 0){return;}lg(Info, accept success: %s, %d\n, clientIp.c_str(), clientPort);char buffer[4096];ssize_t n read(newSockfd, buffer, sizeof(buffer) - 1);/*我们已经得到了新的sock可以直接用read接口吗不可以因为我们这里是单线程/单进程直接read可能会被阻塞住以前我们的代码是直接托管给了多线程/多进程他们的阻塞不会影响主进程。我们这个单进程应该将新到来的newSocketfd交给select让select帮我们来管理*/}
}所以此时我们需要将HandleEvent函数里面的newSocketfd传递给Start函数。如何做呢可以使用数组作为该类的成员 – 这个数组也叫做辅助数组这也是select最大的特点之一使用辅助数组让文件描述符在函数之间互相传递。 #include Socket.hppclass SelectServer
{static const int MaxFdNum sizeof(fd_set) * 8; //因为位图fd_set有大小所以最多只能管理1024个fdstatic const int DefaultNum -1;
public:SelectServer(uint16_t port):_port(port){//初始化辅助数组for (int i 0; i MaxFdNum; i){read_fd_array[i] DefaultNum;}}void Init(){_listensock.CreateSocket();_listensock.Setsockopt();_listensock.Bind(_port);_listensock.Listen();}void HandleEvent(fd_set* rfd){for (int i 0; i MaxFdNum; i)//遍历所有的read_fd_array中是否有满足条件FD_ISSET{int socket read_fd_array[i];if (socket DefaultNum)continue;if (FD_ISSET(socket, rfd) true _listensock.Fd() socket){//代码走到这里就说明我们的连接事件就绪了string clientIp; uint16_t clientPort;int newSockfd _listensock.Accept(clientIp, clientPort); //会不会阻塞在这里不会因为走到这里说明资源已经就绪了if (newSockfd 0){return;}lg(Info, accept success, sock: %d, clientIp: %s, clientPort: %d\n, newSockfd, clientIp.c_str(), clientPort);//添加新的文件描述符到辅助数组for (int i 0; i MaxFdNum; i){if (read_fd_array[i] DefaultNum){if (_maxfd newSockfd)_maxfd newSockfd;read_fd_array[i] newSockfd;break;}if (i MaxFdNum - 1){lg(Warning, server is full, close sock: %d, newSockfd);close(newSockfd);}}}else if (FD_ISSET(socket, rfd) true){//说明是其他的套接字读事件就绪了char buffer[4096];int n read(socket, buffer, sizeof(buffer) - 1);if (n 0){buffer[n] 0;lg(Info, client say:%s, buffer);}else if (n 0){//说明对端把连接关闭了lg(Info, client closed...);//把该socket就可以移除关心状态了read_fd_array[i] DefaultNum;close(socket);}else{//读出错了lg(Error, read error);//把该socket就可以移除关心状态了read_fd_array[i] DefaultNum;close(socket);}}}}void Start(){//将lisntensockfd设置进辅助数组read_fd_array[0] _listensock.Fd();_maxfd read_fd_array[0];while (1){//因为select的参数为输入输出参数所以我们每次调用select的时候都需要重新设置一遍select的参数所以需要将下面的语句写入循环里fd_set readfd;FD_ZERO(readfd); //一定要设置为0否则可能会出现select失败的问题for (int i 0; i MaxFdNum; i){if (read_fd_array[i] ! DefaultNum){if (_maxfd read_fd_array[i]) _maxfd read_fd_array[i];//将辅助数组中所要关心的文件描述符全部都设置进去FD_SET(read_fd_array[i], readfd);}}int n select(_maxfd 1, readfd, nullptr, nullptr, 0);//为了易于观察我们将timout参数设置为0if (n 0){std::cout select error: strerror(errno) std::endl;break;}if (n 0){//超时返回非阻塞返回lg(Info, timeout...);}else{HandleEvent(readfd);}}}~SelectServer(){_listensock.Close();}
private:int _maxfd; //最大的文件描述符是什么Socket _listensock;uint16_t _port;int read_fd_array[MaxFdNum]; //记录的是需要关心的读事件的文件描述符-1表示不关心非-1表示需要关心的文件描述符是什么// int write_fd_array[MaxFdNum]; //仅为了理解多路转接IO不考虑这两种情况情况简单化只考虑读事件// int except_fd_array[MaxFdNum];
};代码运行结果 注意我们的可是单进程哦。实现了并发处理多个请求。 整理代码完整代码(允许结果和上面演示效果相同) Log.hpp文件 – 往期文章实现过
#pragma once
#include iostream
#include string
#include stdarg.h
#include time.h
#include sys/types.h
#include sys/stat.h
#include fcntl.h
#include unistd.husing namespace std;#define Info 0
#define Debug 1
#define Warning 2
#define Error 3
#define Fatal 4#define Screen 1
#define Onefile 2
#define Classfile 3#define fileName log.txt
//使用前需要创建log目录
class Log
{
public:Log(){printMethod Screen;path ./log/;}void Enable(int method){printMethod method;}void printOneFile(string logname, const string logtxt){logname path logname;int fd open(logname.c_str(), O_WRONLY | O_APPEND | O_CREAT, 0666);//open只会创建文件不会创建目录if (fd 0){perror(open failed);return;}write(fd, logtxt.c_str(), logtxt.size());close(fd);}void printClassFile(int level, const string logtxt){string filename fileName;filename .;filename leveltoString(level);printOneFile(filename, logtxt);}void printLog(int level, const string logtxt){if (printMethod Screen){cout logtxt endl;return;}else if (printMethod Onefile){printOneFile(fileName, logtxt);return;}else if (printMethod Classfile){printClassFile(level, logtxt);return;}}const char* leveltoString(int level){if (level Info) return Info;else if (level Debug) return Debug;else if (level Error) return Error;else if (level Fatal) return Fatal;else return default;}void operator()(int level, const char* format, ...){time_t t time(nullptr);struct tm* st localtime(t);char leftbuffer[4096];snprintf(leftbuffer, sizeof(leftbuffer), year: %d, month: %d, day: %d, hour: %d, minute: %d, second: %d\n\[%s]:,st-tm_year 1900, st-tm_mon 1, st-tm_mday, st-tm_hour, st-tm_min, st-tm_sec, leveltoString(level));char rightbuffer[4096];va_list start;va_start(start, format);vsnprintf(rightbuffer, sizeof(rightbuffer), format, start);va_end(start);char logtxt[4096 * 2];snprintf(logtxt, sizeof(logtxt), %s %s\n, leftbuffer, rightbuffer);printLog(level, logtxt);}
private:int printMethod;string path;//路径与文件名解耦最后将路径和文件粘合起来再用open打开即可
};
Sock.hpp文件 – 往期文章实现过
#include sys/types.h
#include sys/socket.h
#include arpa/inet.h
#include cstring
#include Log.hppLog lg;
class Socket
{
public:Socket(){}~Socket(){}void CreateSocket(){_sockfd socket(AF_INET, SOCK_STREAM, 0);if (_sockfd 0){lg(Fatal, create socket failed:%s, strerror(errno));exit(1);}}void Bind(uint16_t serverPort){struct sockaddr_in server;server.sin_family AF_INET;server.sin_port htons(serverPort);server.sin_addr.s_addr INADDR_ANY;int n bind(_sockfd, (struct sockaddr*)server, sizeof(server));if (n 0){lg(Fatal, bind socket failed:%s, strerror(errno));exit(2);}}void Listen(){int n listen(_sockfd, 5);if (n 0){lg(Fatal, listen socket failed:%s, strerror(errno));exit(3);}}int Accept(string* clientIp, uint16_t* clinetPort){struct sockaddr_in peer;socklen_t len sizeof(peer);int newsocket accept(_sockfd, (struct sockaddr*)peer, len);if (newsocket 0){lg(Error, accept error:%s, strerror(errno));return -1;}*clientIp inet_ntoa(peer.sin_addr);*clinetPort ntohs(peer.sin_port);return newsocket;}int Connect(const string serverIp, uint16_t serverPort){struct sockaddr_in server;server.sin_family AF_INET;server.sin_addr.s_addr inet_addr(serverIp.c_str());server.sin_port htons(serverPort);int n connect(_sockfd, (struct sockaddr*)server, sizeof(server));if (n 0){lg(Error, connect error:%s, strerror(errno));return -1;}return 0;}void Setsockopt(){int opt 1;if (setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, (const void*)opt, sizeof(opt)) 0)lg(Error, %s\n, strerror(errno));}void Close(){close(_sockfd);}int Fd(){return _sockfd;}
private:int _sockfd;
};SelectServer.hpp文件
#include Socket.hppclass SelectServer
{static const int MaxFdNum sizeof(fd_set) * 8; //因为位图fd_set有大小所以最多只能管理1024个fdstatic const int DefaultNum -1;
public:SelectServer(uint16_t port):_port(port){//初始化辅助数组for (int i 0; i MaxFdNum; i){read_fd_array[i] DefaultNum;}}void Init(){_listensock.CreateSocket();_listensock.Setsockopt();_listensock.Bind(_port);_listensock.Listen();}void Recver(int socket, int pos){char buffer[4096];int n read(socket, buffer, sizeof(buffer) - 1);if (n 0){buffer[n] 0;lg(Info, client say:%s, buffer);}else if (n 0){//说明对端把连接关闭了lg(Info, client closed...);//把该socket就可以移除关心状态了read_fd_array[pos] DefaultNum;close(socket);}else{//读出错了lg(Error, read error);//把该socket就可以移除关心状态了read_fd_array[pos] DefaultNum;close(socket);}}void Acceptor(){string clientIp; uint16_t clientPort;int newSockfd _listensock.Accept(clientIp, clientPort); //会不会阻塞在这里不会因为走到这里说明资源已经就绪了if (newSockfd 0){return;}lg(Info, accept success, sock: %d, clientIp: %s, clientPort: %d\n, newSockfd, clientIp.c_str(), clientPort);//添加新的文件描述符到辅助数组for (int i 0; i MaxFdNum; i){if (read_fd_array[i] DefaultNum){if (_maxfd newSockfd)_maxfd newSockfd;read_fd_array[i] newSockfd;break;}if (i MaxFdNum - 1){lg(Warning, server is full, close sock: %d, newSockfd);close(newSockfd);}}}void Dispatcher(fd_set* rfd) //事件派发器{for (int i 0; i MaxFdNum; i)//遍历所有的read_fd_array中是否有满足条件FD_ISSET{int socket read_fd_array[i];if (socket DefaultNum)continue;if (FD_ISSET(socket, rfd) true _listensock.Fd() socket){//代码走到这里就说明我们的连接事件就绪了Acceptor();}else if (FD_ISSET(socket, rfd) true){//说明是其他的套接字读事件就绪了Recver(socket, i);}}}void EventLoop() //事件循环{//将lisntensockfd设置进辅助数组read_fd_array[0] _listensock.Fd();_maxfd read_fd_array[0];while (1){//因为select的参数为输入输出参数所以我们每次调用select的时候都需要重新设置一遍select的参数所以需要将下面的语句写入循环里fd_set readfd;FD_ZERO(readfd); //一定要设置为0否则可能会出现select失败的问题for (int i 0; i MaxFdNum; i){if (read_fd_array[i] ! DefaultNum){if (_maxfd read_fd_array[i]) _maxfd read_fd_array[i];//将辅助数组中所要关心的文件描述符全部都设置进去FD_SET(read_fd_array[i], readfd);}}int n select(_maxfd 1, readfd, nullptr, nullptr, 0);//为了易于观察我们将timout参数设置为0if (n 0){std::cout select error: strerror(errno) std::endl;break;}if (n 0){//超时返回非阻塞返回lg(Info, timeout...);}else{Dispatcher(readfd); //事件派发器}}}~SelectServer(){_listensock.Close();}
private:int _maxfd; //最大的文件描述符是什么Socket _listensock;uint16_t _port;int read_fd_array[MaxFdNum]; //记录的是需要关心的读事件的文件描述符-1表示不关心非-1表示需要关心的文件描述符是什么// int write_fd_array[MaxFdNum]; //仅为了理解多路转接IO不考虑这两种情况情况简单化只考虑读事件// int except_fd_array[MaxFdNum];
};main.cc文件
#include SelectServer.hppint main()
{SelectServer svr(8080);svr.Init();svr.EventLoop();return 0;
}select服务器的优缺点 优点能实现成多路转接的服务器即单进程也能处理多用户的请求 缺点 1.同时等待的fd是有上限的因为位图fd_set是有大小的。 2.用户必须借助第三方数组来维护合法的fd 3.使用select接口设置参数时每次都要对关心的fd进行事件重置。 4.数据拷贝的频率比较高select函数需要每次都需要重新设置参数每传一次位图fd_set内核就需要拷贝一次。 5.如果只有一个fd就绪用户层也需要全部遍历虽然可以改进但没必要 6.内核中检测位图fd_set的事件就绪也要遍历这也就是为什么select的第一个参数传的是文件描述符的最大值因为内核要以这个范围来遍历 多路转接之poll 为了解决上面一些的问题poll就出现了 fdsstruct pollfd数组的首地址 nfds数组元素的个数 timeout设置超时时间单位是ms 填0非阻塞等待 小于0表示阻塞等待 大于0每过timeout就超时一次。理解同select fd你要让内核关心的某一个fd events输入参数用户让内核关心fd的事件 revents输出参数内核告诉用户你要关心的fd哪些事件就绪了 从这里看出来了poll将输入和输出参数分离所以poll不需要像select一样每次都需要将参数重新设定。 short是一个16bit的位图events和revents的取值如下: 事件描 述是否可作为输入是否可作为输出POLLIN数据(包括普通数据和优先数据)可读是是POLLRDNORM普通数据可读是是POLLRDBAND优先级带数据可读Linux不支持是是POLLPRI高优先级数据可读比如TCP带外数据是是POLLOUT数据包括普通数据和优先数据可写是是POLLWRNORM普通数据可写是是POLLWRBAND优先级带数据可写是是POLLRDHUPPOLLRDHUPTCP连接被对方关闭或者对方关闭了写操作。它GNU 引入是是POLLERR错误否是POLLHUP挂起。比如管道的写端被关闭后读端描述符上将收到POLLHUP事件否是POLLNVAL文件描述符没有打开否是 select只分为了读、写、异常事件而poll将fd的事件分的更细了。 实现一个简易的poll服务器
参考代码 Socket.hpp、Log.hpp文件同上
PollerServer.hpp文件
#include Socket.hpp
#include poll.h
#include vectorclass PollServer
{static const int DefaultNum -1;
public:PollServer(uint16_t port):_port(port){}void Init(){_listensock.CreateSocket();_listensock.Setsockopt();_listensock.Bind(_port);_listensock.Listen();}void Recver(int socket, int pos){char buffer[4096];int n read(socket, buffer, sizeof(buffer) - 1);if (n 0){buffer[n] 0;lg(Info, client say:%s, buffer);}else if (n 0){//说明对端把连接关闭了lg(Info, client closed...);//把该socket就可以移除关心状态了_event_fds.erase(_event_fds.begin() pos);close(socket);}else{//读出错了lg(Error, read error);//把该socket就可以移除关心状态了_event_fds.erase(_event_fds.begin() pos);close(socket);}}void Acceptor(){string clientIp; uint16_t clientPort;int newSockfd _listensock.Accept(clientIp, clientPort); //会不会阻塞在这里不会因为走到这里说明资源已经就绪了if (newSockfd 0){return;}lg(Info, accept success, sock: %d, clientIp: %s, clientPort: %d\n, newSockfd, clientIp.c_str(), clientPort);//添加新的文件描述符到辅助数组struct pollfd event;event.fd newSockfd;event.events | POLLIN;_event_fds.push_back(move(event));}void Dispatcher() //事件派发器{for (int i 0; i _event_fds.size(); i)//遍历所有的_event_fds中是否有revents事件就绪的{int socket _event_fds[i].fd;if ((_event_fds[i].revents POLLIN) true){if (_listensock.Fd() socket){//代码走到这里就说明我们的连接事件就绪了Acceptor();}else{//说明是其他的套接字读事件就绪了Recver(socket, i);}}else if ((_event_fds[i].revents POLLOUT) true){//写事件就绪}else{//...}}}void EventLoop() //事件循环{//将lisntensockfd设置进辅助数组struct pollfd event;event.fd _listensock.Fd();event.events | POLLIN;_event_fds.push_back(move(event)); //将listensock添加到_event_fds数组里while (1){int n poll(_event_fds.data(), _event_fds.size(), -1); //易于观察就设置为-1if (n 0){std::cout poll error: strerror(errno) std::endl;break;}if (n 0){//超时返回非阻塞返回lg(Info, timeout...);}else{Dispatcher(); //事件派发器}}}~PollServer(){_listensock.Close();}
private:Socket _listensock;uint16_t _port;std::vectorstruct pollfd _event_fds;
};main.cc文件
#include PollServer.hppint main()
{PollServer svr(8080);svr.Init();svr.EventLoop();return 0;
}运行结果 poll服务器的优缺点 优点 pollfd结构包含了要监视的event事件和就绪的revent事件输入与输出分离了接口使用比select方便。 poll解除了fd有上限的问题数组为vector了vector最大能有多大取决于操作系统了不关poll函数的事了。 缺点 用户层和内核层都需要遍历该数组都是o(n)的效率如果用户要关心的fd有上万个频繁的遍历会影响效率问题。 每次调用poll都需要把大量的pollfd结果从用户态拷贝到内核中。 有大量fd若只有少量的fd处于就绪状态也需要全部线性遍历一遍随着监视描述符数量的增长其效率也会线性下降。 总结解决了select的1、3问题 多路转接之epoll 按照man手册的说法是为处理大批量的socketfd而作了改进的poll。它是在2.5.44内核中被引进的linux2.6下它几乎具备了之前所说的一切优点被公认为性能最好的多路转接IO。现在只要涉及服务器组件的都用的是epoll。 size该参数已经被废弃但是要填大于0的值 该函数会创建一个epoll模型返回epoll模型的文件描述符什么是epoll模型后面会讲。 epfdepoll模型的描述符 events输出型参数struct epoll_event数组的首地址 maxevents该数组的个数 timeout意义同epoll参数的timeout 返回值和select、poll一样 fd当事件就绪时能知道该事件是哪个fd就绪了 events可以是以下几个宏的集合 事件描 述EPOLLIN表示对应的文件描述符可以读(包括对端SOCKET正常关闭)EPOLLOUT表示对应的文件描述符可以写POLLPRI表示对应的文件描述符有紧急的数据可读(这里应该表示有带外数据到来POLLERR表示示对应的文件描述符发生错误POLLPRI表示对应的文件描述符被挂断POLLPRI将EPOLL设为边缘触发(Edge Triggered)模式, 这是相对于水平触发(Level Triggered)来说的EPOLLONESHOT只监听一次事件, 当监听完这次事件之后, 如果还需要继续监听这个socket的话, 需要再次把这个socket加入到EPOLL队列里 epfdepoll模型 op、fd、event用户告诉内核要对fd的event事件进行op处理 op EPOLL_CTL_ADD添加事件 EPOLL_CTL_MOD修改事件 EPOLL_CTL_DEL删除事件 epoll解决了select的所有缺点epoll是怎样设计的呢它的原理是什么 epoll原理 如图所示内核中的三个部分红黑树回调函数就绪队列组成了epoll模型。epoll模型并且由struct_file结构体所指向想管理epoll模型就能通过fd找到epoll模型。 红黑树管理已注册的文件描述符epoll_ctl对红黑树的增删改操作时间复杂度为O(logN) 就绪事件的通知机制每当有fd就绪了网卡驱动就会通知(调用回调函数)epoll并执行内核曾经注册好的回调函数。有了这个就绪事件通知机制内核就不用再O(N)的遍历所有的文件描述符是否已经就绪 就绪列表的管理epoll_wait所捞取的都是已经就绪的 epoll的优势 1.用户层不再需要每次调用接口时都将关心fd拷贝给内核了 2.管理fd个数没有上限 3.不再需要辅助数组了因为内核中的红黑树就代替了曾经用户要管理的数组 4.内核中通过事件通知机制O(1)就能将就绪fd放入就绪队列不再需要遍历所有的fd来判断是否就绪 5.对事件的管理更方便只需要对epoll_ctl进行操作 select和poll的底层如图所示 用epoll实现一个简易的echo服务器
参考代码 Log.hpp文件 和 Socket.hpp文件和之前一样
Epoller.hpp文件
#include sys/epoll.h
#include Log.hpp
#include cstringclass Epoller
{
public:Epoller(){}void EpollerCreate(){_epollfd epoll_create(64);if (_epollfd 0){lg(Fatal, Create epoll failed: %s, strerror(errno));abort();}}void EpollerUpate(int oper, int fd, uint32_t event){int n 0;if (oper EPOLL_CTL_ADD || oper EPOLL_CTL_MOD){struct epoll_event epEvent;epEvent.data.fd fd;epEvent.events 0;epEvent.events | event;n epoll_ctl(_epollfd, oper, fd, epEvent);if (n 0){lg(Error, epoll ctl failed: %s, strerror(errno));}}else if (oper EPOLL_CTL_DEL){n epoll_ctl(_epollfd, oper, fd, 0);if (n 0){lg(Error, epoll ctl failed: %s, strerror(errno));}}}int Wait(struct epoll_event* events, int n, int timeout){int m epoll_wait(_epollfd, events, n, timeout);if (m 0){lg(Error, epoll wait failed: %s, strerror(errno));}else if (m 0){//超时返回非阻塞返回lg(Info, timeout...);}return m;}int Fd(){return _epollfd;}
private:int _epollfd;
};nocpy.hpp文件 //编程技巧写防拷贝的类的时候直接继承就不用自己再手动写了
class nocpy
{
public:nocpy(){}~nocpy(){}
private:nocpy(const nocpy) delete;nocpy operator(const nocpy) delete;
};EpollServer.hpp文件
#include Socket.hpp
#include Epoller.hpp
#include nocpy.hpp
#include memoryclass EpollServer : public nocpy
{static const int num 64; //一次性最多捞取上来多少个就绪fd, 一次捞取不完下一次可以继续捞取
public:EpollServer(uint16_t port):_port(port), _listensock_ptr(new Socket), _epoller_ptr(new Epoller){}void Init(){_listensock_ptr-CreateSocket();_listensock_ptr-Setsockopt();_listensock_ptr-Bind(_port);_listensock_ptr-Listen();lg(Info, create listen socket success, socket: %d, _listensock_ptr-Fd());_epoller_ptr-EpollerCreate();lg(Info, create epoller success);}void Recver(int socket, int pos){char buffer[4096];int n read(socket, buffer, sizeof(buffer) - 1);if (n 0){buffer[n] 0;lg(Info, client say:%s, buffer);string s server echo:;s buffer;int m write(socket, s.c_str(), s.size());if (m 0){lg(Error, write failed: %s, strerror(errno));}}else if (n 0){//说明对端把连接关闭了lg(Info, client closed...);//把该socket就可以移除关心状态了_epoller_ptr-EpollerUpate(EPOLL_CTL_DEL, socket, 0);close(socket);}else{//读出错了lg(Error, read error);//把该socket就可以移除关心状态了_epoller_ptr-EpollerUpate(EPOLL_CTL_DEL, socket, 0);close(socket);}}void Acceptor(){string clientIp; uint16_t clientPort;int newSockfd _listensock_ptr-Accept(clientIp, clientPort); //会不会阻塞在这里不会因为走到这里说明资源已经就绪了if (newSockfd 0){return;}lg(Info, accept success, sock: %d, clientIp: %s, clientPort: %d\n, newSockfd, clientIp.c_str(), clientPort);//添加新的文件描述符到Epoller里面_epoller_ptr-EpollerUpate(EPOLL_CTL_ADD, newSockfd, EPOLLIN);}void Dispatcher(struct epoll_event* recvs, int n) //事件派发器{for (int i 0; i n; i)//recvs数组里面全部都是就绪的fd{int socket recvs[i].data.fd;if ((recvs[i].events EPOLLIN) true){if (_listensock_ptr-Fd() socket){//代码走到这里就说明我们的连接事件就绪了Acceptor();}else{//说明是其他的套接字读事件就绪了Recver(socket, i);}}else if ((recvs[i].events EPOLLOUT) true){//写事件就绪}else{//...}}}void EventLoop() //事件循环{//添加_listensock到Epoller里面_epoller_ptr-EpollerUpate(EPOLL_CTL_ADD, _listensock_ptr-Fd(), EPOLLIN);struct epoll_event recvs[num];while (1){int n _epoller_ptr-Wait(recvs, num, -1); //易于观察就设置为-1Dispatcher(recvs, n); //捞取上来了n个fd}}~EpollServer(){_listensock_ptr-Close();}
private:std::shared_ptrSocket _listensock_ptr; //智能指针优势1.可共享拷贝资源 - 不怕浅拷贝 2.灵活的生命周期可随时释放 3.可延迟初始化 4.异常安全std::shared_ptrEpoller _epoller_ptr;uint16_t _port;
};main.cc文件
#include EpollServer.hppint main()
{std::unique_ptrEpollServer svr(new EpollServer(8080 ));svr-Init();svr-EventLoop();return 0;
}代码运行结果 epoll的LT和ET工作模式
什么是LT和ET 一个现象 我们将这行代码注释掉上层不处理新连接到来的这个事件 代码运行结果 发现底层会一直通知上层有新事件到来请上层处理。 一旦有新的连接到来或者有新的数据到来上层如果你不取走底层会一直通知你去取走哦这种模式就叫做LT LT水平触发Level Triggered ET边缘触发Edge Triggered LT、ET和示波器很像 LT一直处于高电平表示为真。ET只有从低点到高点变化的时候才为真 epoll默认模式LT模式。事件到来但是上层不处理高电平一直有效 ET底层数据或者连接从无到有从有到多变化的时候才会通知我们一次。 给大家举一个例子 张三是个快递员是一个比较负责的快递员。假如你买的快递到了有6个快递下楼取一下你可能在打游戏没时间取过一会儿张三又给你打电话你有6个快递你下来取一下。你游戏没打完就是不取但是呢张三期间也一直打电话。张三的同事李四李四也有你的快递路过张三就给李四说我帮你派发吧。张三现在一共有十个快递让你下来取你不下来取张三就一直会给你打电话张三送快递的模式就叫做LT模式。 过了一周又要派发你的快递。但是是李四来派发你的快递李四给你打电话说你不来取我就不给你打电话了我只通知你一次。你一听那我还是下去取吧要是走了我快递里就找不到了我的快递在一堆快递里除了快递员不知道哪个是我的快递。你取的时候发现你只能拿走6个当中的4个一次拿不完你把快递拿上去之后剩下的快递你不知道在哪了索性就不要了你又去打你的游戏了。李四看到张三也有你的快递还人情就帮张三派发又给你打电话你有新的快递到了你不下来取我就走了。你一听这人只通知一次你就又下去取了这次就把上次没取完的一并给取走了。 总结 张三是有快递就一直给你打电话。 李四是从无到有从有到多只给你打一次电话你没取干净我也不再通知你。 你认为两个快递员谁的工作效率更高呢 ET因为这个快递员在单位时间内通知的人数是更多的。ET一小时内可以通知50人而LT可能只通知了10人。主要是ET的通知效率高。 ET因为只会通知一次所以会倒逼程序员使用ET工作模式的时候每次通知都必须要把本轮数据全部取走。你怎么知道你把本次就绪底层的数据读取完毕了呢循环读取知道读取不到数据了。一般的fd是阻塞式的fd如果没有数据了会阻塞所以在ET模式下我们的fd必须是非阻塞的。 ET的通知效率高不仅仅如此ET的IO效率也更高原因在于每通知一次就要求程序员把本轮数据全部取走这意味着tcp会向对方通告一个更大的窗口从而概率上让对方一次能给我发送更多的数据提高了网络的吞吐量则提高IO效率。 ET的效率也不是一定比LT高LT也可以将所有的fd设置成为非阻塞然后循环读取通知第一次的时候就全部取走不就和ET一样了嘛。LT是epoll的默认行为使用ET能够减少epoll触发的次数但是代价就是强逼着程序员一次就绪响应就把所有的数据都处理完。所以说如果LT情况下如果也能做到每次就绪的文件描述符都立即处理不让这个就绪被重复提示的话其实性能也是一样的。那为什么LT不代替ET呢因为程序员不能保证完全的替代会可能写出bug。 实现一个简易的reactor服务器 我们之前read函数只读一次因为tcp是面向字节流的所以不能保证读上来的是一个完整的报文。 如果读一次没有读完那需要读第二次也就是循环读所以读到的数据需要存到一个缓冲区里。 下面我们就用epoll来实现一个ET模式的reactor服务器 TcpServer.hpp文件
#include Socket.hpp
#include Epoller.hpp
#include nocpy.hpp
#include memory
#include functional
#include unordered_mapconst int num 64; //一次性最多捞取上来多少个fdclass Connection;
using func_t functionvoid(std::shared_ptrConnection);
class TcpServer;uint32_t EVENT_IN (EPOLLIN | EPOLLET);
uint32_t EVENT_OUT (EPOLLOUT | EPOLLET);//每一个文件描述符都有一套自己的缓冲区和回调函数每当该文件描述符就绪了就调用自己的回调函数
class Connection
{
public:Connection(int fd, TcpServer* tcp_server_ptr):_sockfd(fd), _tcp_server_ptr(tcp_server_ptr){}void SetHandler(func_t recv_cb, func_t send_cb, func_t except_cb){_recv_cb recv_cb;_send_cb send_cb;_except_cb except_cb;}
private:int _sockfd;std::string _inbuffer;std::string _outbuffer;func_t _recv_cb; //读回调函数func_t _send_cb; //写回调函数func_t _except_cb; //异常回调函数std::shared_ptrTcpServer _tcp_server_ptr; //回指指针
};class TcpServer : public nocpy
{
public:TcpServer(uint16_t port):_port(port), _listensock_ptr(new Socket), _epoller_ptr(new Epoller){}void Init(){_listensock_ptr-CreateSocket();_listensock_ptr-Setsockopt();_listensock_ptr-Bind(_port);_listensock_ptr-SetNonBlock(); //设置非阻塞_listensock_ptr-Listen();_epoller_ptr-EpollerCreate();lg(Info, create listen socket success, socket: %d, _listensock_ptr-Fd());}void Dispatcher(int n){for (int i 0; i n; i){}}void Start(){_epoller_ptr-EpollerUpate(EPOLL_CTL_ADD, _listensock_ptr-Fd(), EVENT_IN);while (1){int n _epoller_ptr-Wait(recvs, num, -1); //为了方便观察设置为-1if (n 0){lg(Error, epoll wait failed: %s, strerror(errno));break;}else if (n 0){lg(Info, timeout...);}else {Dispatcher(n);}}}~TcpServer(){}
private:std::shared_ptrSocket _listensock_ptr;std::shared_ptrEpoller _epoller_ptr;struct epoll_event recvs[num]; //捞取就绪事件的数组std::unordered_mapint, std::shared_ptrConnection _connections; //管理所有的连接int _port;
};更新代码 TcpServer.hpp文件
#include Socket.hpp
#include Epoller.hpp
#include nocpy.hpp
#include memory
#include functional
#include unordered_map
#include cassertconst int num 64; //一次性最多捞取上来多少个fdclass Connection;
using func_t functionvoid(std::shared_ptrConnection);
class TcpServer;uint32_t EVENT_IN (EPOLLIN | EPOLLET);
uint32_t EVENT_OUT (EPOLLOUT | EPOLLET);//每一个文件描述符都有一套自己的缓冲区和回调函数每当该文件描述符就绪了就调用自己的回调函数
class Connection
{
public:Connection(int fd, TcpServer* tcp_server_ptr):_sockfd(fd), _tcp_server_ptr(tcp_server_ptr){}void SetHandler(func_t recv_cb, func_t send_cb, func_t except_cb){_recv_cb recv_cb;_send_cb send_cb;_except_cb except_cb;}public:int _sockfd;std::string _inbuffer;std::string _outbuffer;func_t _recv_cb; //读回调函数func_t _send_cb; //写回调函数func_t _except_cb; //异常回调函数std::shared_ptrTcpServer _tcp_server_ptr; //回指指针
};class TcpServer : public nocpy
{
public:TcpServer(uint16_t port):_port(port), _listensock_ptr(new Socket), _epoller_ptr(new Epoller){}void Init(){_listensock_ptr-CreateSocket();_listensock_ptr-Setsockopt();_listensock_ptr-Bind(_port);_listensock_ptr-SetNonBlock(); //设置非阻塞_listensock_ptr-Listen();_epoller_ptr-EpollerCreate();lg(Info, create listen socket success, socket: %d, _listensock_ptr-Fd());AddConnection(_listensock_ptr-Fd(), EVENT_IN, nullptr, nullptr, nullptr);}void AddConnection(int fd, uint32_t event, func_t recv_cb, func_t send_cb, func_t except_cb){//建立连接的同时挂到epoller中//1.将listensock添加到Connection中同时将listen和Connection放入Connections中std::shared_ptrConnection conn(new Connection(_listensock_ptr-Fd(), this));conn-SetHandler(recv_cb, send_cb, except_cb);_connections.insert(make_pair(_listensock_ptr-Fd(), conn));//2.添加对应的事件放入到epoller中_epoller_ptr-EpollerUpate(EPOLL_CTL_ADD, fd, event);}void Dispatcher(int n){//对比之前我现在还没写Acceptor和Recv函数就已经把Dispatcher函数完成了这就是代码的解耦for (int i 0; i n; i){uint32_t event recvs[i].events;int fd recvs[i].data.fd;auto pos _connections.find(fd);assert(pos ! _connections.end());//统一把事件异常转换成为读写问题这样就只用考虑读和写即可if (event EPOLLERR)event | (EPOLLIN | EPOLLOUT);if (event EPOLLHUP)event | (EPOLLIN | EPOLLOUT);if (event EPOLLIN){if (pos-second-_recv_cb)pos-second-_recv_cb(pos-second);}if (event EPOLLOUT){if (pos-second-_send_cb)pos-second-_send_cb(pos-second);}}}void Start(){while (1){int n _epoller_ptr-Wait(recvs, num, -1); //为了方便观察设置为-1if (n 0){lg(Error, epoll wait failed: %s, strerror(errno));break;}else if (n 0){lg(Info, timeout...);}else {Dispatcher(n);}}}~TcpServer(){}
private:std::shared_ptrSocket _listensock_ptr;std::shared_ptrEpoller _epoller_ptr;struct epoll_event recvs[num]; //捞取就绪事件的数组std::unordered_mapint, std::shared_ptrConnection _connections; //管理所有的连接int _port;
};更新代码 TcpServer.hpp文件
#include Socket.hpp
#include Epoller.hpp
#include nocpy.hpp
#include memory
#include functional
#include unordered_map
#include cassertconst int num 64; //一次性最多捞取上来多少个fdclass Connection;
using func_t functionvoid(std::shared_ptrConnection);
class TcpServer;uint32_t EVENT_IN (EPOLLIN | EPOLLET);
uint32_t EVENT_OUT (EPOLLOUT | EPOLLET);//每一个文件描述符都有一套自己的缓冲区和回调函数每当该文件描述符就绪了就调用自己的回调函数
class Connection
{
public:Connection(int fd, TcpServer* tcp_server_ptr):_sockfd(fd), _tcp_server_ptr(tcp_server_ptr){}void SetHandler(func_t recv_cb, func_t send_cb, func_t except_cb){_recv_cb recv_cb;_send_cb send_cb;_except_cb except_cb;}public:int _sockfd;std::string _inbuffer;std::string _outbuffer;func_t _recv_cb; //读回调函数func_t _send_cb; //写回调函数func_t _except_cb; //异常回调函数std::shared_ptrTcpServer _tcp_server_ptr; //回指指针
};class TcpServer : public nocpy
{
public:TcpServer(uint16_t port):_port(port), _listensock_ptr(new Socket), _epoller_ptr(new Epoller){}void Init(){_listensock_ptr-CreateSocket();_listensock_ptr-Setsockopt();_listensock_ptr-Bind(_port);_listensock_ptr-SetNonBlock(); //设置非阻塞_listensock_ptr-Listen();_epoller_ptr-EpollerCreate();lg(Info, create listen socket success, socket: %d, _listensock_ptr-Fd());AddConnection(_listensock_ptr-Fd(), EVENT_IN, std::bind(TcpServer::Acceptor, this, std::placeholders::_1), nullptr, nullptr);}void AddConnection(int fd, uint32_t event, func_t recv_cb, func_t send_cb, func_t except_cb){//建立连接的同时挂到epoller中//1.将fd添加到Connection中同时将fd和Connection放入Connections中std::shared_ptrConnection conn(new Connection(fd, this));conn-SetHandler(recv_cb, send_cb, except_cb);_connections.insert(make_pair(fd, conn));//2.添加对应的事件放入到epoller中_epoller_ptr-EpollerUpate(EPOLL_CTL_ADD, fd, event);}void Acceptor(std::shared_ptrConnection conn){while (1){std::string clientIp;uint16_t clientPort;Socket newSocket _listensock_ptr-Accept(clientIp, clientPort);if (newSocket.Fd() 0){if (errno EAGAIN || errno EWOULDBLOCK){//如果底层没有数据了break;}else if (errno EINTR){//被信号打断了continue;}lg(Error, listensock accept failed: %s, strerror(errno));break;}lg(Info, get a new socket: %d, clientIp: %s, clientPort: %d, newSocket.Fd(), clientIp.c_str(), clientPort);newSocket.SetNonBlock();AddConnection(newSocket.Fd(), EVENT_IN, nullptr, nullptr, nullptr);}}void Dispatcher(int n){//对比之前我现在还没写Acceptor和Recv函数就已经把Dispatcher函数完成了这就是代码的解耦for (int i 0; i n; i){uint32_t event recvs[i].events;int fd recvs[i].data.fd;auto pos _connections.find(fd);assert(pos ! _connections.end());//统一把事件异常转换成为读写问题这样就只用考虑读和写即可if (event EPOLLERR)event | (EPOLLIN | EPOLLOUT);if (event EPOLLHUP)event | (EPOLLIN | EPOLLOUT);if (event EPOLLIN){if (pos-second-_recv_cb)pos-second-_recv_cb(pos-second);}if (event EPOLLOUT){if (pos-second-_send_cb)pos-second-_send_cb(pos-second);}}}void Start(){while (1){int n _epoller_ptr-Wait(recvs, num, -1); //为了方便观察设置为-1if (n 0){lg(Error, epoll wait failed: %s, strerror(errno));break;}else if (n 0){lg(Info, timeout...);}else {Dispatcher(n);}}}~TcpServer(){}
private:std::shared_ptrSocket _listensock_ptr;std::shared_ptrEpoller _epoller_ptr;struct epoll_event recvs[num]; //捞取就绪事件的数组std::unordered_mapint, std::shared_ptrConnection _connections; //管理所有的连接int _port;
};用三个客户端连接代码运行结果 更新代码 TcpServer.hpp文件
#include Socket.hpp
#include Epoller.hpp
#include nocpy.hpp
#include memory
#include functional
#include unordered_map
#include cassertconst int num 64; //一次性最多捞取上来多少个fdclass Connection;
using func_t functionvoid(std::shared_ptrConnection);
class TcpServer;uint32_t EVENT_IN (EPOLLIN | EPOLLET);
uint32_t EVENT_OUT (EPOLLOUT | EPOLLET);//每一个文件描述符都有一套自己的缓冲区和回调函数每当该文件描述符就绪了就调用自己的回调函数
class Connection
{
public:Connection(int fd, TcpServer* tcp_server_ptr):_sockfd(fd), _tcp_server_ptr(tcp_server_ptr){}void SetHandler(func_t recv_cb, func_t send_cb, func_t except_cb){_recv_cb recv_cb;_send_cb send_cb;_except_cb except_cb;}
public:int _sockfd;std::string _inbuffer;std::string _outbuffer;func_t _recv_cb; //读回调函数func_t _send_cb; //写回调函数func_t _except_cb; //异常回调函数std::shared_ptrTcpServer _tcp_server_ptr; //回指指针std::string _ip;uint16_t _port;
};class TcpServer : public nocpy
{
public:TcpServer(uint16_t port, func_t OnMessage):_port(port), _listensock_ptr(new Socket), _epoller_ptr(new Epoller), _OnMessage(OnMessage){}void Init(){_listensock_ptr-CreateSocket();_listensock_ptr-Setsockopt();_listensock_ptr-Bind(_port);_listensock_ptr-SetNonBlock(); //设置非阻塞_listensock_ptr-Listen();_epoller_ptr-EpollerCreate();lg(Info, create listen socket success, socket: %d, _listensock_ptr-Fd());AddConnection(_listensock_ptr-Fd(), EVENT_IN, std::bind(TcpServer::Acceptor, this, std::placeholders::_1), nullptr, nullptr);}void AddConnection(int fd, uint32_t event, func_t recv_cb, func_t send_cb, func_t except_cb){//建立连接的同时挂到epoller中//1.将fd添加到Connection中同时将fd和Connection放入Connections中std::shared_ptrConnection conn(new Connection(fd, this));conn-SetHandler(recv_cb, send_cb, except_cb);_connections.insert(make_pair(fd, conn));//2.添加对应的事件放入到epoller中_epoller_ptr-EpollerUpate(EPOLL_CTL_ADD, fd, event);}void Excepter(std::shared_ptrConnection conn) //异常事件触发了转为读事件就会调用读函数读的时候肯定会异常直接调用异常函数即可{lg(Warning, Excepter hander sockefd: %d, client info: %s:%d, conn-_sockfd, conn-_ip.c_str(), conn-_port);auto it _connections.find(conn-_sockfd);assert (it ! _connections.end());_connections.erase(it); //因为此时connection被_connections管理着的一旦删了it则connection也会被释放_epoller_ptr-EpollerUpate(EPOLL_CTL_DEL, conn-_sockfd, 0);close(conn-_sockfd);}void Recver(std::shared_ptrConnection conn){int fd conn-_sockfd;char buffer[4096];while (1){int n recv(fd, buffer, sizeof(buffer) - 1, 0);//虽然这里的最后一个参数时阻塞读取但是之前设置了非阻塞所以还是非阻塞读取if (n 0){if (errno EAGAIN || errno EWOULDBLOCK){//如果底层没有数据了break;}else if (errno EINTR){//被信号打断了continue;}lg(Error, read failed: %s, strerror(errno));conn-_except_cb(conn);return;}else if (n 0){lg(Info, server closed...);conn-_except_cb(conn);return;}buffer[n] 0;conn-_inbuffer buffer; //因为read读上来的数据可能不完整所以读后要放到_inbuffer来统一处理}_OnMessage(conn); //统一处理1.数据有了但是不一定全所以要检测数据是否全。2.如果检测到完整数据就处理}void Acceptor(std::shared_ptrConnection conn){while (1){std::string clientIp;uint16_t clientPort;Socket newSocket _listensock_ptr-Accept(clientIp, clientPort);if (newSocket.Fd() 0){if (errno EAGAIN || errno EWOULDBLOCK){//如果底层没有数据了break;}else if (errno EINTR){//被信号打断了continue;}lg(Error, listensock accept failed: %s, strerror(errno));break;}lg(Info, get a new socket: %d, clientIp: %s, clientPort: %d, newSocket.Fd(), clientIp.c_str(), clientPort);newSocket.SetNonBlock();AddConnection(newSocket.Fd(), EVENT_IN, std::bind(TcpServer::Recver, this, std::placeholders::_1),\nullptr, std::bind(TcpServer::Excepter, this, std::placeholders::_1));}}void Dispatcher(int n){//对比之前我现在还没写Acceptor和Recv函数就已经把Dispatcher函数完成了这就是代码的解耦for (int i 0; i n; i){uint32_t event recvs[i].events;int fd recvs[i].data.fd;auto pos _connections.find(fd);assert(pos ! _connections.end());//统一把事件异常转换成为读写问题这样就只用考虑读和写即可if (event EPOLLERR)event | (EPOLLIN | EPOLLOUT);if (event EPOLLHUP)event | (EPOLLIN | EPOLLOUT);if (event EPOLLIN){if (pos-second-_recv_cb)pos-second-_recv_cb(pos-second);}if (event EPOLLOUT){if (pos-second-_send_cb)pos-second-_send_cb(pos-second);}}}void Start(){while (1){int n _epoller_ptr-Wait(recvs, num, -1); //为了方便观察设置为-1if (n 0){lg(Error, epoll wait failed: %s, strerror(errno));break;}else if (n 0){lg(Info, timeout...);}else {Dispatcher(n);}}}~TcpServer(){}
private:std::shared_ptrSocket _listensock_ptr;std::shared_ptrEpoller _epoller_ptr;struct epoll_event recvs[num]; //捞取就绪事件的数组std::unordered_mapint, std::shared_ptrConnection _connections; //管理所有的连接int _port;func_t _OnMessage; //让上层处理信息
};main.cc文件
#include TcpServer.hppvoid DefaultOnMessage(std::shared_ptrConnection conn)
{std::cout sock: conn-_sockfd 的缓冲区里的数据为 conn-_inbuffer std::endl;
}int main()
{std::unique_ptrTcpServer svr(new TcpServer(8080, DefaultOnMessage));svr-Init();svr-Start();return 0;
}代码运行结果如下 但是发现了一个问题类似于循环引用异常事件关闭连接时导致程序出错。 更新代码 #include Socket.hpp
#include Epoller.hpp
#include nocpy.hpp
#include memory
#include functional
#include unordered_map
#include cassertconst int num 64; //一次性最多捞取上来多少个fdclass Connection;
using func_t functionvoid(std::shared_ptrConnection);
class TcpServer;uint32_t EVENT_IN (EPOLLIN | EPOLLET);
uint32_t EVENT_OUT (EPOLLOUT | EPOLLET);//每一个文件描述符都有一套自己的缓冲区和回调函数每当该文件描述符就绪了就调用自己的回调函数
class Connection
{
public:Connection(int fd):_sockfd(fd){}void SetHandler(func_t recv_cb, func_t send_cb, func_t except_cb){_recv_cb recv_cb;_send_cb send_cb;_except_cb except_cb;}
public:int _sockfd;std::string _inbuffer;std::string _outbuffer;func_t _recv_cb; //读回调函数func_t _send_cb; //写回调函数func_t _except_cb; //异常回调函数std::shared_ptrTcpServer _tcp_server_ptr; //回指指针std::string _ip;uint16_t _port;
};class TcpServer : public nocpy, public std::enable_shared_from_thisTcpServer //继承此的作用仅是为了能获取到this对应的shared_ptr
{
public:TcpServer(uint16_t port, func_t OnMessage):_port(port), _listensock_ptr(new Socket), _epoller_ptr(new Epoller), _OnMessage(OnMessage){}void Init(){_listensock_ptr-CreateSocket();_listensock_ptr-Setsockopt();_listensock_ptr-Bind(_port);_listensock_ptr-SetNonBlock(); //设置非阻塞_listensock_ptr-Listen();_epoller_ptr-EpollerCreate();lg(Info, create listen socket success, socket: %d, _listensock_ptr-Fd());AddConnection(_listensock_ptr-Fd(), EVENT_IN, std::bind(TcpServer::Acceptor, this, std::placeholders::_1), nullptr, nullptr);}void AddConnection(int fd, uint32_t event, func_t recv_cb, func_t send_cb, func_t except_cb){//建立连接的同时挂到epoller中//1.将fd添加到Connection中同时将fd和Connection放入Connections中std::shared_ptrConnection conn(new Connection(fd));conn-SetHandler(recv_cb, send_cb, except_cb);_connections.insert(make_pair(fd, conn));if (fd _listensock_ptr-Fd())conn-_tcp_server_ptr shared_ptrTcpServer(this); //如果是listensock连接就构造智能指针来管理TcpServerelseconn-_tcp_server_ptr shared_from_this(); //如果是其他连接就共享该资源来管理//2.添加对应的事件放入到epoller中_epoller_ptr-EpollerUpate(EPOLL_CTL_ADD, fd, event);}void Excepter(std::shared_ptrConnection conn) //异常事件触发了转为读事件就会调用读函数读的时候肯定会异常直接调用异常函数即可{lg(Warning, Excepter hander sockefd: %d, client info: %s:%d, conn-_sockfd, conn-_ip.c_str(), conn-_port);auto it _connections.find(conn-_sockfd);assert (it ! _connections.end());_connections.erase(it); //因为此时connection被_connections管理着的一旦删了it则connection也会被释放_epoller_ptr-EpollerUpate(EPOLL_CTL_DEL, conn-_sockfd, 0);close(conn-_sockfd);}void Recver(std::shared_ptrConnection conn){int fd conn-_sockfd;char buffer[4096];while (1){int n recv(fd, buffer, sizeof(buffer) - 1, 0);//虽然这里的最后一个参数时阻塞读取但是之前设置了非阻塞所以还是非阻塞读取if (n 0){if (errno EAGAIN || errno EWOULDBLOCK){//如果底层没有数据了break;}else if (errno EINTR){//被信号打断了continue;}lg(Error, read failed: %s, strerror(errno));conn-_except_cb(conn);return;}else if (n 0){lg(Info, server closed...);conn-_except_cb(conn);return;}buffer[n] 0;conn-_inbuffer buffer; //因为read读上来的数据可能不完整所以读后要放到_inbuffer来统一处理}_OnMessage(conn); //统一处理1.数据有了但是不一定全所以要检测数据是否全。2.如果检测到完整数据就处理}void Acceptor(std::shared_ptrConnection conn){while (1){std::string clientIp;uint16_t clientPort;Socket newSocket _listensock_ptr-Accept(clientIp, clientPort);if (newSocket.Fd() 0){if (errno EAGAIN || errno EWOULDBLOCK){//如果底层没有数据了break;}else if (errno EINTR){//被信号打断了continue;}lg(Error, listensock accept failed: %s, strerror(errno));break;}lg(Info, get a new socket: %d, clientIp: %s, clientPort: %d, newSocket.Fd(), clientIp.c_str(), clientPort);newSocket.SetNonBlock();AddConnection(newSocket.Fd(), EVENT_IN, std::bind(TcpServer::Recver, this, std::placeholders::_1),\nullptr, std::bind(TcpServer::Excepter, this, std::placeholders::_1));}}void Dispatcher(int n){//对比之前我现在还没写Acceptor和Recv函数就已经把Dispatcher函数完成了这就是代码的解耦for (int i 0; i n; i){uint32_t event recvs[i].events;int fd recvs[i].data.fd;auto pos _connections.find(fd);assert(pos ! _connections.end());//统一把事件异常转换成为读写问题这样就只用考虑读和写即可if (event EPOLLERR)event | (EPOLLIN | EPOLLOUT);if (event EPOLLHUP)event | (EPOLLIN | EPOLLOUT);if (event EPOLLIN){if (pos-second-_recv_cb)pos-second-_recv_cb(pos-second);}if (event EPOLLOUT){if (pos-second-_send_cb)pos-second-_send_cb(pos-second);}}}void Start(){while (1){int n _epoller_ptr-Wait(recvs, num, -1); //为了方便观察设置为-1if (n 0){lg(Error, epoll wait failed: %s, strerror(errno));break;}else if (n 0){lg(Info, timeout...);}else {Dispatcher(n);}}}~TcpServer(){}
private:std::shared_ptrSocket _listensock_ptr;std::shared_ptrEpoller _epoller_ptr;struct epoll_event recvs[num]; //捞取就绪事件的数组std::unordered_mapint, std::shared_ptrConnection _connections; //管理所有的连接int _port;func_t _OnMessage; //让上层处理信息
};运行结果如下 每一个socket都配套有一个读写缓冲区为什么要有读缓冲区因为你不能保证一次read上来的数据就是一个完整的数据。为什么要有写缓冲区因为你不能保证内核缓冲区是有空间的也就是你无法保证发送条件是否就绪。 异常和读事件我们都处理了如何处理写事件呢 通常情况下发送缓冲区一般都是有空间的写事件一般都是就绪的如果我们设置对EPOLLOUT关心那EPOLLOUT几乎每次都是就绪会导致epoll_wait经常返回浪费CPU资源。 结论对于读事件设置常关心对于写事件按需设置。什么是按需设置直接写入如果写入完成就结束。如果没有将这一轮的数据写完outbuffer里还有数据我们就需要对写事件进行关心了如果写完了就去掉对写事件的关心。 更新代码 TcpServer.hpp文件
#include Socket.hpp
#include Epoller.hpp
#include nocpy.hpp
#include memory
#include functional
#include unordered_map
#include cassertconst int num 64; //一次性最多捞取上来多少个fdclass Connection;
using func_t functionvoid(std::shared_ptrConnection);
class TcpServer;uint32_t EVENT_IN (EPOLLIN | EPOLLET);
uint32_t EVENT_OUT (EPOLLOUT | EPOLLET);//每一个文件描述符都有一套自己的缓冲区和回调函数每当该文件描述符就绪了就调用自己的回调函数
class Connection
{
public:Connection(int fd):_sockfd(fd){}void SetHandler(func_t recv_cb, func_t send_cb, func_t except_cb){_recv_cb recv_cb;_send_cb send_cb;_except_cb except_cb;}
public:int _sockfd;std::string _inbuffer;std::string _outbuffer;func_t _recv_cb; //读回调函数func_t _send_cb; //写回调函数func_t _except_cb; //异常回调函数std::shared_ptrTcpServer _tcp_server_ptr; //回指指针std::string _ip;uint16_t _port;
};class TcpServer : public nocpy, public std::enable_shared_from_thisTcpServer //继承此的作用仅是为了能获取到this对应的shared_ptr
{
public:TcpServer(uint16_t port, func_t OnMessage):_port(port), _listensock_ptr(new Socket), _epoller_ptr(new Epoller), _OnMessage(OnMessage){}void Init(){_listensock_ptr-CreateSocket();_listensock_ptr-Setsockopt();_listensock_ptr-Bind(_port);_listensock_ptr-SetNonBlock(); //设置非阻塞_listensock_ptr-Listen();_epoller_ptr-EpollerCreate();lg(Info, create listen socket success, socket: %d, _listensock_ptr-Fd());AddConnection(_listensock_ptr-Fd(), EVENT_IN, std::bind(TcpServer::Acceptor, this, std::placeholders::_1), nullptr, nullptr);}void AddConnection(int fd, uint32_t event, func_t recv_cb, func_t send_cb, func_t except_cb){//建立连接的同时挂到epoller中//1.将fd添加到Connection中同时将fd和Connection放入Connections中std::shared_ptrConnection conn(new Connection(fd));conn-SetHandler(recv_cb, send_cb, except_cb);_connections.insert(make_pair(fd, conn));if (fd _listensock_ptr-Fd()) //如果是listensock连接就构造智能指针来管理TcpServerconn-_tcp_server_ptr shared_ptrTcpServer(this);elseconn-_tcp_server_ptr shared_from_this(); //如果是其他连接就共享该资源来管理//2.添加对应的事件放入到epoller中_epoller_ptr-EpollerUpate(EPOLL_CTL_ADD, fd, event);}void Sender(std::shared_ptrConnection conn){while (1){int n write(conn-_sockfd, conn-_outbuffer.c_str(), conn-_outbuffer.size());if (n 0){conn-_outbuffer.erase(0, n);if (conn-_outbuffer.empty() true){return;}}else if (n 0){return;}else{if (errno EWOULDBLOCK || errno EAGAIN){//说明底层数据不就绪即内核写缓冲区没有空间了break;}if (errno EINTR){continue;}lg(Error, write failed, socket: %d, conn-_sockfd);conn-_except_cb(conn);return;}}if (!conn-_outbuffer.empty()){//说明没有将数据发完则需要设置写事件关心了uint32_t event EVENT_OUT | EVENT_IN;_epoller_ptr-EpollerUpate(EPOLL_CTL_MOD, conn-_sockfd, event);}else {//说明将数据发完了则取消对写事件关心了uint32_t event EVENT_IN;_epoller_ptr-EpollerUpate(EPOLL_CTL_MOD, conn-_sockfd, event);}}void Excepter(std::shared_ptrConnection conn) //异常事件触发了转为读事件就会调用读函数读的时候肯定会异常直接调用异常函数即可{lg(Warning, Excepter hander sockefd: %d, client info: %s:%d, conn-_sockfd, conn-_ip.c_str(), conn-_port);auto it _connections.find(conn-_sockfd);assert (it ! _connections.end());_connections.erase(it); //因为此时connection被_connections管理着的一旦删了it则connection也会被释放_epoller_ptr-EpollerUpate(EPOLL_CTL_DEL, conn-_sockfd, 0);close(conn-_sockfd);}void Recver(std::shared_ptrConnection conn){int fd conn-_sockfd;char buffer[4096];while (1){int n recv(fd, buffer, sizeof(buffer) - 1, 0);//虽然这里的最后一个参数时阻塞读取但是之前设置了非阻塞所以还是非阻塞读取if (n 0){if (errno EAGAIN || errno EWOULDBLOCK){//如果底层没有数据了break;}else if (errno EINTR){//被信号打断了continue;}lg(Error, read failed: %s, strerror(errno));conn-_except_cb(conn);return;}else if (n 0){lg(Info, server closed...);conn-_except_cb(conn);return;}buffer[n] 0;conn-_inbuffer buffer; //因为read读上来的数据可能不完整所以读后要放到_inbuffer来统一处理}_OnMessage(conn); //统一处理1.数据有了但是不一定全所以要检测数据是否全。2.如果检测到完整数据就处理}void Acceptor(std::shared_ptrConnection conn){while (1){std::string clientIp;uint16_t clientPort;Socket newSocket _listensock_ptr-Accept(clientIp, clientPort);if (newSocket.Fd() 0){if (errno EAGAIN || errno EWOULDBLOCK){//如果底层没有数据了break;}else if (errno EINTR){//被信号打断了continue;}lg(Error, listensock accept failed: %s, strerror(errno));break;}lg(Info, get a new socket: %d, clientIp: %s, clientPort: %d, newSocket.Fd(), clientIp.c_str(), clientPort);newSocket.SetNonBlock();AddConnection(newSocket.Fd(), EVENT_IN, std::bind(TcpServer::Recver, this, std::placeholders::_1),\std::bind(TcpServer::Sender, this, std::placeholders::_1),\std::bind(TcpServer::Excepter, this, std::placeholders::_1));}}void Dispatcher(int n){//对比之前我现在还没写Acceptor和Recv函数就已经把Dispatcher函数完成了这就是代码的解耦for (int i 0; i n; i){uint32_t event recvs[i].events;int fd recvs[i].data.fd;auto pos _connections.find(fd);assert(pos ! _connections.end());//统一把事件异常转换成为读写问题这样就只用考虑读和写即可if (event EPOLLERR)event | (EPOLLIN | EPOLLOUT);if (event EPOLLHUP)event | (EPOLLIN | EPOLLOUT);if (event EPOLLIN){if (pos-second-_recv_cb)pos-second-_recv_cb(pos-second);}if (event EPOLLOUT){if (pos-second-_send_cb)pos-second-_send_cb(pos-second);}}}void Start(){while (1){int n _epoller_ptr-Wait(recvs, num, -1); //为了方便观察设置为-1if (n 0){lg(Error, epoll wait failed: %s, strerror(errno));break;}else if (n 0){lg(Info, timeout...);}else {Dispatcher(n);}}}~TcpServer(){}
private:std::shared_ptrSocket _listensock_ptr;std::shared_ptrEpoller _epoller_ptr;struct epoll_event recvs[num]; //捞取就绪事件的数组std::unordered_mapint, std::shared_ptrConnection _connections; //管理所有的连接int _port;func_t _OnMessage; //让上层处理信息
};main.cc文件
#include TcpServer.hppstd::string Handle(std::string inbuffer) //业务处理
{std::string tmp inbuffer;inbuffer.clear();return tmp;
}
void DefaultOnMessage(std::shared_ptrConnection conn)
{string response_str Handle(conn-_inbuffer);if (response_str.empty()) return;conn-_outbuffer response_str;if (conn-_send_cb)conn-_send_cb(conn);
}int main()
{std::unique_ptrTcpServer svr(new TcpServer(8080, DefaultOnMessage));svr-Init();svr-Start();return 0;
}代码运行结果如下 套用之前写的网络版本计算器业务逻辑 - 文末附完整代码 代码运行结果 reactor模型是半同步半异步的模型 半同步体现的是等由epoll来做的保证了就绪事件的通知和进行IO 半异步的体现负责了业务处理如果把业务处理放到线程池中去处理就是半异步了。因为我们的业务处理并不耗时所以就没有开多线程 还有一种模式叫做Proactor纯异步的编写方式在linux服务器设计里面没有我们就不涉及了。 reactor也叫做反应堆是什么意思呢如同打地鼠的游戏 玩游戏的人就相当于是一个多路转接我们要检测每一个洞口有没有地鼠出来虽然没出来但是我们知道一旦出来了我就要执行我的回调方法来砸它 – 回调函数是提前设置好的。 游戏的面板就相当于我们的reactor每一个洞就相当于Connection老鼠上来了就叫做事件就绪执行砸方法就是执行回调函数。这种就叫做反应堆 redis底层用的就是单reactor处理用的是reactor的LT模式 完整版代码如下 文件目录
Epoller.hpp文件
#include sys/epoll.h
#include Log.hpp
#include cstringclass Epoller
{
public:Epoller(){}void EpollerCreate(){_epollfd epoll_create(64);if (_epollfd 0){lg(Fatal, Create epoll failed: %s, strerror(errno));abort();}}void EpollerUpate(int oper, int fd, uint32_t event){int n 0;if (oper EPOLL_CTL_ADD || oper EPOLL_CTL_MOD){struct epoll_event epEvent;epEvent.data.fd fd;epEvent.events 0;epEvent.events | event;n epoll_ctl(_epollfd, oper, fd, epEvent);if (n 0){lg(Error, epoll ctl failed: %s, strerror(errno));}}else if (oper EPOLL_CTL_DEL){n epoll_ctl(_epollfd, oper, fd, 0);if (n 0){lg(Error, epoll ctl failed: %s, strerror(errno));}}}int Wait(struct epoll_event* events, int n, int timeout){int m epoll_wait(_epollfd, events, n, timeout);if (m 0){lg(Error, epoll wait failed: %s, strerror(errno));}else if (m 0){//超时返回非阻塞返回lg(Info, timeout...);}return m;}int Fd(){return _epollfd;}
private:int _epollfd;
};Log.hpp文件
#pragma once
#include iostream
#include string
#include stdarg.h
#include time.h
#include sys/types.h
#include sys/stat.h
#include fcntl.h
#include unistd.husing namespace std;#define Info 0
#define Debug 1
#define Warning 2
#define Error 3
#define Fatal 4#define Screen 1
#define Onefile 2
#define Classfile 3#define fileName log.txt
//使用前需要创建log目录
class Log
{
public:Log(){printMethod Screen;path ./log/;}void Enable(int method){printMethod method;}void printOneFile(string logname, const string logtxt){logname path logname;int fd open(logname.c_str(), O_WRONLY | O_APPEND | O_CREAT, 0666);//open只会创建文件不会创建目录if (fd 0){perror(open failed);return;}write(fd, logtxt.c_str(), logtxt.size());close(fd);}void printClassFile(int level, const string logtxt){string filename fileName;filename .;filename leveltoString(level);printOneFile(filename, logtxt);}void printLog(int level, const string logtxt){if (printMethod Screen){cout logtxt endl;return;}else if (printMethod Onefile){printOneFile(fileName, logtxt);return;}else if (printMethod Classfile){printClassFile(level, logtxt);return;}}const char* leveltoString(int level){if (level Info) return Info;else if (level Debug) return Debug;else if (level Error) return Error;else if (level Fatal) return Fatal;else return default;}void operator()(int level, const char* format, ...){time_t t time(nullptr);struct tm* st localtime(t);char leftbuffer[4096];snprintf(leftbuffer, sizeof(leftbuffer), year: %d, month: %d, day: %d, hour: %d, minute: %d, second: %d\n\[%s]:,st-tm_year 1900, st-tm_mon 1, st-tm_mday, st-tm_hour, st-tm_min, st-tm_sec, leveltoString(level));char rightbuffer[4096];va_list start;va_start(start, format);vsnprintf(rightbuffer, sizeof(rightbuffer), format, start);va_end(start);char logtxt[4096 * 2];snprintf(logtxt, sizeof(logtxt), %s %s\n, leftbuffer, rightbuffer);printLog(level, logtxt);}
private:int printMethod;string path;//路径与文件名解耦最后将路径和文件粘合起来再用open打开即可
};nocpy.hpp文件
#pragma once
//编程技巧写防拷贝的类的时候直接继承就不用自己再手动写了
class nocpy
{
public:nocpy(){}~nocpy(){}
private:nocpy(const nocpy) delete;nocpy operator(const nocpy) delete;
};Socket.hpp文件
#pragma once
#include sys/types.h
#include sys/socket.h
#include arpa/inet.h
#include cstring
#include Log.hppLog lg;
class Socket
{
public:Socket(int fd -1):_sockfd(fd){}~Socket(){}void CreateSocket(){_sockfd socket(AF_INET, SOCK_STREAM, 0);if (_sockfd 0){lg(Fatal, create socket failed:%s, strerror(errno));exit(1);}}void Bind(uint16_t serverPort){struct sockaddr_in server;server.sin_family AF_INET;server.sin_port htons(serverPort);server.sin_addr.s_addr INADDR_ANY;int n bind(_sockfd, (struct sockaddr*)server, sizeof(server));if (n 0){lg(Fatal, bind socket failed:%s, strerror(errno));exit(2);}}void Listen(){int n listen(_sockfd, 5);if (n 0){lg(Fatal, listen socket failed:%s, strerror(errno));exit(3);}}int Accept(string* clientIp, uint16_t* clinetPort){struct sockaddr_in peer;socklen_t len sizeof(peer);int newsocket accept(_sockfd, (struct sockaddr*)peer, len);if (newsocket 0){return -1;}*clientIp inet_ntoa(peer.sin_addr);*clinetPort ntohs(peer.sin_port);return newsocket;}int Connect(const string serverIp, uint16_t serverPort){struct sockaddr_in server;server.sin_family AF_INET;server.sin_addr.s_addr inet_addr(serverIp.c_str());server.sin_port htons(serverPort);int n connect(_sockfd, (struct sockaddr*)server, sizeof(server));if (n 0){return -1;}return 0;}void Setsockopt(){int opt 1;if (setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, (const void*)opt, sizeof(opt)) 0)lg(Error, %s\n, strerror(errno));}void SetNonBlock(){int flag fcntl(_sockfd, F_GETFL);fcntl(_sockfd, F_SETFL, flag | O_NONBLOCK);}void Close(){close(_sockfd);}int Fd(){return _sockfd;}
private:int _sockfd;
};TcpServer.hpp文件如下
#include Socket.hpp
#include Epoller.hpp
#include nocpy.hpp
#include memory
#include functional
#include unordered_map
#include cassertconst int num 64; //一次性最多捞取上来多少个fdclass Connection;
using func_t functionvoid(std::shared_ptrConnection);
class TcpServer;uint32_t EVENT_IN (EPOLLIN | EPOLLET);
uint32_t EVENT_OUT (EPOLLOUT | EPOLLET);//每一个文件描述符都有一套自己的缓冲区和回调函数每当该文件描述符就绪了就调用自己的回调函数
class Connection
{
public:Connection(int fd):_sockfd(fd){}void SetHandler(func_t recv_cb, func_t send_cb, func_t except_cb){_recv_cb recv_cb;_send_cb send_cb;_except_cb except_cb;}
public:int _sockfd;std::string _inbuffer;std::string _outbuffer;func_t _recv_cb; //读回调函数func_t _send_cb; //写回调函数func_t _except_cb; //异常回调函数std::shared_ptrTcpServer _tcp_server_ptr; //回指指针std::string _ip;uint16_t _port;
};class TcpServer : public nocpy, public std::enable_shared_from_thisTcpServer //继承此的作用仅是为了能获取到this对应的shared_ptr
{
public:TcpServer(uint16_t port, func_t OnMessage):_port(port), _listensock_ptr(new Socket), _epoller_ptr(new Epoller), _OnMessage(OnMessage){}void Init(){_listensock_ptr-CreateSocket();_listensock_ptr-Setsockopt();_listensock_ptr-Bind(_port);_listensock_ptr-SetNonBlock(); //设置非阻塞_listensock_ptr-Listen();_epoller_ptr-EpollerCreate();lg(Info, create listen socket success, socket: %d, _listensock_ptr-Fd());AddConnection(_listensock_ptr-Fd(), EVENT_IN, std::bind(TcpServer::Acceptor, this, std::placeholders::_1), nullptr, nullptr);}void AddConnection(int fd, uint32_t event, func_t recv_cb, func_t send_cb, func_t except_cb){//建立连接的同时挂到epoller中//1.将fd添加到Connection中同时将fd和Connection放入Connections中std::shared_ptrConnection conn(new Connection(fd));conn-SetHandler(recv_cb, send_cb, except_cb);_connections.insert(make_pair(fd, conn));if (fd _listensock_ptr-Fd()) //如果是listensock连接就构造智能指针来管理TcpServerconn-_tcp_server_ptr shared_ptrTcpServer(this);elseconn-_tcp_server_ptr shared_from_this(); //如果是其他连接就共享该资源来管理//2.添加对应的事件放入到epoller中_epoller_ptr-EpollerUpate(EPOLL_CTL_ADD, fd, event);}void Sender(std::shared_ptrConnection conn){while (1){int n write(conn-_sockfd, conn-_outbuffer.c_str(), conn-_outbuffer.size());if (n 0){conn-_outbuffer.erase(0, n);if (conn-_outbuffer.empty() true){return;}}else if (n 0){return;}else{if (errno EWOULDBLOCK || errno EAGAIN){//说明底层数据不就绪即内核写缓冲区没有空间了break;}if (errno EINTR){continue;}lg(Error, write failed, socket: %d, conn-_sockfd);conn-_except_cb(conn);return;}}if (!conn-_outbuffer.empty()){//说明没有将数据发完则需要设置写事件关心了uint32_t event EVENT_OUT | EVENT_IN;_epoller_ptr-EpollerUpate(EPOLL_CTL_MOD, conn-_sockfd, event);}else {//说明将数据发完了则取消对写事件关心了uint32_t event EVENT_IN;_epoller_ptr-EpollerUpate(EPOLL_CTL_MOD, conn-_sockfd, event);}}void Excepter(std::shared_ptrConnection conn) //异常事件触发了转为读事件就会调用读函数读的时候肯定会异常直接调用异常函数即可{lg(Warning, Excepter hander sockefd: %d, client info: %s:%d, conn-_sockfd, conn-_ip.c_str(), conn-_port);auto it _connections.find(conn-_sockfd);assert (it ! _connections.end());_connections.erase(it); //因为此时connection被_connections管理着的一旦删了it则connection也会被释放_epoller_ptr-EpollerUpate(EPOLL_CTL_DEL, conn-_sockfd, 0);close(conn-_sockfd);}void Recver(std::shared_ptrConnection conn){int fd conn-_sockfd;char buffer[4096];while (1){int n recv(fd, buffer, sizeof(buffer) - 1, 0);//虽然这里的最后一个参数时阻塞读取但是之前设置了非阻塞所以还是非阻塞读取if (n 0){if (errno EAGAIN || errno EWOULDBLOCK){//如果底层没有数据了break;}else if (errno EINTR){//被信号打断了continue;}lg(Error, read failed: %s, strerror(errno));conn-_except_cb(conn);return;}else if (n 0){lg(Info, server closed...);conn-_except_cb(conn);return;}buffer[n] 0;conn-_inbuffer buffer; //因为read读上来的数据可能不完整所以读后要放到_inbuffer来统一处理}_OnMessage(conn); //统一处理1.数据有了但是不一定全所以要检测数据是否全。2.如果检测到完整数据就处理}void Acceptor(std::shared_ptrConnection conn){while (1){std::string clientIp;uint16_t clientPort;Socket newSocket _listensock_ptr-Accept(clientIp, clientPort);if (newSocket.Fd() 0){if (errno EAGAIN || errno EWOULDBLOCK){//如果底层没有数据了break;}else if (errno EINTR){//被信号打断了continue;}lg(Error, listensock accept failed: %s, strerror(errno));break;}lg(Info, get a new socket: %d, clientIp: %s, clientPort: %d, newSocket.Fd(), clientIp.c_str(), clientPort);newSocket.SetNonBlock();AddConnection(newSocket.Fd(), EVENT_IN, std::bind(TcpServer::Recver, this, std::placeholders::_1),\std::bind(TcpServer::Sender, this, std::placeholders::_1),\std::bind(TcpServer::Excepter, this, std::placeholders::_1));}}void Dispatcher(int n){//对比之前我现在还没写Acceptor和Recv函数就已经把Dispatcher函数完成了这就是代码的解耦for (int i 0; i n; i){uint32_t event recvs[i].events;int fd recvs[i].data.fd;auto pos _connections.find(fd);assert(pos ! _connections.end());//统一把事件异常转换成为读写问题这样就只用考虑读和写即可if (event EPOLLERR)event | (EPOLLIN | EPOLLOUT);if (event EPOLLHUP)event | (EPOLLIN | EPOLLOUT);if (event EPOLLIN){if (pos-second-_recv_cb)pos-second-_recv_cb(pos-second);}if (event EPOLLOUT){if (pos-second-_send_cb)pos-second-_send_cb(pos-second);}}}void Start(){while (1){int n _epoller_ptr-Wait(recvs, num, -1); //为了方便观察设置为-1if (n 0){lg(Error, epoll wait failed: %s, strerror(errno));break;}else if (n 0){lg(Info, timeout...);}else {Dispatcher(n);}}}~TcpServer(){}
private:std::shared_ptrSocket _listensock_ptr;std::shared_ptrEpoller _epoller_ptr;struct epoll_event recvs[num]; //捞取就绪事件的数组std::unordered_mapint, std::shared_ptrConnection _connections; //管理所有的连接int _port;func_t _OnMessage; //让上层处理信息
};main.cc文件
#include TcpServer.hpp
#include ServerCal.hppvoid DefaultOnMessage(std::shared_ptrConnection conn)
{ServerCal calculator;std::string response_str calculator.Calculator(conn-_inbuffer);if (response_str.empty()) return;conn-_outbuffer response_str;if (conn-_send_cb)conn-_send_cb(conn);
}int main()
{std::unique_ptrTcpServer svr(new TcpServer(8080, DefaultOnMessage));svr-Init();svr-Start();return 0;
}Protocol.hpp文件
#pragma once
#include iostream
#include string
#include jsoncpp/json/json.husing namespace std;const string blank_space_sep ;
const string protocol_sep \n;
//添加报头内容 转为 长度\n内容\n
bool Encode(string *package, const string content)
{package-clear();*package to_string(content.size());*package protocol_sep;*package content;*package protocol_sep;return true;
}
//去除报头长度\n内容\n 转为 内容 -- 可能存在本来是5\n1 1\n 但接受到的是5\n1
bool Decode(string package, string *content)
{content-clear();int pos package.find(protocol_sep);if (pos string::npos) return false;int len stoi(package.substr(0, pos));int totalLen pos len 2;if (package.size() totalLen) return false;//如果报文不完整说明*content package.substr(pos protocol_sep.size(), len);package.erase(0, totalLen);return true;
}//提取字符串
bool Extract(const string in, int x, int y, char oper)
{int pos in.find(blank_space_sep);if (pos string::npos) return false;x stoi(in.substr(0, pos));oper *in.substr(pos blank_space_sep.size(), 1).c_str();int pos2 in.rfind(blank_space_sep);if (pos2 string::npos) return false;y stoi(in.substr(pos2 blank_space_sep.size()));return true;
}//请求协议
class Request
{
public:Request(int x 0, int y 0, char oper ):_x(x), _y(y), _oper(oper){}//序列化 -- 将结构体转为_x _ybool Serialize(string* out){
#ifdef MySelfout-clear();*out to_string(_x);*out blank_space_sep;*out _oper;*out blank_space_sep;*out to_string(_y);return true;
#elseJson::Value root;root[x] _x;root[y] _y;root[oper] _oper;Json::StyledWriter w;*out w.write(root);return true;
#endif}//反序列化 -- 将_x _y转为结构体bool Deserialize(const string in){
#ifdef MySelfreturn Extract(in, _x, _y, _oper);
#elseJson::Reader r;Json::Value root;r.parse(in, root);_x root[x].asInt();_y root[y].asInt();_oper root[oper].asInt();return true;
#endif}
public:int _x;int _y;char _oper;
};//响应协议
class Response
{const string blank_space_sep ;
public:Response(int result 0, int code 0):_result(), _code(code){}//序列化 -- 将结构体转为_result _codebool Serialize(string* out){
#ifdef MySelf*out to_string(_result) blank_space_sep to_string(_code);return true;
#else Json::Value root;root[code] _code;root[result] _result;Json::StyledWriter w;*out w.write(root);return true;
#endif}//反序列化 -- 将_result _code转为结构体bool Deserialize(const string in){
#ifdef MySelfint pos in.find(blank_space_sep);if (pos string::npos) return false;_result stoi(in.substr(0, pos));_code stoi(in.substr(pos blank_space_sep.size()));return true;
#elseJson::Reader r;Json::Value root;r.parse(in, root);_result root[result].asInt();_code root[code].asInt();return true;
#endif}
public:int _result;int _code; // 0可信否则!0具体是几表明对应的错误原因
};ServerCal.hpp文件
#pragma once
#include Protocol.hpp//服务器的计算服务
class ServerCal
{
public:ServerCal(){}Response CalculatorHelper(const Request req){int x req._x;char oper req._oper;int y req._y;Response rsp(0, 0);switch (oper){case :{rsp._result x y;rsp._code 0;break;}case -:{rsp._result x - y;rsp._code 0;break;}case *:{rsp._result x * y;rsp._code 0;break;}case /:{if (y 0){rsp._result 0;rsp._code -1;break;}rsp._result x / y;rsp._code 0;break;}case %:{if (y 0){rsp._result 0;rsp._code -1;break;}rsp._result x % y;rsp._code 0;break;}default:break;}return rsp;}string Calculator(string s){string content;if(!Decode(s, content))//将长度/n内容/n - 内容return ;Request rq;rq.Deserialize(content);//将内容反序列化Response rsp CalculatorHelper(rq);//得到答案内容content.clear();rsp.Serialize(content);//序列化答案内容string ret;Encode(ret, content);//将内容 - 长度/n内容/nreturn ret;}
};TcpClient.cc文件
#include Protocol.hpp
#include Socket.hppstatic void Usage(const std::string proc)
{std::cout \nUsage: proc \tserverIp\tport\n std::endl;
}
int main(int argc, char* argv[])
{if (argc ! 3){Usage(argv[0]);exit(0);}string serverIp argv[1];uint16_t serverPort atoi(argv[2]);Socket skt;skt.CreateSocket();skt.Connect(serverIp, serverPort);string streamBuffer;while (1){cout Enter#;fflush(stdout);string content;getline(cin, content);//提取字符串得到”请求反序列化“Request rq;Extract(content, rq._x, rq._y, rq._oper);string s;rq.Serialize(s);string ret;Encode(ret, s);write(skt.Fd(), ret.c_str(), ret.size());char readBuffer[4096];int n read(skt.Fd(), readBuffer, sizeof(readBuffer) - 1);if (n 0){readBuffer[n] 0;streamBuffer readBuffer;string ret;Decode(streamBuffer, ret);Response rsp;rsp.Deserialize(ret);cout code: rsp._code result: rsp._result endl;}else if (n 0){cerr Server closed... endl;break;}else {cerr read failed: strerror(errno) endl;exit(3);}}return 0;
}Makefile文件
1main
.PHONY:all
all:$1 TcpClient
$1:$1.ccg -o $ $^ -stdc11 -ljsoncpp
TcpClient:TcpClient.ccg -o $ $^ -stdc11 -ljsoncpp
.PHONY:clean
clean:rm -rf $1 TcpClient
