Linux——五种IO模型

一IO基本理解

二五种IO模型

1五种IO模型示意图

2同步IO和异步IO

二非阻塞IO

1fcntl

2实现非阻塞IO

三多路复用

1select

1.1定位和作用

1.2介绍参数

1.3编写多路复用代码

1.4优缺点

2poll

2.1作用和定位

2.2介绍参数

2.3修改select代码

3epoll

3.1介绍参数

3.2工作原理

理解数据到达主机

3.3编写epoll代码

3.4优点

3.5工作模式

理解LT和ET

四Reactor

完整源代码

一IO基本理解

前面学习了网络通信，我们认识到：网络通信本质上是进程间通信；而进程间通信本质上是IO；关于IO：I:Input；O：Output；这又是什么意思？

站在进程角度上：Input是外面交给进程的数据；Output是进程发出去的数据；站在系统角度上：Input是数据从硬件交给OS；Output是数据从OS输出出去；站在内存角度上：Input是内存数据交给磁盘；Output是磁盘数据交给内存

总之：IO本质上是外设与内设之间的交互

那对应我们来说：IO该如何理解呢？？

C/C++解除到的IO接口：read/recv write/send 当底层有新数据到来时(前提接收/发送缓冲区未被占满的情况下)，这些接口才能给我们返回fd：让我们读/写数据，否则会一直阻塞下去知道新数据的到来！

所以说：IO = 等 + 拷贝

那什么叫做高效的IO呢？<-> 在单位时间内，减少IO等的比率

二五种IO模型

在现实生活中，有很多人喜欢钓鱼，也被我们称之为钓鱼佬；它们钓鱼方式都是按照自己对钓鱼的理解去实现的，所以也就产生出不同的钓鱼方式；这也就总共可分为以下五种：(钓鱼 = 等 + 钓)

张三：一动不动，检测鱼漂，钓；（阻塞IO）

李四：一直在动，随便检测鱼漂，钓；（非阻塞IO）

王五：在鱼漂上挂铃铛(铃铛没响就刷抖音)，钓；（信号驱动IO）

赵六：准备很多钓鱼竿，定期检测鱼竿，钓；（多路复用IO）

田七：找司机小王钓，自己去做别的事（田七没有参与调用，只是发起钓鱼）；（异步IO）

在以上例子中：

人：系统调用；

鱼竿：sockfd；

湖：系统内部；

鱼：数据；

鱼漂浮动：数据就绪；

钓：接收数据

田七：发起IO 小王：操作系统

通过上面的例子可以基本对应5中IO模型，但又有问题了：

阻塞IO vs 非阻塞IO

IO = 等 + 拷贝：拷贝(钓)的方式是一样的，就是等的方式不同(一个一直等，一个一直动)

谁的钓鱼效率最高呢？（单位时间内等待的时间短）

你可以会说是王五或者田七：但是答案是赵六！因为如果你是鱼的话，湖面上有104个诱饵(光赵六就有100个)，你吃到赵六的鱼饵的概率高，也就是赵六更容易有鱼咬钩，自然比别人效率高，而王五田七他们只有一个鱼竿(假设与咬钩概率相等)，一天内钓上来鱼的数量是差不多的，只是多做了一些事情(多刷了几个视频，多参加了几个会(对于别人来说))

1五种IO模型示意图

阻塞 IO(常见): 在内核将数据准备好之前, 系统调用会一直等待；所有的套接字，默认都是阻塞方式

非阻塞 IO: 如果内核还未将数据准备好, 系统调用仍然会直接返回, 并且返回错误码

信号驱动 IO: 内核将数据准备好的时候, 使用 SIGIO 信号通知应用程序进行 IO操作

IO 多路转接: 虽然从流程图上看起来和阻塞 IO 类似. 实际上最核心在于 IO 多路转接能够同时等待多个文件描述符的就绪状态

异步 IO: 由内核在数据拷贝完成时, 通知应用程序(而信号驱动是告诉应用程序何时可以开始拷贝数据)

2同步IO和异步IO

我们也将IO模型分为两大类：同步IO(前四种IO模型)和异步IO

之前也有人为了信号驱动IO是同步还是异步争吵过，但这里把信号驱动IO归结为同步IO：看似王五在做别的事，但是鱼咬钩你是不是要去把鱼给钓上来呢？实际上还是受到IO影响(刷着视频突然鱼咬钩了你要立刻放下手机去钓鱼)！

两者区别：你有没有参与IO(等)的过程

二非阻塞IO

非阻塞 IO 往往需要程序员循环的方式反复尝试读写文件描述符, 这个过程称为轮询：这
对 CPU 来说是较大的浪费, 一般只有特定场景下才使用

1fcntl

作用：将fd设置成非阻塞IO

函数的第二个参数：

• 复制一个现有的描述符（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).

2实现非阻塞IO

#include <iostream>
#include <sys/fcntl.h>
#include <string>
#include <unistd.h>// 设置非阻塞轮询方式
void NoBlock(int fd)
{int f1 = fcntl(fd, F_GETFL);if (f1 < 0){std::cerr << "fcntl fail" << std::endl;return;}fcntl(fd, F_SETFL, f1 | O_NONBLOCK);
}int main()
{char buffer[128];NoBlock(0);while (true){ssize_t n = read(0, buffer, sizeof(buffer) - 1);if (n > 0){buffer[n] = 0;std::cout << "echo:" << buffer;}else if (n == 0){std::cout << "read none" << std::endl;break;}else{// 如何区别read是出错了还是非阻塞了？ -> 通过errno来判断！if (errno == EWOULDBLOCK){std::cout << "Non blocking polling" << std::endl;std::cout << "Do other things" << std::endl;sleep(1);continue;}else{std::cout << "read error" << std::endl;break;}}}return 0;
}

现象：

为什么我输入的时候信息会与打印的信息混杂在一起？--> OS(默认)回显

如果read被信号中断了，循环就会结束，不往下进行读了：我们也要保证该情况不会影响我们读取

#include <iostream>
#include <sys/fcntl.h>
#include <string>
#include <unistd.h>// 设置非阻塞轮询方式
void NoBlock(int fd)
{int f1 = fcntl(fd, F_GETFL);if (f1 < 0){std::cerr << "fcntl fail" << std::endl;return;}fcntl(fd, F_SETFL, f1 | O_NONBLOCK);
}int main()
{char buffer[128];NoBlock(0);while (true){ssize_t n = read(0, buffer, sizeof(buffer) - 1);if (n > 0){buffer[n] = 0;std::cout << "echo:" << buffer;}else if (n == 0)//ctrl+d{std::cout << "read none" << std::endl;break;}else{// 如何区别read是出错了还是非阻塞了？ -> 通过errno来判断！if (errno == EWOULDBLOCK){std::cout << "Non blocking polling" << std::endl;std::cout << "Do other things" << std::endl;sleep(1);continue;//未来是要break的！}else if(errno==EINTR)//read被信号中断{continue;}else{std::cout << "read error" << std::endl;break;}}}return 0;
}

三多路复用

IO多路复用（Input/Output Multiplexing）是一种在单个线程中管理多个输入/输出通道的技术。它允许一个线程同时监听多个输入流（例如sockfd、fd等），并在有数据可读或可写时进行相应的处理，而不需要为每个通道创建一个独立的线程。

1select

系统提供 select 函数来实现多路复用输入/输出模型

1.1定位和作用

定位：它只负责进行等，不负责IO

作用：等待多个fd(等该fd上面的新事件就绪)，通知用户新事件已经就绪，可以来IO了

1.2介绍参数

1.3编写多路复用代码

使用select要注意以下细节：

1.select后accept还是阻塞吗？

一定不会阻塞(事件已经就绪)

2.read能直接读吗？如果不能，谁最清楚底层fd的数据就绪了？

绝对不能读，读取的时候，条件不一定满足；由select来统一监管(select知道)

3.select要正常工作，要借助辅助数组(fd_array)，来保存所有合法的fd！ (select的fd_set* 是输入输出参数，它会继续参数重置)

4.select统一监管是怎么做到的？

(accept后)将最新的fd添加到辅助数组(fd_array)就行了

5.就绪了循环处理所有事件

//SelectServer.hpp
#pragma once#include <cstdint>
#include <memory>
#include <sys/select.h>
#include "Socket.hpp"const static int gnum = sizeof(fd_set) * 8;
const static int gfd = -1;class SelectServer
{
public:SelectServer(uint16_t port) : _port(port), _st(std::make_unique<TcpSocket>()){_st->Tcp_ServerSocket(_port);}void Init(){for (int i = 0; i < gnum; i++){fd_array[i] = gfd;}// 自己先设置fd_array[0] = _st->Sockfd();}void Loop(){while (true){fd_set ft;FD_ZERO(&ft); // 使用前先清除int fd_max = gfd;// 合法的fd添加到ft中for (int i = 0; i < gnum; i++){if (fd_array[i] == gfd)continue;FD_SET(fd_array[i], &ft);// 更新出最大值fd_maxif (fd_max < fd_array[i])fd_max = fd_array[i];}struct timeval tl = {10, 0};// 只关心读事件int n = ::select(fd_max + 1, &ft, nullptr, nullptr, nullptr);switch (n){case 0:std::cout << "time done " << tl.tv_sec << ' ' << tl.tv_usec << std::endl;break;case -1:std::cerr << "select fail" << std::endl;break;default:std::cout << "have even: " << n << std::endl;HandEvent(ft); // 处理事件(不出来会一直通知)PrintFd_array();break;}}}//一定会存在大量的不用类型的sockfd!void HandEvent(fd_set &ft){for (int i = 0; i < gnum; i++){if (fd_array[i] == gfd)continue;// 一定是合法的fd// 判断fd是否就绪if (FD_ISSET(fd_array[i], &ft)) // 动态检测{// 事件就绪// sockfd类型?if (fd_array[i] == _st->Sockfd()){Link();}else{Read_Write(i);}}}}void Link(){InetAddr ir;int sockfd = _st->AcceptSocket(&ir); // 一定不会阻塞!if (sockfd > 0){// 新的sockfd -> fd_array 让select进行管理bool tmp = false;for (int i = 1; i < gnum; i++){if (fd_array[i] == gfd){tmp = true;fd_array[i] = sockfd;std::cout << "add:" << fd_array[i] << std::endl;break;}}// fd_array满了if (!tmp){std::cout << "fd_array full" << std::endl;::close(sockfd);}}}void Read_Write(int i){//没协议，可以直接读(这里考虑IO)char buffer[1024];ssize_t n = ::read(fd_array[i], buffer, sizeof(buffer) - 1); // 阻塞？Noif (n > 0){buffer[n] = 0;std::cout << "Server read:" << buffer << std::endl;std::string content = "<html><body><h1>hello warld</h1></body></html>";std::string ech_str = "HTTP/1.0 200 OK\r\n";ech_str += "Content-Length: " + std::to_string(content.size()) + "\r\n\r\n";ech_str += content;//(这里)默认是就绪的::write(fd_array[i], ech_str.c_str(), ech_str.size());//临时}else if (n == 0){std::cout << "user quit.." << std::endl;::close(fd_array[i]);//清理fdfd_array[i] = gfd;}else{std::cout << "recv fail.." << std::endl;::close(fd_array[i]);//清理fdfd_array[i] = gfd;}}void PrintFd_array(){for (int i = 0; i < gnum; i++){if (fd_array[i] == gfd)continue;std::cout << fd_array[i] << ' ';}std::cout << std::endl;}private:uint16_t _port;std::unique_ptr<Socket> _st;// fd数组->select管理int fd_array[gnum];
};//Main.cc
#include <iostream>
#include <memory>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>
#include <pthread.h>
#include <cstring>
#include <string>#include"Socket.hpp"
#include"SelectServer.hpp"int main(int args, char *argv[])
{if (args != 2){std::cerr << "./Server Localport" << std::endl;exit(0);}uint16_t serverport = std::stoi(argv[1]);std::unique_ptr<SelectServer> sr=std::make_unique<SelectServer>(serverport);sr->Init();sr->Loop();return 0;
}//InetAddr.hpp
#pragma once
#include<iostream>
#include<string>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>class InetAddr
{void ToHost(){//_ip=inet_ntoa(_addr.sin_addr);//4字节地址->char*_port=ntohs(_addr.sin_port);char buffer[124];inet_ntop(AF_INET,&_addr.sin_addr,buffer,sizeof(buffer));_ip=buffer;}
public:InetAddr(){}InetAddr(const struct sockaddr_in& addr):_addr(addr){ToHost();}std::string Ip(){return _ip;}uint16_t Port(){return _port;}struct sockaddr_in Addr(){return _addr;}bool operator==(const InetAddr& ad){return (this->_ip==ad._ip&&this->_port==ad._port);}std::string User(){std::string tmp=_ip+" "+std::to_string(_port)+":";return tmp;} private:std::string _ip;uint16_t _port;struct sockaddr_in _addr;
};//Socket.hpp
#pragma once#include "InetAddr.hpp"class Socket;
enum
{SOCK_ERROR = 1,BIND_ERROR,LISTEN_ERROR,
};
const int gbacklog = 8;class Socket
{
public:virtual void CreateSocket() = 0;virtual void InitSocket(uint16_t port) = 0;virtual int AcceptSocket(InetAddr *addr) = 0;                         // 对象/变量virtual bool ConnectSocket(uint16_t port, const std::string &ip) = 0; // clinet连接成功与失败virtual int Sockfd() = 0;virtual void Close() = 0;virtual ssize_t Recv(std::string *out) = 0;virtual ssize_t Send(const std::string &in) = 0;public:void Tcp_ServerSocket(uint16_t port){CreateSocket();InitSocket(port);}bool Tcp_ClientSocket(uint16_t port, const std::string &ip){CreateSocket();return ConnectSocket(port, ip);}
};class TcpSocket : public Socket
{
public:TcpSocket(){}TcpSocket(int sockfd): _sockfd(sockfd){}~TcpSocket(){}virtual void CreateSocket() override{_sockfd = ::socket(AF_INET, SOCK_STREAM, 0);if (_sockfd < 0){std::cerr << "socket fail" << std::endl;exit(SOCK_ERROR);}std::cout << "socket sucess" << std::endl;}virtual void InitSocket(uint16_t port) override{struct sockaddr_in perr;memset(&perr, 0, sizeof(perr));perr.sin_family = AF_INET;perr.sin_port = htons(port);perr.sin_addr.s_addr = INADDR_ANY;if (::bind(_sockfd, (struct sockaddr *)&perr, sizeof(perr)) < 0){std::cerr << "bind fail" << std::endl;exit(BIND_ERROR);}std::cout << "bind sucess" << std::endl;if (::listen(_sockfd, gbacklog) < 0){std::cerr << "listen fail" << std::endl;exit(LISTEN_ERROR);}std::cout << "listen sucess" << std::endl;}virtual int AcceptSocket(InetAddr *addr) override // 外层要获取客户端信息 TODO{struct sockaddr_in client;socklen_t len = sizeof(client);int sockfd = ::accept(_sockfd, (struct sockaddr *)&client, &len);if (sockfd < 0){std::cerr << "accept fail" << std::endl;return -1;}*addr = client;std::cout << "get a new link " <<addr->User()<< std::endl;return sockfd;}virtual bool ConnectSocket(uint16_t port, const std::string &ip) override{struct sockaddr_in server;memset(&server, 0, sizeof(server));server.sin_family = AF_INET;server.sin_port = htons(port);inet_pton(AF_INET, ip.c_str(), &server.sin_addr);int n = connect(_sockfd, (struct sockaddr *)&server, sizeof(server));if (n < 0){return false;}return true;}virtual int Sockfd() override{return _sockfd;}virtual void Close() override{if (_sockfd > 0){::close(_sockfd);}}virtual ssize_t Recv(std::string *out) override{char buffer[4096];ssize_t n = ::recv(_sockfd, buffer, sizeof(buffer) - 1, 0);if (n > 0){buffer[n] = 0;*out += buffer; // 细节}return n;}virtual ssize_t Send(const std::string &in) override{return ::send(_sockfd, in.c_str(), in.size(), 0);}private:int _sockfd; // 两个角色
};

1.4优缺点

优点： • 实现简单 • 兼容性好，可以实现跨平台 • 在一些老内核中就只能使用select来实现多路复用

缺点： • 需要手动设置 fd 集合（使用起来麻烦(fd 集合混杂着不同的事件类型)）
• 需要把 fd 集合从用户态拷贝到内核态（OS不相信任何人）（无法避免）
• 需要在内核遍历传递进来的所有 fd
• select 支持的文件描述符数量太小（进程打开的文件描述符也是有上限）

2poll

poll主要来解决手动设置核fd太小的问题（其它缺点poll同样存在）

2.1作用和定位

作用：等待多个fd上面的事件就绪，通知用户事件已经就绪，可以进行IO了

定位：只负责等：等就绪后进行事件派发

2.2介绍参数

events和revents的取值

仅仅用一个结构体就解决了：用户让OS关心的事件(设置)和用户让我关心的事件(返回)

不用因为select设计的缺陷，对fd和fd关心的事件进行重新设定！

而数量太小问题：poll就从此就给了用户，由用户定义fd数量后交给poll

2.3修改select代码

//PollServer.hpp
#pragma once#include <cstdint>
#include <memory>
#include <sys/poll.h>
#include "Socket.hpp"const static int gnum = sizeof(fd_set) * 8;
const static int gfd = -1;class PollServer
{
public:PollServer(uint16_t port) : _port(port), _st(std::make_unique<TcpSocket>()){_st->Tcp_ServerSocket(_port);}void Init(){for (int i = 0; i < gnum; i++){fd_array[i].fd = gfd;fd_array[i].events = 0;fd_array[i].revents = 0;}// 自己先设置fd_array[0].fd = _st->Sockfd();fd_array[0].events = POLLIN;}void Link(){InetAddr ir;int sockfd = _st->AcceptSocket(&ir);if (sockfd > 0){// 新的sockfd -> fd_array 让select进行管理bool tmp = false;for (int i = 1; i < gnum; i++){if (fd_array[i].fd == gfd){tmp = true;fd_array[i].fd = sockfd;fd_array[i].events = POLLIN;std::cout << "add:" << fd_array[i].fd << std::endl;break;}}// fd_array满了if (!tmp){std::cout << "fd_array full" << std::endl;}}}void Read_Write(int i){char buffer[1024];ssize_t n = ::read(fd_array[i].fd, buffer, sizeof(buffer) - 1); // 阻塞？Noif (n > 0){buffer[n] = 0;std::cout << "Server read:" << buffer << std::endl;std::string content = "<html><body><h1>hello warld</h1></body></html>";std::string ech_str = "HTTP/1.0 200 OK\r\n";ech_str += "Content-Type: text/html\r\n";ech_str += "Content-Length: " + std::to_string(content.size()) + "\r\n\r\n";ech_str += content;::write(fd_array[i].fd, ech_str.c_str(), ech_str.size());}else if (n == 0){std::cout << "user quit.." << std::endl;::close(fd_array[i].fd);fd_array[i].fd = gfd;fd_array[i].events = 0;fd_array[i].revents = 0;}else{std::cout << "recv fail.." << std::endl;::close(fd_array[i].fd);fd_array[i].fd = gfd;fd_array[i].events = 0;fd_array[i].revents = 0;}}void HandEvent(){for (int i = 0; i < gnum; i++){if (fd_array[i].fd == gfd)continue;// 哪个读事件就绪if (fd_array[i].revents & POLLIN){// 处理特定sockfdif (fd_array[i].fd == _st->Sockfd()){Link();}// 处理普通fdelse{Read_Write(i);}}}}void Loop(){while (true){// 不用设置啦 -> 优雅int n = ::poll(fd_array, gnum, 1000);switch (n){case 0:std::cout << "time done " << std::endl;break;case -1:std::cerr << "select fail" << std::endl;break;default:std::cout << "have even: " << n << std::endl;HandEvent();PrintFd_array();break;}}}void PrintFd_array(){for (int i = 0; i < gnum; i++){if (fd_array[i].fd == gfd)continue;std::cout << fd_array[i].fd << ' ';}std::cout << std::endl;}private:uint16_t _port;std::unique_ptr<Socket> _st;// fd数组->select管理pollfd fd_array[gnum];
};

poll的底层也是要OS遍历所有的fd，来获取fd和对应的事件（效率与select一样不高）

所以就有了下面的epoll~

3epoll

按照 man 手册的说法: 为处理大批量句柄而作了改进的 poll
它是在 2.5.44 内核中被引进的(epoll(4) is a new API introduced in Linux kernel 2.5.44)它几乎具备了之前所说的一切优点， 被公认为 Linux2.6 下性能最好的多路 I/O 就绪通知方法

3.1介绍参数

event可以是几个宏的集合

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

虽然说是epoll的加强版，但两者差别真的挺大的！

3.2工作原理

介绍完参数，此刻的你或许有很多问题：不急先来谈谈epoll的工作原理

理解数据到达主机

主机从网络中收数据时通过网卡来解收的，从而往上交给数据链路层；数据链路层是OS管的，也就是说OS怎么知道网卡有数据了需要进行拷贝了？

有人会说：简单，让OS定期检测一遍不就行了？OS要管理小到进程行大到外设，所有事情都要轮询OS不得忙死，所以OS是通过网卡给我发送信号(硬件中断)来得知网卡有数据啦！

epoll函数与OS的关系

但实际上，OS管理就绪队列时不用创建节点的！

OS也要对epoll模型作管理：先描述，在组织

3.3编写epoll代码

//ServerEpoll.hpp
#pragma once#include "Socket.hpp"const static int gnum = 128;
const static int gsize = 128;class ServerEpoll
{
public:ServerEpoll(uint16_t port): _port(port), _st(std::make_unique<TcpSocket>()){_st->Tcp_ServerSocket(_port);_epfd = ::epoll_create(gsize);if (_epfd < 0){LOG(FATAL, "epoll_creat fail\n");exit(1);}LOG(INFO, "epoll_creat sucess,_epfd:%d\n", _epfd);}~ServerEpoll(){if (_epfd > 0)::close(_epfd);_st->Close();}void Init(){epoll_event et;et.data.fd = _st->Sockfd(); // 为了在事件就绪时知道fd就绪了et.events = EPOLLIN;int n = ::epoll_ctl(_epfd, EPOLL_CTL_ADD, _st->Sockfd(), &et);if (n < 0){LOG(ERROR, "epoll_ctl fail\n");exit(2);}LOG(INFO, "epoll_ctl sucess,add %d\n", _st->Sockfd());}void Start(){int time = -1;while (true){int n = ::epoll_wait(_epfd, ets, gnum, time);switch (n){case 0:LOG(INFO, "epoll_wait time done\n");break;case -1:LOG(INFO, "epoll_wait fail\n");break;default:LOG(INFO, "epoll sucess total:%d\n", n);Handler(n);break;}}}void Handler(int n){for (int i = 0; i < n; i++){int sockfd = ets[i].data.fd;uint32_t event = ets[i].events;LOG(INFO, "%d 事件就绪 event:%s\n", sockfd, EventToString(event).c_str());if (event & EPOLLIN){if (sockfd == _st->Sockfd()){Accept();}else{HandlerIO(sockfd);}}}}void Accept(){InetAddr ar;int sockfd = _st->AcceptSocket(&ar);if (sockfd < 0){LOG(ERROR, "gain line error sockfd: %d\n", sockfd);return;}LOG(INFO, "gain link:%d client:%s\n", sockfd, ar.User().c_str());epoll_event et;et.data.fd = sockfd;et.events = EPOLLIN;// OS在底层建立红黑树节点 -- select/poll要循环自己维护~int n = ::epoll_ctl(_epfd, EPOLL_CTL_ADD, sockfd, &et);if (n < 0){LOG(ERROR, "epoll_ctl fail\n");exit(2);}LOG(INFO, "epoll_ctl sucess,add %d\n", sockfd);}void HandlerIO(int fd){// 不保证读到的数据是正确的:要通过协议来保证~char buffer[4096];ssize_t n = ::recv(fd, buffer, sizeof(buffer) - 1, 0);if (n > 0){buffer[n] = 0;std::cout << buffer;std::string content = "<html><body><h1>Hello Warld</h1></body></html>";std::string ech_str = "HTTP/1.0 200 OK\r\n";ech_str += "Content-Length:" + std::to_string(content.size()) + "\r\n";ech_str += "\r\n";ech_str += content;::send(fd, ech_str.c_str(), ech_str.size(), 0);}else if (n == 0){LOG(INFO, "Clinet quit fd:%d\n", fd);// epoll移除 fd fd要保证:健康&&合法::epoll_ctl(_epfd, EPOLL_CTL_DEL, fd, nullptr);// 再关闭::close(fd);}else{LOG(INFO, "read fail\n");// epoll移除 fd fd要保证:健康&&合法::epoll_ctl(_epfd, EPOLL_CTL_DEL, fd, nullptr);::close(fd);}}std::string EventToString(uint32_t event){std::string et;if (event & EPOLLIN)et += "EPOLLIN";if (event & EPOLLOUT)et += " | EPOLLOUT";return et;}private:uint16_t _port;std::unique_ptr<Socket> _st;int _epfd;epoll_event ets[gnum]; // 不用自己维护
};//Main.cc
#include<iostream>
#include<memory>
#include<sys/epoll.h>#include"ServerEpoll.hpp"int main(int args,char* argc[])
{if(args!=2){std::cout<<"./Main Port"<<std::endl;exit(1);}uint16_t port=std::stoi(argc[1]);std::unique_ptr<ServerEpoll> usr=std::make_unique<ServerEpoll>(port);usr->Init();usr->Start();return 0;
}//InetAddr.hpp
#pragma once
#include<iostream>
#include<string>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>class InetAddr
{void ToHost(){//_ip=inet_ntoa(_addr.sin_addr);//4字节地址->char*_port=ntohs(_addr.sin_port);char buffer[124];inet_ntop(AF_INET,&_addr.sin_addr,buffer,sizeof(buffer));_ip=buffer;}
public:InetAddr(){}InetAddr(const struct sockaddr_in& addr):_addr(addr){ToHost();}std::string Ip(){return _ip;}uint16_t Port(){return _port;}struct sockaddr_in Addr(){return _addr;}bool operator==(const InetAddr& ad){return (this->_ip==ad._ip&&this->_port==ad._port);}std::string User(){std::string tmp=_ip+" "+std::to_string(_port)+":";return tmp;} private:std::string _ip;uint16_t _port;struct sockaddr_in _addr;
};//Log.hpp
#pragma once#include<iostream>
#include<string>
#include<cstring>
#include<fstream>
#include <stdarg.h>
#include <unistd.h>
#include <sys/syscall.h>enum
{DEBUG = 1,INFO,WARNING,ERROR,FATAL
};std::string Getlevel(int level)
{switch (level){case DEBUG:return "DEBUG";break;case INFO:return "INFO";break;case WARNING:return "WARNING";break;case ERROR:return "ERROR";break;case FATAL:return "FATAL";break;default:return "";break;}
}std::string Gettime()
{time_t now = time(nullptr);struct tm *time = localtime(&now);char buffer[128];snprintf(buffer, sizeof(buffer), "%d-%02d-%02d %02d:%02d:%02d",time->tm_year + 1900,time->tm_mon + 1,time->tm_mday,time->tm_hour,time->tm_min,time->tm_sec);return buffer;
}struct log_message
{std::string _level;int _id;std::string _filename;int _filenumber;std::string _cur_time;std::string _message;
};#define SCREAM 1
#define FILE 2#define DEVELOP 3
#define OPERATION 4const std::string gpath = "./log.txt";
pthread_mutex_t gmutex = PTHREAD_MUTEX_INITIALIZER;class log
{
public:log(const std::string &path = gpath, const int status = DEVELOP): _mode(SCREAM), _path(path), _status(status){}void SelectMode(int mode){_mode = mode;}void SelectStatus(int status){_status = status;}void PrintScream(const log_message &le){printf("[%s][%d][%s][%d][%s] %s",le._level.c_str(),le._id,le._filename.c_str(),le._filenumber,le._cur_time.c_str(),le._message.c_str());}void PrintFile(const log_message &le){std::fstream in(_path, std::ios::app);if (!in.is_open())return;char buffer[1024];snprintf(buffer, sizeof(buffer), "[%s][%d][%s][%d][%s] %s",le._level.c_str(),le._id,le._filename.c_str(),le._filenumber,le._cur_time.c_str(),le._message.c_str());in.write(buffer, strlen(buffer)); // 不用sizeofin.close();}void PrintLog(const log_message &le){// 过滤if (_status == OPERATION)return;// 线程安全pthread_mutex_lock(&gmutex);switch (_mode){case SCREAM:PrintScream(le);break;case FILE:PrintFile(le);break;default:break;}pthread_mutex_unlock(&gmutex);}void logmessage(int level, const std::string &filename, int filenumber, const char *message, ...){log_message le;le._level = Getlevel(level);le._id = syscall(SYS_gettid);le._filename = filename;le._filenumber = filenumber;le._cur_time = Gettime();va_list vt;va_start(vt, message);char buffer[128];vsnprintf(buffer, sizeof(buffer), message, vt);va_end(vt);le._message = buffer;// 打印日志PrintLog(le);}~log(){}private:int _mode;std::string _path;int _status;
};// 方便上层调用
log lg;// ##不传时可忽略参数
#define LOG(level, message, ...)                                          \do                                                                    \{                                                                     \lg.logmessage(level, __FILE__, __LINE__, message, ##__VA_ARGS__); \} while (0)#define SleftScream()          \do                         \{                          \lg.SelectMode(SCREAM); \} while (0)
#define SleftFile()          \do                       \{                        \lg.SelectMode(FILE); \} while (0)#define SleftDevelop()            \do                            \{                             \lg.SelectStatus(DEVELOP); \} while (0)
#define SleftOperation()            \do                              \{                               \lg.SelectStatus(OPERATION); \} while (0)//#pragma once#include "InetAddr.hpp"
#include "Log.hpp"class Socket;
enum
{SOCK_ERROR = 1,BIND_ERROR,LISTEN_ERROR,
};
const int gbacklog = 8;class Socket
{
public:virtual void CreateSocket() = 0;virtual void InitSocket(uint16_t port) = 0;virtual int AcceptSocket(InetAddr *addr) = 0;                         // 对象/变量virtual bool ConnectSocket(uint16_t port, const std::string &ip) = 0; // clinet连接成功与失败virtual int Sockfd() = 0;virtual void Close() = 0;virtual ssize_t Recv(std::string *out) = 0;virtual ssize_t Send(const std::string &in) = 0;public:void Tcp_ServerSocket(uint16_t port){CreateSocket();InitSocket(port);}bool Tcp_ClientSocket(uint16_t port, const std::string &ip){CreateSocket();return ConnectSocket(port, ip);}
};class TcpSocket : public Socket
{
public:TcpSocket(){}TcpSocket(int sockfd): _sockfd(sockfd){}~TcpSocket(){}virtual void CreateSocket() override{_sockfd = ::socket(AF_INET, SOCK_STREAM, 0);if (_sockfd < 0){LOG(ERROR,"create socket fail\n");exit(SOCK_ERROR);}LOG(INFO,"create socket sucess,sockfd:%d\n",_sockfd);}virtual void InitSocket(uint16_t port) override{struct sockaddr_in perr;memset(&perr, 0, sizeof(perr));perr.sin_family = AF_INET;perr.sin_port = htons(port);perr.sin_addr.s_addr = INADDR_ANY;if (::bind(_sockfd, (struct sockaddr *)&perr, sizeof(perr)) < 0){LOG(ERROR,"bind fail\n");exit(BIND_ERROR);}LOG(INFO,"bind sucess\n");if (::listen(_sockfd, gbacklog) < 0){LOG(ERROR,"listen fail\n");exit(LISTEN_ERROR);}LOG(INFO,"listen sucess\n");}virtual int AcceptSocket(InetAddr *addr) override // 外层要获取客户端信息 TODO{struct sockaddr_in client;socklen_t len = sizeof(client);int sockfd = ::accept(_sockfd, (struct sockaddr *)&client, &len);if (sockfd < 0){LOG(ERROR,"link fail\n");return -1;}*addr = client;return sockfd;}virtual bool ConnectSocket(uint16_t port, const std::string &ip) override{struct sockaddr_in server;memset(&server, 0, sizeof(server));server.sin_family = AF_INET;server.sin_port = htons(port);inet_pton(AF_INET, ip.c_str(), &server.sin_addr);int n = connect(_sockfd, (struct sockaddr *)&server, sizeof(server));if (n < 0){return false;}return true;}virtual int Sockfd() override{return _sockfd;}virtual void Close() override{if (_sockfd > 0){::close(_sockfd);}}virtual ssize_t Recv(std::string *out) override{char buffer[4096];ssize_t n = ::recv(_sockfd, buffer, sizeof(buffer) - 1, 0);if (n > 0){buffer[n] = 0;*out += buffer; // 细节}return n;}virtual ssize_t Send(const std::string &in) override{return ::send(_sockfd, in.c_str(), in.size(), 0);}private:int _sockfd; // 两个角色
};

3.4优点

• 接口使用方便: 虽然拆分成了三个函数, 但是反而使用起来更方便高效. 不需要每次循环都设置关注的文件描述符, 也做到了输入输出参数分离开
• 数据拷贝轻量: 只在合适的时候调用 EPOLL_CTL_ADD 将文件描述符结构拷贝到内核中, 这个操作并不频繁(而 select/poll 都是每次循环都要进行拷贝)
• 事件回调机制: 避免使用遍历, 而是使用回调函数的方式, 将就绪的文件描述符结构加入到就绪队列中(填写就绪队列的节点), epoll_wait 返回直接访问就绪队列就知道哪些文件描述符就绪. 这个操作时间复杂度 O(1)，拷贝到用户定义的数组虽然也是O(N)，但它严格按顺序进行拷贝

3.5工作模式

在XX快递站中有两个快递员：张三和李四；张三工作相对负责，而李四工作较毛躁；有一天住在5楼的小王在网上买了5个快递，现在正在派发：有三个在张三手上，有两个在李四手上；现在张三先到了小王楼下，带电话给小王叫他下来去快递，但由于小王在玩游戏，也就没下去取；张三一看小王没下来就重复给小王打电话；等到小王游戏打完后就下楼拿快递了，但他只拿了两个就上楼了；现在李四到了问张三：你还有多少个快递没派完？“还有一个小王的”张三说；那正好我手上有2个小王的包裹，你帮我一起派发吧！张三也只好接下了2个小王新增的包裹，继续给小王打电话，知道他吧包裹全取走为止...

后来小王又在网上买了5个快递，派发是恰好3个在李四手上，2个在张三手上；现在轮到李四先到小王楼下；由于李四工作比较毛躁，打电话给小王说：我只打电话给你一次，如果你不下来取快递，派发完别人的我就先走了！小王一听对方语气有点强硬，也就刚上了就不下去取；此时张三到了小王楼下，看到李四问有小王的快递没？一听有，张三就把小王的快递交给了李四说：上次我帮了你，这次你帮我下；李四也是挺讲义气的，也就把包裹拿来了，这次手上新增了2个包裹，再次打给了小王，小王说：你不是说不再给我打电话的吗？“没办法，手上又新增了你的2个快递，你快点下来拿把，不然我就走了！”小王也就得毕竟是自己的快递没必要不下去拿，也就下去把5个快递都拿回家了~

张三：只要底层一直有快递（数据），就一直通知小王（上层）

李四：包裹从无到有，从有到多的时候，才会通知小王（上层）

张三和李四就对应着epoll的两种不同的工作模式：水平触发(LT)边缘触发(ET)

理解LT和ET

场景：客户端向(ET模式下)服务器发送10k请求数据，服务器只读了1k数据

在ET模式下只通知一次，本轮数据如果没读完不会再通知；如果服务器没读完数据，导致双方都要发数据给对方，对方才能进行处理数据（陷入死循环）

这就要求在ET模式下读事件一旦就绪，必须把数据读完；那我怎么知道我把数据读完了？

这就好比关系还不错的朋友来向你借钱，第一次借100，下一次再借100，那他下次会不会再向你借钱呢？肯定会（知道你有钱~）；那他怎么知道你没钱了？第三次向你借钱时，你发信息给他：我只有36.66块钱了，这是你朋友就知道你没钱了！

也就是在设计时进行循环读取，只到读取不到数据就证明没数据了！

如果这时朋友发信息再次向你借钱，你没回复他，他也不知道是什么情况(没钱还是没看到信息)，就会一直僵持着

也就是说：循环读取势必会引起阻塞问题，但服务器敢让你阻塞吗(一阻塞可能服务器就挂了)！所以我们要对fd设置非阻塞来解决该问题！

以上便是为了引出结论：在ET模式下，所有的fd必须是非阻塞的！LT模式阻塞非阻塞都行~

ET更高效，体现在：1通知效率更高；2IO效率更高（给对方通告一个更大的接收窗口，让对方能发送更多的数据，提高IO效率）

那LT也可以设置非阻塞，也可以循环读取所有数据啊！但它会有退路可言(设置成阻塞,一次一次读数据)而ET一定要让上层读取全部数据！这在设计上ET也比LT要复杂些；但虽然ET更高效，也并不意味着LT就没用了：在一些场景中如果要实现响应是有顺序的或者是要求设计简单的，就选LT~

四Reactor

编写Reactor的基本框架

在Reactor的代码中，我们要来解决读和写的问题

关于读：我们能够直接读吗？ 不能！

我们要怎么知道接收缓冲区有数据了？ 通过epoll等待读事件就绪！

我们能保证读到的就是一个完整的请求吗？ 不能！

如何保证？ 引入协议：对读到的内容进行报文解析

关于写：a当我们获取一个新的sockfd时，输入和输出缓冲区都是空的

b读事件就绪：就是输入缓冲区有了数据/底层有新连接到来

c写事件就绪：不是关心数据，而是关心发送缓冲区中有没有空间，有空间就证明写事件就绪；否则写事件不满足！

d把sockfd托管给select，poll，epoll：原因是sockfd上事件没就绪！

所以默认sockfd新建的情况下，读事件是不就绪的：接收缓冲区暂时没数据，所以我们要把它进行托管

而默认sockfd新建的情况下，写事件是就绪的：接收缓冲区本来就有空间，所以我们直接写

当写条件不满足时，我们才要按需开启对EPOLLOUT的关心！

什么情况下写条件不满足？ 发送缓冲区写满了&&数据还没发完

设置EPOLLOUT之后，我们就不用管了！后续的剩余数据epoll会自动给我发送！

补充细节：

1如果我们直接开启EPOLLOUT关心，epoll就会一直就绪

2如果未来我们发完数据了，对EPOLLOUT的关心就会被关闭

3如果缓冲区没满&&数据发完了，我们就不用开启对EPOLLOUT关心

4如果我们设置对EPOLLOUT的关心，epoll默认只会就绪一次！

5直接发，我们怎么知道写入条件不满足？我们写入时进行判断errno == EWOULDBLOCK就是缓冲区被写满了！

在代码中我们没有sockfd的概念，只有一个一个的connection到来：所以我们要对connection进行描述并构建成对象，有Reactor对象来管理connection对象(通过unordered_map数据结构来组织)

accept设置连接时并不知道有多少个连接即将到来，所以我们要把listensockfd设置成非阻塞，循环获取新连接！

获取新连接时，我们要对读写异常方法进行bind回调函数，我们不在listener类中进行bind，我们统一在Main.cc(外部)将方法bind，怎么实现？在connection类中定义一个type：表示该连接是监听sockfd还是普通sockfd；在Reactor类中定义一批方法集(读写异常方法)，对外提供接口设置这批方法（也就是先在外部bind）；从此我们AddConnection()函数添加连接是通过type判断是那种连接需要添加从而进行注册对应的方法(注册进connection类的成员方法中，后面等新事件到来时就能通过connection找到要执行的方法从而回调出去！)

在connection类中定义一个Reactor* R指针？ 进行回指找到Reacor中的AddConnection()函数（在获取连接时我们要把新连接添加到unordered_map中要通过Reactor对象才能实现）

完整源代码

未来服务器将不会有sockfd的概念，只有一个一个的Connection

//Connection.hpp
#pragma once
#include <string>
#include <functional>#define ListenConnection 0
#define NornalConnection 1class Connection;
class Reactor;
using handler = std::function<void(Connection *)>;class Connection
{
public:Connection(int sockfd): _sockfd(sockfd){}~Connection(){}void SetEvent(uint32_t event){_event = event;}void SetReactor(Reactor *R){_R = R;}void SetType(int type){_type = type;}void SetAddr(InetAddr &addr){_addr = addr;}void AppendInbuffer(const std::string &in){_inbuffer += in;}void AppendOutbuffer(const std::string &in){_outbuffer += in;}int Sockfd(){return _sockfd;}uint32_t Event(){return _event;}handler Recver(){return _recver;}handler Sender(){return _sender;}handler Excepter(){return _excepter;}Reactor *R(){return _R;}int Type(){return _type;}std::string &Inbuffer() // 引用！{return _inbuffer;}InetAddr Addr(){return _addr;}std::string &Outbuffer(){return _outbuffer;}void RegisterHandler(handler recver, handler sender, handler excepter){_recver = recver;_sender = sender;_excepter = excepter;}void DiscardOutbuffer(ssize_t n){_outbuffer.erase(0, n);}void Close(){if (_sockfd >= 0)::close(_sockfd);}private:int _sockfd;uint32_t _event;std::string _inbuffer;std::string _outbuffer;handler _recver;   // 处理读事件handler _sender;   // 处理写事件handler _excepter; // 处理异常事件Reactor *_R; // 定义了一个Reactor指针?int _type;InetAddr _addr;
};

Connection被Reactor进行管理

//Reactor.hpp
#pragma once
#include <unordered_map>
#include <memory>#include "Multiplex.hpp"
#include "Connection.hpp"// 只对Connection进行管理
class Reactor
{const static int gnum = 128;public:Reactor(): _epoller(std::make_unique<Epoller>()), _isrunning(false){}~Reactor(){}// 我不想让外部去new Connection对象// void AddConnection(int sockfd, uint32_t event, handler recver, handler sender, handler excepter)void AddConnection(int sockfd, uint32_t event, InetAddr &addr, int type){// 构建Connection -- 不要用智能指针!!!Connection *conn = new Connection(sockfd);conn->SetEvent(event);conn->SetReactor(this); // 设置conn与R的回指conn->SetType(type);    // 设置connection的类型conn->SetAddr(addr);// 注册读写异常方法// conn->RegisterHandler(recver, sender, excepter);if (conn->Type() == ListenConnection)conn->RegisterHandler(_OnAccepter, nullptr, nullptr);if (conn->Type() == NornalConnection)conn->RegisterHandler(_OnRecver, _OnSender, _OnExcepter);// epoll托管 - epoll模型进行封装if (_epoller->AddEvent(conn->Sockfd(), conn->Event()))_conn.insert({sockfd, conn}); // TcpServer管理}void Dispatcher(){int timeout = -1;_isrunning = true;while (_isrunning){LoopOnce(timeout);// Do Other ThingPrintDebug();}_isrunning = false;}void LoopOnce(int timeout){int n = _epoller->WaitEvent(revs, gnum, timeout);for (int i = 0; i < n; i++){int sockfd = revs[i].data.fd;uint32_t event = revs[i].events;if (event & EPOLLERR)event |= (EPOLLIN | EPOLLOUT);if (event & EPOLLHUP)event |= (EPOLLIN | EPOLLOUT);if (event & EPOLLIN){if (IsConnExist(sockfd) && _conn[sockfd]->Recver())_conn[sockfd]->Recver()(_conn[sockfd]); // 写事件派发}if (event & EPOLLOUT){if (IsConnExist(sockfd) && _conn[sockfd]->Sender())_conn[sockfd]->Sender()(_conn[sockfd]); // 读事件派发}}}bool IsConnExist(int sockfd){return _conn.find(sockfd) != _conn.end();}void SetListenConnection(handler OnAccepter){_OnAccepter = OnAccepter;}void SetNornalConnection(handler OnRecver, handler OnSender, handler OnExcepter){_OnRecver = OnRecver;_OnSender = OnSender;_OnExcepter = OnExcepter;}handler OnAccepter(){return _OnAccepter;}handler OnRecver(){return _OnRecver;}handler OnSender(){return _OnSender;}handler OnExcepter(){return _OnExcepter;}void PrintDebug(){std::string str;for (auto &conn : _conn){str += std::to_string(conn.second->Sockfd()) + " ";}LOG(DEBUG, "conn list:%s\n", str.c_str());}void EnableReadWriteEvent(int sockfd, bool read, bool write){if (!IsConnExist(sockfd))return;uint32_t event = (read ? EPOLLIN : 0) | (write ? EPOLLOUT : 0) | EPOLLET;// 修改_conn[sockfd]->SetEvent(event);// 设置进内核_epoller->ModEvent(_conn[sockfd]->Sockfd(), _conn[sockfd]->Event());}void DelConnection(int sockfd){// 进行安全检测if (!IsConnExist(sockfd))return;LOG(INFO, "client %s quit,server delete all sources\n", _conn[sockfd]->Addr().User().c_str());// 先从epoller中移除->细节！epoller移除的sockfd必须是健康的// EnableReadWriteEvent(sockfd,false,false);_epoller->DelEvent(sockfd);// 关闭sockfd_conn[sockfd]->Close();// 从_conn中去除_conn.erase(sockfd);}private:// sockfd与connection进行映射std::unordered_map<int, Connection *> _conn;std::unique_ptr<Multiplex> _epoller;bool _isrunning;epoll_event revs[gnum];// 设置方法集进行统一处理handler _OnAccepter;handler _OnRecver;handler _OnSender;handler _OnExcepter;
};

epoll接口封装成对象

//Multiplex.hpp
#pragma once
#include <sys/epoll.h>#include "Log.hpp"
#include "Comm.hpp"using namespace log_ns;
class Multiplex
{
public:virtual bool AddEvent(int sockfd, uint32_t event) = 0;virtual int WaitEvent(struct epoll_event revt[], int num, int timeout) = 0;virtual bool ModEvent(int sockfd, uint32_t event) = 0;virtual bool DelEvent(int sockfd) = 0;
};class Epoller : public Multiplex
{const static int size = 128;private:bool EventHelper(int sockfd, uint32_t event, int oper){struct epoll_event evt;evt.data.fd = sockfd;evt.events = event;int n = ::epoll_ctl(_epfd, oper, sockfd, &evt);if (n < 0){LOG(ERROR, "epoll_ctl failed\n");return false;}LOG(INFO, "epoll_ctl %d events is %s sucess\n", sockfd, EventToString(event).c_str());return true;}public:Epoller(){_epfd = ::epoll_create(size);if (_epfd < 0){LOG(FATAL, "epoll_create error\n");exit(EPOLL_CTL_ERROR);}LOG(INFO, "epoll_create sucess efd: %d\n", _epfd);}~Epoller(){}std::string EventToString(uint32_t event){std::string str;if (event & EPOLLIN)str += "EPOLLIN";if (event & EPOLLOUT)str += "|EPOLLOUT";if (event & EPOLLET)str += "|EPOLLET";return str;}virtual bool AddEvent(int sockfd, uint32_t event) override{return EventHelper(sockfd, event, EPOLL_CTL_ADD);}virtual bool ModEvent(int sockfd, uint32_t event) override{return EventHelper(sockfd, event, EPOLL_CTL_MOD);}virtual bool DelEvent(int sockfd) override{return EventHelper(sockfd, 0, EPOLL_CTL_DEL);}virtual int WaitEvent(epoll_event revs[], int num, int timeout) override{return ::epoll_wait(_epfd, revs, num, timeout);}private:int _epfd;
};// class Poller:private Multiplex
// {// };// class Select:private Multiplex
// {// };

对listensockfd的封装，内含读事件(连接)方法

//Listener.hpp
#pragma once
#include <memory>#include "Socket.hpp"
#include "Connection.hpp"
#include "Log.hpp"
#include "Reactor.hpp"using namespace socket_ns;
using namespace log_ns;
class Listener
{
public:Listener(uint16_t port): _port(port), _ListenSocket(std::make_unique<TcpSocket>()){_ListenSocket->Tcp_ServerSocket(port);}int Listensockfd(){return _ListenSocket->Sockfd();}// listensockfd 读方法void Accepter(Connection *conn){while (true){errno = 0;InetAddr addr;int code = 0;int sockfd = _ListenSocket->AcceptSocket(&addr, &code);if (sockfd > 0){LOG(INFO, "get a new link %s sockfd: %d\n", addr.User().c_str(), sockfd);conn->R()->AddConnection(sockfd, EPOLLIN | EPOLLET, addr, NornalConnection); }else{if (code == EWOULDBLOCK){LOG(INFO, "gain connection is finished\n");break;}else if (code == EINTR){continue;}else{LOG(ERROR, "gain connection failed\n");break;}}}}private:uint16_t _port;std::unique_ptr<Socket> _ListenSocket;
};//Socket.hpp
#pragma once
#include <iostream>
#include <cstring>
#include <functional>
#include <unistd.h>
#include <memory>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/wait.h>
#include <pthread.h>#include "Log.hpp"
#include "InetAddr.hpp"
#include "Comm.hpp"namespace socket_ns
{using namespace log_ns;const static int gbacklog = 8;// 模板方法模式class Socket{public:virtual void CreateSocket() = 0;virtual void InitSocket(uint16_t port, int backlog = gbacklog) = 0;virtual int AcceptSocket(InetAddr *addr, int *code) = 0;              // 对象/变量virtual bool ConnectSocket(uint16_t port, const std::string &ip) = 0; // clinet连接成功与失败virtual int Sockfd() = 0;virtual void Close() = 0;virtual ssize_t Recv(std::string *out) = 0;virtual ssize_t Send(const std::string &in) = 0;virtual void ReuseAddr() = 0;public:void Tcp_ServerSocket(uint16_t port){CreateSocket();ReuseAddr();InitSocket(port);}bool Tcp_ClientSocket(uint16_t port, const std::string &ip){CreateSocket();return ConnectSocket(port, ip);}};class TcpSocket : public Socket{public:TcpSocket(){}TcpSocket(int sockfd): _sockfd(sockfd){}~TcpSocket(){}virtual void CreateSocket() override{_sockfd = ::socket(AF_INET, SOCK_STREAM, 0);if (_sockfd < 0){LOG(FATAL, "socket fail\n");exit(SOCK_ERROR);}LOG(INFO, "socket sucess sockfd: %d\n", _sockfd);NoBlock(_sockfd); // 设置非阻塞}virtual void InitSocket(uint16_t port, int backlog) override{struct sockaddr_in perr;memset(&perr, 0, sizeof(perr));perr.sin_family = AF_INET;perr.sin_port = htons(port);perr.sin_addr.s_addr = INADDR_ANY;if (::bind(_sockfd, (struct sockaddr *)&perr, sizeof(perr)) < 0){LOG(FATAL, "bind fail\n");exit(BIND_ERROR);}if (::listen(_sockfd, backlog) < 0){LOG(ERROR, "listen fail\n");exit(LISTEN_ERROR);}}virtual int AcceptSocket(InetAddr *addr, int *code) override{struct sockaddr_in client;socklen_t len = sizeof(client);int sockfd = ::accept(_sockfd, (struct sockaddr *)&client, &len);*code = errno;if (sockfd < 0){return -1;}NoBlock(sockfd); // 设置非阻塞*addr = client;return sockfd;}virtual bool ConnectSocket(uint16_t port, const std::string &ip) override{struct sockaddr_in server;memset(&server, 0, sizeof(server));server.sin_family = AF_INET;server.sin_port = htons(port);inet_pton(AF_INET, ip.c_str(), &server.sin_addr);int n = connect(_sockfd, (struct sockaddr *)&server, sizeof(server));if (n < 0){return false;}return true;}virtual int Sockfd() override{return _sockfd;}virtual void Close() override{if (_sockfd > 0){::close(_sockfd);}}virtual ssize_t Recv(std::string *out) override{char buffer[4096];ssize_t n = ::recv(_sockfd, buffer, sizeof(buffer) - 1, 0);if (n > 0){buffer[n] = 0;*out += buffer; // 细节}return n;}virtual ssize_t Send(const std::string &in) override{return ::send(_sockfd, in.c_str(), in.size(), 0);}virtual void ReuseAddr() override{int opt = 1;::setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));}private:int _sockfd; // 两个角色};// class UdpServer:public Socket//{}
}//Log.hpp
#pragma once
#include <pthread.h>
#include <fstream>
#include <syscall.h>
#include <stdarg.h>
#include <unistd.h>
#include <cstring>namespace log_ns
{enum{DEBUG = 1,INFO,WARNING,ERROR,FATAL};std::string Getlevel(int level){switch (level){case DEBUG:return "DEBUG";break;case INFO:return "INFO";break;case WARNING:return "WARNING";break;case ERROR:return "ERROR";break;case FATAL:return "FATAL";break;default:return "";break;}}std::string Gettime(){time_t now = time(nullptr);struct tm *time = localtime(&now);char buffer[128];snprintf(buffer, sizeof(buffer), "%d-%02d-%02d %02d:%02d:%02d",time->tm_year + 1900,time->tm_mon + 1,time->tm_mday,time->tm_hour,time->tm_min,time->tm_sec);return buffer;}struct log_message{std::string _level;int _id;std::string _filename;int _filenumber;std::string _cur_time;std::string _message;};#define SCREAM_TYPE 1
#define FILE_TYPE 2#define DEVELOP 3
#define OPERATION 4const std::string gpath = "./log.txt";pthread_mutex_t gmutex = PTHREAD_MUTEX_INITIALIZER;class log{public:log(const std::string &path = gpath, const int status = DEVELOP): _mode(SCREAM_TYPE), _path(path), _status(status){}void SelectMode(int mode){_mode = mode;}void SelectStatus(int status){_status = status;}void PrintScream(const log_message &le){printf("[%s][%d][%s][%d][%s] %s",le._level.c_str(),le._id,le._filename.c_str(),le._filenumber,le._cur_time.c_str(),le._message.c_str());}void PrintFile(const log_message &le){std::fstream in(_path, std::ios::app);if (!in.is_open())return;char buffer[1024];snprintf(buffer, sizeof(buffer), "[%s][%d][%s][%d][%s] %s",le._level.c_str(),le._id,le._filename.c_str(),le._filenumber,le._cur_time.c_str(),le._message.c_str());in.write(buffer, strlen(buffer)); // 不用sizeofin.close();}void PrintLog(const log_message &le){// 过滤if (_status == OPERATION)return;// 线程安全pthread_mutex_lock(&gmutex);switch (_mode){case SCREAM_TYPE:PrintScream(le);break;case FILE_TYPE:PrintFile(le);break;default:break;}pthread_mutex_unlock(&gmutex);}void logmessage(const std::string &filename, int filenumber, int level, const char *message, ...){log_message le;le._level = Getlevel(level);le._id = syscall(SYS_gettid);le._filename = filename;le._filenumber = filenumber;le._cur_time = Gettime();va_list vt;va_start(vt, message);char buffer[128];vsnprintf(buffer, sizeof(buffer), message, vt);va_end(vt);le._message = buffer;// 打印日志PrintLog(le);}~log(){}private:int _mode;std::string _path;int _status;};// 方便上层调用log lg;
// ##不传时可忽略参数
#define LOG(level, message, ...)                                          \do                                                                    \{                                                                     \lg.logmessage(__FILE__, __LINE__, level, message, ##__VA_ARGS__); \} while (0)#define SleftScream()               \do                              \{                               \lg.SelectMode(SCREAM_TYPE); \} while (0)
#define SleftFile()               \do                            \{                             \lg.SelectMode(FILE_TYPE); \} while (0)}//InetAddr.hpp
#pragma once
#include <iostream>
#include <string>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>class InetAddr
{void ToHost(){//_ip=inet_ntoa(_addr.sin_addr);//4字节地址->char*_port = ntohs(_addr.sin_port);char buffer[124];inet_ntop(AF_INET, &_addr.sin_addr, buffer, sizeof(buffer));_ip = buffer;}public:InetAddr(const std::string &ip, uint16_t port){_ip = ip;_port = port;_addr.sin_family = AF_INET6;_addr.sin_port = htons(port);_addr.sin_addr.s_addr = INADDR_ANY;}InetAddr(){}InetAddr(const struct sockaddr_in &addr): _addr(addr){ToHost();}std::string Ip(){return _ip;}uint16_t Port(){return _port;}struct sockaddr_in Addr(){return _addr;}bool operator==(const InetAddr &ad){return (this->_ip == ad._ip && this->_port == ad._port);}std::string User(){std::string tmp = _ip + ":" + std::to_string(_port);return tmp;}private:std::string _ip;uint16_t _port;struct sockaddr_in _addr;
};

对读写异常方法进行封装

//HandlerConnection.hpp
#pragma once
#include "Connection.hpp"
#include "Log.hpp"using namespace log_ns;class HandlerConnection
{static const int gsize = 1024;public:HandlerConnection(handler server){_server = server;}// 就绪的connvoid HandlerRecver(Connection *conn){while (true){errno = 0;char bufferstr[gsize];ssize_t n = ::recv(conn->Sockfd(), bufferstr, sizeof(bufferstr), 0);if (n > 0){bufferstr[n] = 0;conn->AppendInbuffer(bufferstr);}else{if (errno == EWOULDBLOCK){break;}else if (errno == EINTR){continue;}else{conn->Excepter()(conn); // 统一处理异常return;                 // 这里是返回!}}}// 读取完成 - 进行内容解析(处理粘包问题)// 交给上层_server(conn);}void HandlerSender(Connection *conn){// 直接写while (true){errno = 0;ssize_t n = ::send(conn->Sockfd(), conn->Outbuffer().c_str(), conn->Outbuffer().size(), 0);if (n > 0){conn->DiscardOutbuffer(n);}else if (n == 0){break;}else{if (errno == EWOULDBLOCK){break;}else if (errno == EINTR){continue;}else{conn->Excepter()(conn);return;}}}// 一定是读条件不就绪(发送缓冲区满了)if (!conn->Outbuffer().empty()){// 开启读事件->剩下的数据epoll会自动帮我发完conn->R()->EnableReadWriteEvent(conn->Sockfd(), true, true);}else{conn->R()->EnableReadWriteEvent(conn->Sockfd(), true, false);}}void HandlerExcepter(Connection *conn){conn->R()->DelConnection(conn->Sockfd());}private:handler _server;
};

读事件要进行报文解析，读到完整请求后进行处理，生成应答后发出！

//Protocol.hpp
#pragma once
#include <iostream>
#include <memory>
#include <jsoncpp/json/json.h>// #define SELF 1// "len\r\n{json}\r\n" -- 自己设计出完整报文 len json的长度
// "\r\n" 第一个:区分len 和json边界 第二个:观察现象方便
const std::string sym = "\r\n";
const std::string space = " ";// jsonstr变成完整报文
std::string Encode(const std::string &jsonstr)
{int len = jsonstr.size();return std::to_string(len) + sym + jsonstr + sym;
}// 把json提取出来
// "len\r"
// "len\r\n{json}\r"
// "len\r\n{json}\r\n"
// "len\r\n{json}\r\n""len\r\n{json}\r\n""len\r\n{j"
std::string Decode(std::string &nameplate)
{size_t pos = nameplate.find(sym);if (pos == std::string::npos){return "";}std::string lenstr = nameplate.substr(0, pos);int len = std::stoi(lenstr);// 计算出完整报文长度int TotalLen = lenstr.size() + sym.size() + len + sym.size();if (nameplate.size() < TotalLen){return "";}// 读报文std::string jsonstr = nameplate.substr(pos + sym.size(), len);// 删报文nameplate.erase(0, TotalLen);return jsonstr;
}class req
{
public:req(){}req(int x, int y, std::string &sym): _x(x), _y(y), _sym(sym){}// 结构化->字符串void Serialize(std::string *out){
#ifdef SELF// 1.自己做 -> "_x _sym _y"// 2.使用现成库:jsoncppJson::Value root;root["x"] = _x;root["y"] = _y;root["sym"] = _sym;Json::FastWriter writer;*out = writer.write(root);
#else//"len\r\n {_x _sym _y} \r\n"std::string x = std::to_string(_x);std::string y = std::to_string(_y);*out = x + space + _sym + space + y;#endif}// 字符串->结构化bool Deserialize(std::string &in){
#ifdef SELFJson::Value root;Json::Reader reader;bool res = reader.parse(in, root);if (!res)return false;_x = root["x"].asInt();_y = root["y"].asInt();_sym = root["sym"].asString();return true;
#else//"len\r\n {_x _sym _y} \r\n"auto left_space = in.find(space);if (left_space == std::string::npos)return false;auto right_space = in.rfind(space);if (right_space == std::string::npos)return false;if (left_space + space.size() + 1 != right_space)return false;std::string x = in.substr(0, left_space);if (x.empty())return false;std::string sym = in.substr(left_space + space.size(), right_space);if (sym.empty())return false;std::string y = in.substr(right_space + space.size());if (y.empty())return false;_x = std::stoi(x.c_str());_sym = sym;_y = std::stoi(y.c_str());return true;#endif}void SetValue(int x, int y, char sym){_x = x;_y = y;_sym = sym;}~req() {}int _x;int _y;std::string _sym; //-x +-*/ _y
};class rep
{
public:rep(): _result(0), _code(0), _des("sucess"){}void Serialize(std::string *out){
#ifdef SELFJson::Value root;root["result"] = _result;root["code"] = _code;root["des"] = _des;Json::FastWriter writer;*out = writer.write(root);
#elsestd::string result = std::to_string(_result);std::string code = std::to_string(_code);*out = result + space + code + space + _des;
#endif}bool Deserialize(std::string &in){
#ifdef SELFJson::Value root;Json::Reader reader;bool res = reader.parse(in, root);if (!res)return false;_result = root["result"].asInt();_code = root["code"].asInt();_des = root["des"].asString();return true;
#elseauto left_space = in.find(space);if (left_space == std::string::npos)return false;auto right_space = in.rfind(space);if (right_space == std::string::npos)return false;if (left_space + space.size() + 1 != right_space)return false;std::string result = in.substr(0, left_space);if (result.empty())return false;std::string code = in.substr(left_space + space.size(), right_space);if (code.empty())return false;std::string des = in.substr(right_space + space.size());if (des.empty())return false;_result = std::stoi(result.c_str());_code = std::stoi(code.c_str());_des = des;return true;
#endif}~rep() {}int _result;int _code; // 0-> sucess 1->div zero 2->mod zero 3->failstd::string _des;
};// 工厂模式
class Factory
{
public:static std::shared_ptr<req> BuidRequest(){return std::make_shared<req>();}static std::shared_ptr<rep> BuidReponse(){return std::make_shared<rep>();}
};//PackageParse.hpp
#pragma once
#include <iostream>
#include <functional>#include "Log.hpp"
#include "Protocol.hpp"
#include "Connection.hpp"
using namespace log_ns;using protocol_t = std::function<std::shared_ptr<rep>(std::shared_ptr<req>)>;class PackageParse
{
public:PackageParse(protocol_t process): _process(process){}~PackageParse(){}void server(Connection *conn){while (true){// 报文解析 -- 不能保证读到完整的报文std::string message = ::Decode(conn->Inbuffer());if (message == "")break;// 反序列化 -- 能保证读到完整报文std::shared_ptr<req> q = Factory::BuidRequest();q->Deserialize(message);// 处理事务auto p = _process(q);// 序列化处理std::string result;p->Serialize(&result);LOG(DEBUG, "client request finish: %s\n", result.c_str());// 添加报头result = Encode(result);// 发送conn->AppendOutbuffer(result);// break;}// 至少处理了一个请求,有一个应答// 方法1：直接发 - 单进程if (!conn->Outbuffer().empty()) conn->Sender()(conn);// 方法2：激活对事件的关心即可,IO全都是由Reactor处理 -> 解决线程安全// 半同步半异步模式//if (!conn->Outbuffer().empty()) conn->R()->EnableReadWriteEvent(conn->Sockfd(),true,true);}private:protocol_t _process;
};//Netcal.hpp
#pragma once
#include "Protocol.hpp"// req -> rep
class Cal
{
public:Cal() {}~Cal() {}std::shared_ptr<rep> Count(std::shared_ptr<req> q){auto p = Factory::BuidReponse();const char *a = q->_sym.c_str();switch (*a){case '+':p->_result = q->_x + q->_y;break;case '-':p->_result = q->_x - q->_y;break;case '*':p->_result = q->_x * q->_y;break;case '/':{if (q->_y == 0){p->_result = -1;p->_code = 1;p->_des = "div zero";}else{p->_result = q->_x / q->_y;}break;}case '%':{if (q->_y == 0){p->_result = -1;p->_code = 2;p->_des = "mod zero";}else{p->_result = q->_x % q->_y;}break;}default:p->_result = -1;p->_code = 3;p->_des = "illegal operation";break;}return p;}
};

现象