---

一、五种IO模型

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

非阻塞IO: 如果内核还未将数据准备好, 系统调用仍然会直接返回, 并且返回EWOULDBLOCK错误码.非阻塞IO往往需要程序员循环的方式反复尝试读写文件描述符, 这个过程称为轮询. 这对CPU来说是较大的浪费, 一般只有特定场景下才使用.

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

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

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

任何IO过程中, 都包含两个步骤. 第一是等待, 第二是拷贝. 而且在实际的应用场景中, 等待消耗的时间往往都远远高于拷贝的时间. 让IO更高效, 最核心的办法就是让等待的时间尽量少。我们用的最多的是多路复用。

同步通信 vs 异步通信

同步和异步关注的是消息通信机制.
所谓同步，就是在发出一个调用时，在没有得到结果之前，该调用就不返回. 但是一旦调用返回，就得到返回值了; 换句话说，就是由调用者主动等待这个调用的结果;
异步则是相反，调用在发出之后，这个调用就直接返回了，所以没有返回结果; 换句话说，当一个异步过程调用发出后，调用者不会立刻得到结果; 而是在调用发出后，被调用者通过状态、通知来通知调用者，或通过回调函数处理这个调用.
另外, 我们回忆在讲多进程多线程的时候, 也提到同步和互斥. 这里的同步通信和进程之间的同步是完全不想干的概念.
进程/线程同步也是进程/线程之间直接的制约关系
是为完成某种任务而建立的两个或多个线程，这个线程需要在某些位置上协调他们的工作次序而等待、
传递信息所产生的制约关系. 尤其是在访问临界资源的时候.

阻塞 vs 非阻塞

阻塞和非阻塞关注的是程序在等待调用结果（消息，返回值）时的状态.
阻塞调用是指调用结果返回之前，当前线程会被挂起. 调用线程只有在得到结果之后才会返回.
非阻塞调用指在不能立刻得到结果之前，该调用不会阻塞当前线程

非阻塞IO，纪录锁，系统V流机制，I/O多路转接（也叫I/O多路复用）,readv和writev函数以及存储映射IO（mmap），这些统称为高级IO，重点讨论多路转接。

二、其他高级IO

非阻塞IO

SetNoBlock

void SetNoBlock(int fd) { int fl = fcntl(fd, F_GETFL); if (fl < 0) { perror("fcntl");return; }fcntl(fd, F_SETFL, fl | O_NONBLOCK); 
}

使用F_GETFL将当前的文件描述符的属性取出来(这是一个位图).
然后再使用F_SETFL将文件描述符设置回去. 设置回去的同时, 加上一个O_NONBLOCK参数.

轮询方式读取输入标准：

#include <iostream>
#include <unistd.h>
#include <fcntl.h>
#include <cstdio>
#include <cstring>
#include <cerrno>void SetNonBlock(int fd)
{int fl = fcntl(fd, F_GETFL);if (fl < 0){perror("fcntl create error");return;}fcntl(fd, F_SETFL, fl | O_NONBLOCK);std::cout << "set nonblock down" << fd << std::endl;return;
}int main()
{char buffer[1024];SetNonBlock(0);sleep(1);while (true){// printf("Please Enter#");// fflush(stdout);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) // 对端网络连接关了{std::cout << "read down " << std::endl;break;}else{std::cerr << "read error " << "n = " << n << " read errno: "<< errno << " str err: " << strerror(errno) << std::endl;sleep(1);// break;}}return 0;
}

I/O多路转接之select

系统提供select函数来实现多路复用输入/输出模型.
select系统调用是用来让我们的程序监视多个文件描述符的状态变化的;
程序会停在select这里等待，直到被监视的文件描述符有一个或多个发生了状态改变;

select函数原型

select的函数原型如下: #include <sys/select.h>

int select(int nfds, fd_set *readfds, fd_set *writefds,
fd_set *exceptfds, struct timeval *timeout);

参数解释:
参数nfds是需要监视的最大的文件描述符值+1；
rdset,wrset,exset分别对应于需要检测的可读文件描述符的集合，可写文件描述符的集 合及异常文件描述符的集合;
参数timeout为结构timeval，用来设置select()的等待时间.如果设置了timeout就是输入输出参数

参数timeout取值:
NULL：则表示select（）没有timeout，select将一直被阻塞，直到某个文件描述符上发生了事件;阻塞等待。
0：仅检测描述符集合的状态，然后立即返回，并不等待外部事件的发生。
特定的时间值：如果在指定的时间段里没有事件发生，select将超时返回。非阻塞的一种。

返回值：
返回值>0表示有多个fd就绪了。返回值==0表示超时，等待过程中没有错误，但是也没有fd就绪。 返回值<0等等出错了。

struct timeval: 给select设置等待方式，每隔timeval时间等待一次阻塞等待。

fd_set（位图）

select代码

selectserver.hpp:

#pragma once
#include <iostream>
#include <sys/select.h>
#include <sys/time.h>
#include "Socket.hpp"
#include "Log.hpp"static const uint16_t defaultport = 8080;
static const int fd_num_max = (sizeof(fd_set) * 8);
int defultfd = -1;class SelectServer
{
private:Sock _listensock; // 监听套接字uint16_t _port;int fd_array[fd_num_max];public:SelectServer(uint16_t port = defaultport): _port(port){for (int i = 0; i < fd_num_max; i++){fd_array[i] = defultfd;// std::cout << "fd_array[" << i << "]: " << fd_array[i] << std::endl;}}bool Init(){_listensock.Socket(); // 创建套接字_listensock.Bind(_port);_listensock.Listen();return true;}void Accepter(){// 连接事件就绪了std::string clientip;uint16_t clientport = 0;int sock = _listensock.Accept(&clientip, &clientport); // 不会阻塞if (sock < 0)return;lg(Info, "accept successful, %s: %d, fd: %d", clientip.c_str(), clientport, sock);int pos = 1;for (; pos < fd_num_max; pos++){if (fd_array[pos] != defultfd)continue;elsebreak;}if (pos == fd_num_max){// 到尽头了lg(Warning, "fd is full...");close(sock);}else{fd_array[pos] = sock;PrintFd();}}void Recver(int fd, int pos){char buffer[1024];ssize_t n = read(fd, buffer, sizeof(buffer) - 1);if (n > 0){buffer[n] = 0;std::cout << "get a massage: " << buffer << std::endl;}else if (n == 0){lg(Info, "client quit...close fd is: %d", fd);close(fd);fd_array[pos] = defultfd; // 本质是从select中移除了}else{lg(Warning, "recv err, fd is: %d", fd);close(fd);fd_array[pos] = defultfd; // 本质是从select中移除了}}void HandlerEvent(fd_set &rfds){for (int i = 0; i < fd_num_max; i++){int fd = fd_array[i];if (fd == defultfd)continue;if (FD_ISSET(fd, &rfds)) // 读事件就绪{if (fd == _listensock.Fd()) // 监听套接字{Accepter(); // 连接管理器}else // 非监听套接字，要开始读取啦{Recver(fd, i);}}}}void Start(){int listensock = _listensock.Fd();fd_array[0] = listensock;while (true){fd_set rfds;FD_ZERO(&rfds); // 清空int maxfd = fd_array[0];for (int i = 0; i < fd_num_max; i++){if (fd_array[i] == defultfd){continue;}FD_SET(fd_array[i], &rfds);if (maxfd < fd_array[i]){maxfd = fd_array[i]; // 动态设定最大文件描述符lg(Info, "maxfd is:%d", maxfd);}}// 不能直接accept，因为accept是监测listensocket上的事件，新连接的到来，相当于读事件的就绪，这里要用selectstruct timeval timeout; // 输入输出 可能要进行周期的重复设置；如果事件就绪，上层一直不处理select会一直通知你timeout.tv_sec = 5;timeout.tv_usec = 0;int n = select(maxfd + 1, &rfds, nullptr, nullptr, /*&timeout*/ nullptr);switch (n){case 0:// 没有描述符就绪std::cout << "time out, timeout:" << timeout.tv_sec << ": " << timeout.tv_usec << std::endl;break;case -1:std::cerr << "select err " << std::endl;break;default:// 有事件就绪了std::cout << "get a new link" << std::endl;HandlerEvent(rfds);break;}}}void PrintFd(){std::cout << "fd list: ";for (int i = 0; i < fd_num_max; i++){if (fd_array[i] == defultfd)continue;std::cout << fd_array[i] << " ";}std::cout << std::endl;}~SelectServer(){_listensock.Close();}
};

select缺点

poll（用select修改）

fds是一个poll函数监听的结构列表. 每一个元素中, 包含了三部分内容: 文件描述符, 监听的事件集合, 返回的事件集合.
nfds表示fds数组的长度.
timeout表示poll函数的超时时间, 单位是毫秒(ms).

events和revents的取值:

返回结果
返回值小于0, 表示出错;
返回值等于0, 表示poll函数等待超时;
返回值大于0, 表示poll由于监听的文件描述符就绪而返回.

poll的优点
不同与select fdset的方式，poll使用一个pollfd的指针实现.使用三个位图来表示三个pollfd结构包含了要监视的event和发生的event，不再使用select“参数-值”传递的方式. 接口使用比select更方便.
poll并没有最大数量限制 (但是数量过大后性能也是会下降).

poll的缺点
poll中监听的文件描述符数目增多时和select函数一样，poll返回后，需要轮询pollfd来获取就绪的描述符.
每次调用poll都需要把大量的pollfd结构从用户态拷贝到内核中.
同时连接的大量客户端在一时刻可能只有很少的处于就绪状态, 因此随着监视的描述符数量的增长, 其效率也会线性下降.

#pragma once 
#include <iostream>
#include <poll.h>
#include <sys/time.h>
#include "Socket.hpp"
#include "Log.hpp"static const uint16_t defaultport = 8080;
static const int fd_num_max = 64;
int defultfd = -1;
int non_event = 0;
class PollServer
{
private:Sock _listensock; // 监听套接字uint16_t _port;struct pollfd _event_fds[fd_num_max];// int fd_array[fd_num_max];public:PollServer(uint16_t port = defaultport): _port(port){for (int i = 0; i < fd_num_max; i++){_event_fds[i].fd = defultfd;_event_fds[i].events = non_event;_event_fds[i].revents = non_event;// std::cout << "fd_array[" << i << "]: " << fd_array[i] << std::endl;}}bool Init(){_listensock.Socket(); // 创建套接字_listensock.Bind(_port);_listensock.Listen();return true;}void Accepter(){// 连接事件就绪了std::string clientip;uint16_t clientport = 0;int sock = _listensock.Accept(&clientip, &clientport); // 不会阻塞if (sock < 0)return;lg(Info, "accept successful, %s: %d, fd: %d", clientip.c_str(), clientport, sock);int pos = 1;for (; pos < fd_num_max; pos++){if (_event_fds[pos].fd != defultfd)continue;elsebreak;}if (pos == fd_num_max){// 到尽头了lg(Warning, "fd is full...");close(sock);}else{_event_fds[pos].fd = sock;_event_fds[pos].events = POLLIN;PrintFd();}}void Recver(int fd, int pos){char buffer[1024];ssize_t n = read(fd, buffer, sizeof(buffer) - 1);if (n > 0){buffer[n] = 0;std::cout << "get a massage: " << buffer << std::endl;}else if (n == 0){lg(Info, "client quit...close fd is: %d", fd);close(fd);_event_fds[pos].fd = defultfd; // 本质是从select中移除了}else{lg(Warning, "recv err, fd is: %d", fd);close(fd);_event_fds[pos].fd = defultfd; // 本质是从select中移除了}}void HandlerEvent(){for (int i = 0; i < fd_num_max; i++){int fd = _event_fds[i].fd;if (fd == defultfd)continue;if (_event_fds[i].revents & POLLIN) // 读事件就绪{if (fd == _listensock.Fd()) // 监听套接字{Accepter(); // 连接管理器}else // 非监听套接字，要开始读取啦{Recver(fd, i);}}}}void Start(){_event_fds[0].fd = _listensock.Fd();_event_fds[0].events = POLLIN;int timeout = 3000; // 每隔3swhile (true){int n = poll(_event_fds, fd_num_max, timeout);switch (n){case 0:// 没有描述符就绪std::cout << "time out, timeout:" << timeout << std::endl;break;case -1:std::cerr << "poll err " << std::endl;break;default:// 有事件就绪了std::cout << "get a new link" << std::endl;HandlerEvent();break;}}}void PrintFd(){std::cout << "fd list: ";for (int i = 0; i < fd_num_max; i++){if (_event_fds[i].fd == defultfd)continue;std::cout << _event_fds[i].fd << " ";}std::cout << std::endl;}~PollServer(){_listensock.Close();}
};

I/O多路转接之epoll高级版改进select和poll的问题

快速了解一下epoll接口：epoll_create epoll_wait epoll_ctl

epoll_ctl

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

epoll的事件注册函数.
它不同于select()是在监听事件时告诉内核要监听什么类型的事件, 而是在这里先注册要监听的事件类型.
第一个参数是epoll_create()的返回值(epoll的句柄).
第二个参数表示动作，用三个宏来表示.
第三个参数是需要监听的fd.
第四个参数是告诉内核需要监听什么事.
第二个参数的取值:
EPOLL_CTL_ADD ：注册新的fd到epfd中；
EPOLL_CTL_MOD ：修改已经注册的fd的监听事件；
EPOLL_CTL_DEL ：从epfd中删除一个fd；
struct epoll_event结构如下:

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

epoll_wait

int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);

收集在epoll监控的事件中已经发送的事件.
参数events是分配好的epoll_event结构体数组.
epoll将会把发生的事件赋值到events数组中 (events不可以是空指针，内核只负责把数据复制到这个events数组中，不会去帮助我们在用户态中分配内存).
maxevents告之内核这个events有多大，这个 maxevents的值不能大于创建epoll_create()时的size.
参数timeout是超时时间 (毫秒，0会立即返回，-1是永久阻塞).
如果函数调用成功，返回对应I/O上已准备好的文件描述符数目，如返回0表示已超时, 返回小于0表示函数失败.

epoll原理

epoll优势：
1.检测就绪O(1)，获取就绪O(n)
2.fd和event没有上限
3.返回值n表示有几个fd就绪啦就绪事件是连续的，有返回值个。

当某一进程调用epoll_create方法时，Linux内核会创建一个eventpoll结构体，这个结构体中有两个成员与epoll的使用方式密切相关.
每一个epoll对象都有一个独立的eventpoll结构体，用于存放通过epoll_ctl方法向epoll对象中添加进来的事件.
这些事件都会挂载在红黑树中，如此，重复添加的事件就可以通过红黑树而高效的识别出来(红黑树的插入时间效率是lgn，其中n为树的高度).
而所有添加到epoll中的事件都会与设备(网卡)驱动程序建立回调关系，也就是说，当响应的事件发生时会调用这个回调方法.
这个回调方法在内核中叫ep_poll_callback,它会将发生的事件添加到rdlist双链表中.
在epoll中，对于每一个事件，都会建立一个 epitem结构体

当调用epoll_wait检查是否有事件发生时，只需要检查eventpoll对象中的rdlist双链表中是否有epitem元素即可.
如果rdlist不为空，则把发生的事件复制到用户态，同时将事件数量返回给用户. 这个操作的时间复杂度是O(1).

总结一下, epoll的使用过程就是三部曲:
调用epoll_create创建一个epoll句柄;
调用epoll_ctl, 将要监控的文件描述符进行注册;
调用epoll_wait, 等待文件描述符就绪;

epoll的优点:
接口使用方便: 虽然拆分成了三个函数, 但是反而使用起来更方便高效. 不需要每次循环都设置关注的文件描述符, 也做到了输入输出参数分离开
数据拷贝轻量: 只在合适的时候调用 EPOLL_CTL_ADD 将文件描述符结构拷贝到内核中, 这个操作并不频繁(而select/poll都是每次循环都要进行拷贝)
事件回调机制: 避免使用遍历, 而是使用回调函数的方式, 将就绪的文件描述符结构加入到就绪队列中, epoll_wait 返回直接访问就绪队列就知道哪些文件描述符就绪. 这个操作时间复杂度O(1). 即使文件描述符数目很多, 效率也不会受到影响.
没有数量限制: 文件描述符数目无上限

epoll代码

EpollServer.hpp:

#pragma once#include <iostream>
#include <sys/epoll.h>
#include "Socket.hpp"
#include "nocopy.hpp"
#include "Log.hpp"
#include "Epoller.hpp"uint32_t EVENT_IN = (EPOLLIN);
uint32_t EVENT_OUT = (EPOLLOUT);
static const uint16_t defaultport = 8080;class EpollServer : public nocopy
{static const int num = 64;public:EpollServer(uint16_t port = defaultport): _port(port), _listensock_ptr(new Sock()), _epoller_ptr(new Epoller()){}void Init(){_listensock_ptr->Socket();_listensock_ptr->Bind(_port);_listensock_ptr->Listen();lg(Info, "create listen socket successful...%d\n", _listensock_ptr->Fd());}void Acceptor(){// 获取一个新连接std::string clientip;uint16_t clientport;int sock = _listensock_ptr->Accept(&clientip, &clientport);if (sock > 0){// 把新连接放入到epoll内核中_epoller_ptr->EpollerUpdate(EPOLL_CTL_ADD, sock, EVENT_IN);lg(Debug, "get a new link...clientip:clientport#%s:%d", clientip.c_str(), clientport);}}void Recver(int fd){char buffer[1024];ssize_t n = read(fd, buffer, sizeof(buffer) - 1); if (n > 0){buffer[n] = 0;std::cout << "get a messge: " << buffer << std::endl;// wrirtestd::string echo_str = "server echo $ ";echo_str += buffer;write(fd, echo_str.c_str(), echo_str.size());}else if (n == 0){lg(Info, "client quit, me too, close fd is : %d", fd);_epoller_ptr->EpollerUpdate(EPOLL_CTL_DEL, fd, 0);close(fd);}else{lg(Warning, "recv error: fd is : %d", fd);_epoller_ptr->EpollerUpdate(EPOLL_CTL_DEL, fd, 0);close(fd);}}void HandlerEvent(struct epoll_event revt[], int num){for (int i = 0; i < num; i++){uint32_t events = revt[i].events;int fd = revt[i].data.fd;if (events & EVENT_IN){if (fd == _listensock_ptr->Fd()) // 是监听套接字，那么就需要接收{Acceptor();}else{// 其他描述符就绪Recver(fd);}}else if (events & EVENT_OUT){}else{}}}void Start(){// 将listensock添加到epoll中 -> listensock和他关心的事件，添加到内核epoll模型中rb_tree._epoller_ptr->EpollerUpdate(EPOLL_CTL_ADD, _listensock_ptr->Fd(), EVENT_IN);struct epoll_event revt[num];for (;;){int n = _epoller_ptr->EpollerWait(revt, num);if (n > 0){lg(Debug, "event happend, fd is: %d", revt[0].data.fd);HandlerEvent(revt, n);}else if (n == 0){lg(Error, "timeout... ");}else{lg(Error, "epoll wait error...");}}}~EpollServer(){_listensock_ptr->Close();}private:std::shared_ptr<Sock> _listensock_ptr;std::shared_ptr<Epoller> _epoller_ptr;uint16_t _port;
};

Epoller.hpp:

#pragma once#include "nocopy.hpp"
#include "Log.hpp"
#include <sys/epoll.h>
#include <cstring>
#include <cerrno>class Epoller : public nocopy
{static const int size = 128;public:Epoller(){_epfd = epoll_create(size);if (_epfd == -1){lg(Error, "epoll_create error, str: %s", strerror(errno));}else{lg(Info, "epoll_create sucessful,fd: %d", _epfd);}}int EpollerWait(struct epoll_event *revent, int num){int n = epoll_wait(_epfd, revent, num, /*_timeout*/-1);return n;}int EpollerUpdate(int oper, int sock, uint32_t event){int n = 0;if (oper == EPOLL_CTL_DEL){n = epoll_ctl(_epfd, oper, sock, nullptr);if (n != 0){lg(Error, "delete epoll_ctr error...");}}else{struct epoll_event epet;epet.events = event;epet.data.fd = sock; // ???n = epoll_ctl(_epfd, oper, sock, &epet);if (n != 0){lg(Error, "epoll_ctr error...");}}return n;}~Epoller(){if (_epfd >= 0){close(_epfd);}}private:int _epfd;int _timeout{3000};
};

nocopy.hpp:

#pragma onceclass nocopy
{
public:nocopy(){}nocopy(const nocopy &) = delete;const nocopy&operator=(const nocopy &) = delete;
};

main.cc:

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

epoll工作方式

epoll有2种工作方式-水平触发(LT)和边缘触发(ET)。
epoll默认LT模式。

有这样一个例子:
我们已经把一个tcp socket添加到epoll描述符
这个时候socket的另一端被写入了2KB的数据
调用epoll_wait，并且它会返回. 说明它已经准备好读取操作
然后调用read, 只读取了1KB的数据
继续调用epoll_wait…

水平触发Level Triggered 工作模式

事件到来，上层不处理，高电平一直有效。

epoll默认状态下就是LT工作模式.
当epoll检测到socket上事件就绪的时候, 可以不立刻进行处理. 或者只处理一部分.
如上面的例子, 由于只读了1K数据, 缓冲区中还剩1K数据, 在第二次调用 epoll_wait 时, epoll_wait仍然会立刻返回并通知socket读事件就绪.
直到缓冲区上所有的数据都被处理完, epoll_wait 才不会立刻返回.
支持阻塞读写和非阻塞读写

边缘触发Edge Triggered工作模式

数据或者连接从无到有从有到多，变化的时候才会通知我们一次。ET的通知效率更高，而且ET的IO效率也更高（因为TCP会向对方通告一个更大的窗口，从而从概率上让对方一次发送更多的数据）。为什么ET模式下fd必须是非阻塞的？ET情况下倒逼着程序员每次通知，都要把本轮数据全部取走。那么就需要循环读取直到读取出错。因为fd默认是阻塞的，所以在ET情况下，所有的fd必须都得是non_block的。

如果我们在第1步将socket添加到epoll描述符的时候使用了EPOLLET标志, epoll进入ET工作模式.
当epoll检测到socket上事件就绪时, 必须立刻处理.
如上面的例子, 虽然只读了1K的数据, 缓冲区还剩1K的数据, 在第二次调用 epoll_wait 的时候, epoll_wait 不会再返回了.
也就是说, ET模式下, 文件描述符上的事件就绪后, 只有一次处理机会.
ET的性能比LT性能更高( epoll_wait 返回的次数少了很多). Nginx默认采用ET模式使用epoll.
只支持非阻塞的读写
select和poll其实也是工作在LT模式下. epoll既可以支持LT, 也可以支持ET.

对比LT和ET

LT是 epoll 的默认行为. 使用 ET 能够减少 epoll 触发的次数. 但是代价就是强逼着程序猿一次响应就绪过程中就把所有的数据都处理完.
相当于一个文件描述符就绪之后, 不会反复被提示就绪, 看起来就比 LT 更高效一些. 但是在 LT 情况下如果也能做到每次就绪的文件描述符都立刻处理, 不让这个就绪被重复提示的话, 其实性能也是一样的.
另一方面, ET 的代码复杂程度更高了.

理解ET模式和非阻塞文件描述符

使用 ET 模式的 epoll, 需要将文件描述设置为非阻塞. 这个不是接口上的要求, 而是 “工程实践” 上的要求.
假设这样的场景: 服务器接受到一个10k的请求, 会向客户端返回一个应答数据. 如果客户端收不到应答, 不会发送第二个10k请求.

如果服务端写的代码是阻塞式的read, 并且一次只 read 1k 数据的话(read不能保证一次就把所有的数据都读出来,参考 man 手册的说明, 可能被信号打断), 剩下的9k数据就会待在缓冲区中

此时由于 epoll 是ET模式, 并不会认为文件描述符读就绪. epoll_wait 就不会再次返回. 剩下的 9k 数据会一直在缓冲区中. 直到下一次客户端再给服务器写数据. epoll_wait 才能返回
但是问题来了.
服务器只读到1k个数据, 要10k读完才会给客户端返回响应数据.
客户端要读到服务器的响应 , 才会发送下一个请求
客户端发送了下一个请求, epoll_wait 才会返回, 才能去读缓冲区中剩余的数据

所以, 为了解决上述问题(阻塞read不一定能一下把完整的请求读完), 于是就可以使用非阻塞轮训的方式来读缓冲区,保证一定能把完整的请求都读出来.
而如果是LT没这个问题. 只要缓冲区中的数据没读完, 就能够让 epoll_wait 返回文件描述符读就绪.

Reactor代码链接（整合一下epoll代码逻辑）

reactor代码链接

半同步半异步模型。