高级I/O编程：从Select、Poll到Epoll的演进与Reactor服务器实现
高级I/O编程
1. 五种IO模型
I/O操作为什么慢？因为 I/O的核心是 等待 + 拷贝（比如 read 操作要先等数据再拷贝，write 同理），但访问外设（像网卡）时等待耗时太长，导致 IO 速度慢。
而高效 IO 的关键，是在单位时间里降低 IO 过程中的等待占比。
-
阻塞IO：在内核在将数据准备好之前，在系统调用处会一直等待，所有的套接字，默认都是阻塞方式。
-
非阻塞IO：如果内核还未将数据准备好，系统调用会直接返回，并且返回
EWOULDBLOCK错误码。- 非阻塞的方式需要通过循环不断 轮询（反复尝试读写文件描述符），轮询会大量消耗CPU资源。
-
信号驱动IO：内核将数据准备好的时候，使用 SIGIO 信号通知进程进行IO操作。
-
等待数据就绪的时间是异步的（由内核通过信号通知），但将数据从内核空间拷贝到用户空间的过程必须是同步完成的。
-
信号驱动I/O的核心缺陷是信号可能丢失或合并。
-
-
IO多路复用/多路转接：使用 select、poll、epoll 等系统调用监控多个文件描述符，当有事件就绪时，通知进程进行实际的IO操作。
-
异步IO：进程发起IO请求后立即返回，内核完成所有工作（等待数据 + 复制到用户空间），完成后通知进程。
-
与信号驱动的区别：
-
信号驱动：内核通知何时可以开始IO。
-
异步IO：内核通知IO操作已完成。
-
-
阻塞 VS 非阻塞： 阻塞会因为IO条件不具备，而导致阻塞等待，直到条件就绪；非阻塞检测到IO条件不具备，出错返回。本质的区别是等待数据就绪的方式不同，非阻塞 I/O 相比于阻塞 I/O 的效率提升，本质上源自于其在等待数据就绪期间能够执行其他任务，而非缩短了内核与用户空间之间的数据拷贝时间，实际数据拷贝量是完全相同的，这一环节在两种模型下是同等耗时且不可避免的。
为什么多路复用效率最高：
-
多路复用的效率飞跃，源于将串行等待并行化，从而消除了等待时间的累加效应。。
-
系统调用的开销减少。
-
减少进程/线程切换上下文的开销。
同步IO与异步IO：只要程序参与了I/O操作的 等待数据就绪 或 数据拷贝 中的任意一个阶段，就是同步I/O。只有这两个阶段完全由内核完成，才是真正的异步I/O。
2. 非阻塞 IO 之 fcntl
文件描述符默认都是阻塞IO。
fcntl系统调用： 对已打开的文件描述符进行各种控制操作。底层本质是修改内核中 struct file 的属性。
函数原型：
#include 根据cmd参数的不同，功能也对应不同。
参数：
-
fd: 要操作的文件描述符。 -
cmd: 控制命令，决定函数的行为。-
F_FUPFD: 复制文件描述符 -
F_GETFD/F_SETFD: 获得/设置文件描述符标记 -
F_GETFL/F_SETFL: 获得/设置文件状态标记 -
F_GETOWN/F_SETOWN: 获得/设置异步I/O所有权
-
实现将文件描述符设置为非阻塞的函数SetBlock
void SetNonBlock(int fd) {
int fl = fcntl(fd, F_GETFL);
if (fl < 0) {
perror("fcntl");
return;
}
fcntl(fd, F_SETFL, fl | O_NONBLOCK); // O_NONBLOCK：将fd设置为非阻塞
}
int main()
{
SetNonBlock(0);
char buffer[1024];
while (true)
{
// Linux中ctrl + d：标识输入结束，read返回值是0，类似读到文件结尾
ssize_t n = read(0, buffer, sizeof(buffer) - 1);
if (n > 0)
{
buffer[n - 1] = 0; // 消除换行符的同时 + ' '
std::cout << buffer << std::endl;
}
else if (n < 0)
{
// 1. 读取出错 2. 底层没有数据准备好
// eagain ewouldblock
if(errno == EAGAIN || errno == EWOULDBLOCK) // 错误码
{
std::cout << "数据没有准备好..." << std::endl;
sleep(1);
// 做你的事情
continue;
}
else if(errno == EINTR) // read被信号打断
{
continue;
}
else
{
// 真正的read出错了
}
}
else
{
break;
}
}
}
-
F_GETFL返回值文件状态标志的位图。 -
EINTR（Error Interrupted）表示系统调用被信号中断（read被中断）。 -
EAGAIN / EWOULDBLOCK表示资源没有就绪。
3. select
3.1 接口介绍
I/O = 等 + 拷贝，select 是 I/O 多路复用中常用的系统调用，用于同时等待多个fd的事件就绪通知机制。
-
select主要负责等。
-
可读：底层有数据，读事件就绪。
-
可写：底层有空间，写事件就绪。
函数原型：
#include 参数：
-
nfds: 需要监听的最大文件描述符值 + 1。 -
readfds、writefds、exceptfds: 输入输出型参数，分别对应需要检测的文件描述符，可读事件的集合、可写事件的集合及异常事件的集合。-
设置的时候：用户告诉内核，需要监听哪些fd上面的对应事件。
-
返回的时候：内核告诉用户，监听的fd上面的对应事件已经就绪。
-
fd_set类型：是一个位图结构。低位 -> 高位分别代表文件描述符0、1、2…，位为1表示文件描述符在该事件集合中，位为0表示文件描述符不在该事件集合中。同时，还提供了操作该位图类型的接口。/* fd_set 的典型实现（不同系统可能略有差异） */ /* 在 /usr/include/sys/select.h 或类似位置 */ /* 定义最大文件描述符数 */ #define FD_SETSIZE 1024 /* 默认值，可通过编译时修改 */ typedef long int __fd_mask; /* fd_set 结构定义 */ #define __NFDBITS (8 * (int) sizeof (__fd_mask)) typedef struct { __fd_mask __fds_bits[FD_SETSIZE / __NFDBITS]; /* 或者在某些系统上是： */ /* __fd_mask fds_bits[howmany(FD_SETSIZE, __NFDBITS)]; */ } fd_set; -
fd_set类型最大可以监听的文件描述符个数：sizeof(fd_set) * 8 = 1024根据系统版本的不同，该值可能会有变化。但是是有上限的。 -
fd_set操作接口：// 用来清除 fd_set 类型的全部比特位 void FD_ZERO(fd_set *set); // 用来设置 fd_set 类型中相关 fd 的比特位 void FD_SET(int fd, fd_set *set); // 用来清除 fd_set 类型中相关 fd 的比特位 void FD_CLR(int fd, fd_set *set); // 用来检测 fd_set 类型中相关 fd 的比特位是否为真 int FD_ISSET(int fd, fd_set *set);
-
-
timeout: 设置 select 的超时时间。struct timeval { time_t tv_sec; /* Seconds */ suseconds_t tv_usec; /* Microseconds */ }-
nullptr: select 阻塞等待，直到某个文件描述符上某个事件就绪。 -
{0 , 0}: 非阻塞等待，立即返回。 -
> 0: 时间内阻塞等待，超时非阻塞返回。
-
返回值：
-
> 0: 在监控的文件描述符中，至少有一个或多个描述符的就绪事件发生。返回值数字 = 所有集合中已就绪的描述符总数。- 假设：描述符3读就绪、描述符4读就绪 + 写就绪，select返回值 = 3。
-
= 0: timeout 设置的时间超时或非阻塞方式等待但没有事件就绪时。 -
-1: select 出错，errno被设置。
3.2 SelectServer.hpp
#pragma once
#include "Socket.hpp"
#include 3.3 select优缺点
特点：
-
跨平台兼容性好：几乎所有操作系统都支持，代码可移植性强。
-
需要构建辅助数组，将监视的文件描述符永久保存到辅助数组中，因为 readfds、writefds、exceptfds 是输入输出型参数， 当 select 返回后，fd_set 集合中只保留就绪的文件描述符，原始集合被破坏。
缺点：
-
事件就绪和设置事件监听不是分离的，通过输入输出型参数同时控制，因此每次调用 select 时，都需要手动重新设置监听的 fd 集合。
-
内核底层也会通过 nfds 和用户设置的监听文件描述符事件 遍历文件描述符数组。效率比较低。
-
select 支持的文件描述符数量有上限。
4. poll
4.1 接口介绍
poll 是 select 的改进版本，解决了 select 一些问题，但仍然有瓶颈。
函数原型：
#include 参数：
-
fds: 一个结构体数组。-
struct pollfd类型：struct pollfd { int fd; /* 文件描述符 */ short events; /* 等待的事件 是一个位图通过按位或的方式 */ short revents; /* 实际发生的事件 是一个位图通过按位或的方式 */ }; -
调用的时候：fd + events 有效，用户告诉内核，监听 fd 对应的 events 事件。
-
返回的时候：fd + revents 有效，内核告诉用户，fd 的 revents 事件就绪。
-
-
nfds: fds的数组长度。 -
timeout: 超时时间。-
-1: 阻塞等待，直到有事件就绪。 -
0: 立即返回，非阻塞模式。 -
> 0: 等待的毫秒数。
-
返回值：
-
> 0: 在监控的文件描述符中，至少有一个或多个描述符的就绪事件发生。返回值数字 = 已就绪的描述符总数。- 假设：描述符3读就绪、描述符4读就绪 + 写就绪，poll返回值 = 2。
-
= 0: timeout 设置的时间超时或非阻塞方式等待但没有事件就绪时。 -
-1: poll 出错，errno被设置。
events 与 revents 的取值：
| 事件 | 描述 | 是否可作为输入 | 是否可作为输出 |
|---|---|---|---|
| POLLIN | 数据（包括普通数据和优先数据）可读 | 是 | 是 |
| POLLRDNORM | 普通数据可读 | 是 | 是 |
| POLLRDBAND | 优先级带数据可读（Linux 不支持） | 是 | 是 |
| POLLPRI | 高优先级数据可读，比如 TCP 带外数据 | 是 | 是 |
| POLLOUT | 数据（包括普通数据和优先数据）可写 | 是 | 是 |
| POLLWRNORM | 普通数据可写 | 是 | 是 |
| POLLWRBAND | 优先级带数据可写 | 是 | 是 |
| POLLRDHUP | TCP 连接被对方关闭，或者对方关闭了写操作（GNU 引入） | 是 | 是 |
| POLLERR | 错误 | 否 | 是 |
| POLLHUP | 挂起（如管道写端关闭后，读端描述符将收到 POLLHUP 事件） | 否 | 是 |
| POLLNVAL | 文件描述符没有打开 | 否 | 是 |
4.2 PollServer.hpp
#pragma once
#include "Socket.hpp"
#include 4.3 poll的优缺点
优点：
-
不同于 select 使用三个位图方式，poll 使用一个结构体指针数组实现。
-
poll 通过 events 和 revents 将设置监听与返回时事件就绪分离，无需每次调用设置。
-
poll 没有 fd 数量限制。
缺点：
- 无论用户还是内核都需要大量的遍历操作，随着监听的文件描述符数量的增多，效率也会线性下降。
5. epoll
epoll 是 Linux 内核实现的高效 I/O 事件通知机制，用于处理大量文件描述符的 I/O 事件。它是 select/ poll 的增强版本。
它几乎具备了之前所说的一切优点，被公认为Linux2.6下性能最好的多路I/O就绪通知方法.
5.1 epoll_create
创建 epoll 实例，内核中创建对应的数据结构。
函数原型：
#include 参数：
size: 该参数在现代内核中已废弃，但仍需传递大于 0 的值。
返回值：
-
成功：返回新的 epoll 文件描述符。
-
失败：返回 -1，并设置
errno。
5.2 epoll_ctl
向 epoll 实例中添加、修改、删除要监控的文件描述符。内核中注册回调机制。
函数原型：
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
参数：
-
epfd: epoll 实例的文件描述符（由epoll_create()返回）。 -
op: 操作类型，必须是以下之一：-
EPOLL_CTL_ADD: 添加新的文件描述符到 epoll 实例中。 -
EPOLL_CTL_MOD: 修改已注册的文件描述符的监听事件。 -
EPOLL_CTL_DEL: 从 epoll 实例中删除文件描述符及事件。
-
-
fd: 要操作的目标文件描述符。 -
event: 事件配置的结构体指针。struct epoll_event:
typedef union epoll_data { void *ptr; // 用户自定义指针 int fd; // 文件描述符（常用） uint32_t u32; // 32位整数 uint64_t u64; // 64位整数 } epoll_data_t; struct epoll_event { uint32_t events; // Epoll 事件（位掩码） epoll_data_t data; // 用户数据（事件触发时返回） };
events可以是如下的取值：
-
EPOLLIN: 表示对应的文件描述符可以读 (包括对端SOCKET正常关闭)；
-
EPOLLOUT: 表示对应的文件描述符可以写；
-
EPOLLPRI: 表示对应的文件描述符有紧急的数据可读 (这里应该表示有带外数据到来)；
-
EPOLLERR: 表示对应的文件描述符发生错误；
-
EPOLLHUP: 表示对应的文件描述符被挂断；
-
EPOLLET: 将EPOLL设为边缘触发(Edge Triggered)模式，默认为水平触发(Level Triggered)。
-
EPOLLONESHOT：只监听一次事件，当监听完这次事件之后，如果还需要继续监听这个socket，需要
epoll_ctl重新添加。
返回值：
-
成功：返回0。
-
失败: 返回 -1 ，并设置
errno。
5.3 epoll_wait
等待被监控的文件描述符上事件就绪，并返回就绪的事件列表。
函数原型：
int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);
参数：
-
epfd: epoll 实例的文件描述符（由epoll_create()返回）。 -
events: 用于接收就绪事件的数组指针，由调用者分配内存。 -
maxevents: 每次调用最多返回的事件数量，必须大于 0，且不能超过 events 数组大小。 -
timeout: 超时时间（毫秒）-
-1: 阻塞等待，直到有时间发生。 -
0: 立即返回，即使没有时间就绪。 -
> 0: 等待指定毫秒数后返回。
-
返回值：
-
成功：返回就绪的文件描述符数量。
-
失败：返回 -1，并设置
errno。 -
超时：返回0。
5.4 epoll 原理
内核中的 epoll 模型：
// 内核源码中的关键结构（简化版）
struct eventpoll {
// 红黑树根节点：存储所有监控的 fd
struct rb_root rbr;
// 双向链表：存储就绪的 fd
struct list_head rdllist;
// 等待队列：存放调用 epoll_wait 而阻塞的进程
wait_queue_head_t wq;
// 互斥锁
struct mutex mtx;
// 其他字段...
};
// 红黑树中和就绪队列中的节点
struct epitem {
// 红黑树节点
struct rb_node rbn;
// 双向链表节点（用于就绪链表）
struct list_head rdllink;
// 监控的文件描述符
int fd;
// 监控的事件
uint32_t events;
// 用户数据
epoll_data_t data;
// 指向所属的 eventpoll
struct eventpoll *ep;
// 等待队列项（用于回调）
struct wait_queue_entry wq;
};
-
epoll 模型的核心由三部分构成：用于管理文件描述符及事件的红黑树、存放就绪事件的就绪队列、以及当事件发生时由内核触发的回调函数。
-
每一个 epoll 对象都有一个独立的 eventpoll 结构体，其中包含了红黑树和就绪队列。
-
红黑树的本质是允许用户主动向内核注册特定文件描述符及其关注的事件类型。
-
这些事件会以节点的方式挂载到红黑树上，因此效率提高了。
-
红黑树的 key 是 fd。
-
-
就绪队列本质是内核告诉用户哪些文件描述符的哪些事件已就绪。
-
用户、内核、就绪队列，可以理解为基于事件就绪的生产者消费者模型。
-
获取就绪事件，如果缓冲区大小不够怎么办？因为是生产者消费者模型，下次继续拿。
-
检测是否有事件就绪，时间复杂度O(1)，只需要关心就绪队列是否为空即可。
-
-
eventpoll 中包含了互斥锁和条件变量，因此，epoll 是线程安全的。
-
红黑树中的每一个节点都会被注册一个回调机制，当该文件描述符上的事件就绪，会自动调用回调，激活红黑树中的节点到就绪队列中。只有指针操作，时间复杂度O（1）。
-
epitem 节点既可以属于红黑树，又可以属于就绪队列。
-
回调机制也被通过链表管理起来了，因为同一个文件描述符可能会被不同的 epoll 实例同时监控。
-
-
epoll_wait 中的 events，内核会严格按照下标0开始，依次拷贝，应用层不需要向 select 或 poll 一样进行非法检测。
-
epoll_create 内核中在创建 eventpoll 结构体，本质是创建内核数据结构（红黑树和就绪队列）。
-
epoll_ctl 内核中在红黑树中增加、修改、或删除节点，添加节点时向该节点注册回调方法。
-
为什么 epoll_create 返回值是一个文件描述符，如何根据文件描述符找到 eventpoll 结构？
- struct files_struct -> fd_array[4] -> struct file -> void* private_data -> struct eventpoll
5.5 epoll 的特点
-
文件描述符没有数量限制。
-
解决了用户设置监听事件与内核告诉用户哪些事件就绪的分离。
-
通过事件回调机制，避免使用遍历。
5.6 epoll 的LT模式与ET模式
epoll 默认模式为 LT 模式。设置 ET 模式将 events 添加 EPOLLET 标志位。
LT（Level Trigger）：水平触发，只要有事件就绪，就一直通知你。
-
有10个就绪，你只读走5个，之后会一直通知
-
如果10kb数据，只读走2kb数据，则之后会一直通知。
ET（Edge Trigger）：边缘触发，只有当事件就绪状态变化时，才通知你一次。
-
有10个就绪，只读走5个，如果没有新的事件就绪，则不会继续通知。
-
如果10kb数据，只读走2kb数据，除非有新的数据到来，否则不会通知。
总结：
-
使用 ET 模式必须一次性读完所有数据，否则可能会造成数据丢失。
-
在ET模式下，用户程序无法预知底层缓冲区是否读完，因此必须使用非阻塞I/O进行循环读取，直到 recv 返回 EAGAIN 错误（表示缓冲区已空），否则剩余数据将无法再次触发事件通知。
-
ET 比 LT 更高效
-
ET通知效率更高，因为ET每次都是有效通知。
-
ET迫使上层尽快读完所有的数据，可以给对方更新一个更大的win窗口（tcp接收缓冲区大小），提高对方滑动窗口的大小，提高 TCP 的传输效率。
-
LT 也可以和 ET 类似循环读取，为什么还要有 ET？
增加 IO 读写方式的确定性，使用 ET 模式是 OS 在约束程序员，如果一次性不读完则代码是存在bug的。若使用 LT 模式，没有人约束程序员，存在不确定性。
6. 多路转接设置监听写事件
-
读事件默认是不就绪的，写事件是默认就绪的。
-
读事件监听要常设，而写事件监听是按需设置的。
-
当发送失败时，再将写事件的监听设置到 epoll 内部。
-
手动开启 EPOLLOUT 的监听时，默认就会触发一次写事件就绪（即使发送缓冲区是满的·）。
7. 基于 Reactor 反应堆模式的服务器
7.1 Common.hpp
#pragma once
#include 7.2 Mutex.hpp
// Mutex.hpp
#pragma once
#include 7.3 Log.hpp
#pragma once
#include 7.4 InetAddr.hpp
#pragma once
#include 7.5 Socket.hpp
#pragma once
#include "Common.hpp"
#include "Log.hpp"
#include "InetAddr.hpp"
namespace SocketModule{
const static int default_backlog = 16;
// 设计为虚基类，TCP和UDP的模板类
class Socket{
public:
virtual ~Socket() {}
virtual void socket() = 0;
virtual void bind(uint16_t server_port) = 0;
virtual void listen(int backlog) = 0;
// virtual std::shared_ptr accept(InetAddr* client_addr) = 0;
virtual int accept(InetAddr* client_addr) = 0;
virtual void close() = 0;
virtual int recv(std::string* res) = 0;
virtual int send(const std::string& message) = 0;
virtual void connect(const InetAddr& server_addr) = 0;
virtual int getfd() = 0;
public:
void buildTcpListenSocket(uint16_t server_port , int backlog = default_backlog) {
socket();
bind(server_port);
listen(default_backlog);
}
void buildTcpClientListendSocket() {
socket();
}
};
const static int default_sockfd = -1;
class TcpSocket : public Socket{
public:
TcpSocket() :_sockfd(default_sockfd) {} // 初始化sockfd为listenfd
// TcpSocket(int acceptfd) :_sockfd(acceptfd) {} // 初始化sockfd为accpetfd
virtual void socket() override {
_sockfd = ::socket(AF_INET , SOCK_STREAM , 0);
if(_sockfd < 0) {
LogModule::LOG(LogModule::LogLevel::FATAL) << "socket create error";
exit(SOCKET_ERROR);
}
LogModule::LOG(LogModule::LogLevel::INFO) << "socket create success - sockfd: " << _sockfd;
}
virtual void bind(uint16_t server_port) override {
InetAddr server_addr(server_port);
int n = ::bind(_sockfd , server_addr.getSockaddr() , server_addr.getSockaddrLen());
if(n < 0) {
LogModule::LOG(LogModule::LogLevel::FATAL) << "bind error";
exit(BIND_ERROR);
}
LogModule::LOG(LogModule::LogLevel::INFO) << "bind success - sockfd: " << _sockfd;
}
virtual void listen(int backlog) override {
int n = ::listen(_sockfd , backlog);
if(n < 0) {
LogModule::LOG(LogModule::LogLevel::FATAL) << "listen error";
exit(LISTEN_ERROR);
}
LogModule::LOG(LogModule::LogLevel::INFO) << "listen success - sockfd: " << _sockfd;
}
// virtual std::shared_ptr accept(InetAddr* client_addr) override {
int accept(InetAddr* client_addr) override {
struct sockaddr_in client;
socklen_t addrlen = sizeof(client);
int acceptfd = ::accept(_sockfd , (struct sockaddr*)&client , &addrlen);
if(acceptfd < 0) {
return -1;
}
client_addr->setSockaddrIn(client);
return acceptfd;
// client_addr->setSockaddrIn(client); // 输出型参数
// return std::make_shared(acceptfd); // 多个服务器可以共享一个客户端的链接
}
virtual void close() override {
if(_sockfd > 0)
::close(_sockfd); // close(listenfd) 或 close(accpetfd)
}
// recv返回值和::recv返回值保持一致
virtual int recv(std::string* res) override {
char buffer[4096];
ssize_t n = ::recv(_sockfd , buffer , sizeof(buffer) , 0);
*res += buffer; // 读取可能会不完整
return n;
}
virtual int send(const std::string& message) override {
ssize_t n = ::send(_sockfd , message.c_str() , message.size() , 0);
return n;
}
virtual void connect(const InetAddr& server_addr) override {
int n = ::connect(_sockfd , server_addr.getSockaddr() , server_addr.getSockaddrLen());
if(n < 0) {
LogModule::LOG(LogModule::LogLevel::FATAL) << "connect server error";
exit(CONNECT_ERROR);
}
}
virtual int getfd() override { return _sockfd; }
private:
int _sockfd; // listenfd 或 accpetfd
};
}
7.6 Epoller.hpp
#pragma once
#include 7.7 Reactor.hpp
#pragma once
#include 7.8 Connection.hpp
#pragma once
#include 7.9 Listener.hpp
#pragma once
#include 7.10 Channel.hpp（多路转接写事件的具体写法）
#pragma once
#include "Connection.hpp"
#include "Reactor.hpp"
#define RECV_SIZE 4096
class Channel : public Connection{
public:
Channel(int sockfd , const InetAddr& client)
:_sockfd(sockfd)
,_client_addr(client)
{
// 1. 设置 Channel 监听的事件和开启 ET 模式
setEvents(EPOLLIN | EPOLLET); // todo
// 2. 设置文件描述符为非阻塞
setNonBlock(_sockfd);
}
virtual int getSockfd() override { return _sockfd; }
virtual void recver() override {
char buffer[RECV_SIZE];
while(true) {
buffer[0] = 0; // 清空字符串
ssize_t n = recv(_sockfd , buffer , RECV_SIZE - 1 , 0); // 非阻塞方式读取
if(n > 0) {
buffer[n] = 0;
_inbuffer += buffer;
} else if(n == 0) {
// 客户端关闭 异常处理
except();
return;
} else {
if(errno == EAGAIN) {
// 非阻塞读取完毕
break;
} else if(errno == EINTR) {
// 读取被信号打断
continue;
} else {
// recv error
LogModule::LOG(LogModule::LogLevel::ERROR) << "channel recv error";
except();
return;
}
}
}
LogModule::LOG(LogModule::LogLevel::DEBUG) << "Channel inbuffer: " << _inbuffer;
// ET 非阻塞读取数据完毕
// 1. 判断是否报文是否是一个完整的报文，如果是多个报文？数据不完整问题或TCP粘包问题
if(!_inbuffer.empty()) {
// 将数据交付给 Protocol 层处理
_outbuffer += _callback(_inbuffer);
}
// 2. 不为空发送
if(!_outbuffer.empty()) {
sender();
}
}
virtual void sender() override {
while(true) {
ssize_t n = send(_sockfd , _outbuffer.c_str() , _outbuffer.size() , 0);
if(n >= 0) {
// n 表示实际发送的字节
_outbuffer.erase(0 , n);
if(_outbuffer.empty())
break;
} else {
// 写事件这里的 errno == EAGAIN 表示的意思是发送缓冲区已满
if(errno == EAGAIN || errno == EWOULDBLOCK)
break;
else if(errno == EINTR)
continue;
else {
except();
return;
}
}
}
// 1.数据发送完毕
// 2.发送缓冲区已满，发送条件不具备
if(!_outbuffer.empty()) {
// 开启对写事件的关心
getReactorPtr()->openWriteReadEvents(_sockfd , true , true);
} else {
// 关闭对写事件的关心
getReactorPtr()->openWriteReadEvents(_sockfd , false , true);
}
}
virtual void except() override {
// 处理读出错、写出错、客户端关闭
// 本质都是关闭链接
getReactorPtr()->delConnection(_sockfd);
}
private:
int _sockfd;
std::string _inbuffer;
std::string _outbuffer;
InetAddr _client_addr;
};
7.11 Protocol.hpp
#pragma once
#include 7.12 Calculate.hpp
#pragma once
#include "Protocol.hpp"
class Calculate{
public:
Response execute(const Request& req) {
switch(req.get_oper()) {
case '+':
return Response(req.get_x() + req.get_y() , true);
case '-':
return Response(req.get_x() - req.get_y() , true);
case '*':
return Response(req.get_x() * req.get_y() , true);
case '/':
{
if(req.get_y() == 0)
return Response(0, false);
else
return Response(req.get_x() / req.get_y() , true);
}
default:
LogModule::LOG(LogModule::LogLevel::WARNING) << "未知的操作符";
}
return Response(0 , false);
}
};
7.13 Main.cc
#include "Calculate.hpp"
#include "Protocol.hpp"
#include "Listener.hpp"
#include "Reactor.hpp"
// reactorServer port
int main(int argc , char* argv[]) {
if(argc != 2) {
std::cout << argv[0] << " port" << std::endl;
exit(USAGE_ERROR);
}
uint16_t port = std::stoi(argv[1]);
LogModule::ENABLE_CONSOLE_FLUSH_STRATEGY();
// 1. 应用层
Calculate cal;
// 2. 表示层
Protocol protocol([&cal](const Request& req) -> Response {
LogModule::LOG(LogModule::LogLevel::DEBUG) << "应用层处理报文";
return cal.execute(req);
});
// 3. 会话层
std::shared_ptr<Connection> listener_ptr = std::make_shared<Listener>(port);
listener_ptr->registerCallback([&protocol](std::string& inbuffer)->std::string{
LogModule::LOG(LogModule::LogLevel::DEBUG) << "表示层处理报文开始";
// 1.能否读取出一个完整的报文
// 2.能否处理多个报文
std::string outbuffer;
while(true) {
std::string request_json;
if(!protocol.decode(inbuffer , request_json)) // 不存在完整的报文
break;
// 来到这里，response_json 里100%存在一个完整的报文
outbuffer += protocol.execute(request_json);
}
LogModule::LOG(LogModule::LogLevel::DEBUG) << "表示层处理报文结束";
return outbuffer;
});
std::unique_ptr<Reactor> reactor_server_ptr = std::make_unique<Reactor>();
reactor_server_ptr->addConnection(listener_ptr);
reactor_server_ptr->start();
return 0;
}
7.14 Makefile
reactorServer: Main.cc
g++ -o $@ $^ -std=c++17 -ljsoncpp
.PHONY:clean
clean:
rm -f reactorServer
