Epoll ET服务器

设计思路

在epoll ET服务器中，我们需要处理如下几种事件：

• 读事件：如果是监听套接字的读事件就绪则调用accept函数获取底层的连接，如果是其他套接字的读事件就绪则调用recv函数读取客户端发来的数据。
• 写事件：写事件就绪则将待发送的数据写入到发送缓冲区当中。
• 异常事件：当某个套接字的异常事件就绪时我们不做过多处理，直接关闭该套接字。

当epoll ET服务器监测到某一事件就绪后，就会将该事件交给对应的服务处理程序进行处理。

Connection类

• ET工作模式下，只有数据从无到有，从有到多的过程epoll才会通知用户，就意味着如果一次性并没有将数据读取完毕，剩下的数据就相当于丢失了，所以我们并不是简单的定义一个缓冲区就可以，我们要保证我们的每一个文件描述符都对应一个自己的输入和输出缓冲区；
• 每一个文件描述符都对应一个自己的输入和输出缓冲区就保证了他们之间的就绪事件不会相互影响了，我们在循环读取过程中将数据保存在该文件描述符对应的inbuffer中，当inbuffer当中可以分离出一个完整的报文后再将其分离出来进行数据处理，这里的inbuffer本质就是用来解决粘包问题的。
• 我们将响应数据发送给客户端也是一样，在数据发送过程中，并不能保证TCP底层有足够的发送缓冲区供我们发送数据，我们可以将要发送的数据保存在一个outbuffer中，当底层TCP有足够的发送缓冲区供我们发送数据时，就依次发送outbuffer中的数据；
• 此之外我们还需要设置文件描述符以及对应的读回调、写回调和异常回调函数以及一个回指指针R。

class TcpServer;
class Connection;using func_t = std::function<void(Connection *)>;// 任意一个Sock都需要对应有自己的输入和输出缓冲区，保证数据没有被读取完成还能继续进行下一次读取
class Connection
{
public:Connection(int sock = -1) : _sock(sock), _svr(nullptr){}void SetCallback(func_t recv_cb, func_t send_cb, func_t except_cb){_recv_cb = recv_cb;_send_cb = send_cb;_except_cb = except_cb;}~Connection(){}public:int _sock; // 负责IO的文件描述符// 三个对应的回调方法func_t _recv_cb;func_t _send_cb;func_t _except_cb;// 接收缓冲区&&发送缓冲区std::string _inbuffer;std::string _outbuffer;// 设置对TcpServer回指指针TcpServer *_svr;
};

TcpServer类

我们的epoll ET服务器的工作流程如下：

• 首先我们需要进行的是监听套接字的创建，绑定与监听；
• 接着就需要我们创建一个多路转接对象了；
• 我们需要将我们的监听套接字添加到epoll模型当中，并且建立监听套接字与Connection之间的映射关系；
• 之后就可以不断调用TcpServer类中的事件分发函数进行事件派发。

在事件处理过程中，会不断向事件分发函数当中新增或删除事件，而每个事件就绪时都会自动调用其对应的回调函数进行处理，所以我们要做的就是不断调用事件分发函数函数进行事件派发即可。

在服务器工作的过程中，我们需要频繁的使用到epoll_ctl和epoll_wait函数，我们直接将其进行一下封装，然后设置为TcpServer类的成员变量即可。

Epoll.hpp

#pragma once
#include <iostream>
#include <sys/epoll.h>class Epoll
{static const int gnum = 128;static const int gtimeout = 5000;public:Epoll(int timeout = gtimeout) : _timeout(timeout){}void CreateEpoll(){_epfd = epoll_create(gnum);if (_epfd <= 0)exit(5);}bool AddSockToEpoll(int sock, uint32_t events){struct epoll_event ev;ev.data.fd = sock;ev.events = events;int n = epoll_ctl(_epfd, EPOLL_CTL_ADD, sock, &ev);return n == 0;}int WaitEpoll(struct epoll_event revs[], int num){return epoll_wait(_epfd, revs, num, _timeout);}~Epoll(){}private:int _epfd; // 指定epoll模型int _timeout;
};

由于每一个文件描述符都对应一个Connection，服务器中就会存在大量的Connection，所以我们可以创建一个哈希表来映射文件描述符与对应Connection之间的关系，也就是“先描述，在组织”，完成对Connection管理工作。

所以此时我们就可以搭建一个简单的模型出来，我们TcpServer初始化阶段就完成4个工作：

• 监听套接字创建；
• 多路转接对象创建；
• 添加listen套接字到epoll模型中；
• 构建一个就绪事件缓冲区；

#pragma once#include <iostream>
#include <functional>
#include <string>
#include <unordered_map>
#include "Sock.hpp"
#include "log.hpp"
#include "Epoll.hpp"class TcpServer;
class Connection;class TcpServer
{static const int gport = 8080;static const int gnum = 128;public:TcpServer(int revs_num = gnum, int port = gport) : _port(port), _revs_num(revs_num){// 1. 创建监听套接字_listensock = Sock::Socket();Sock::Bind(_listensock, _port);Sock::Listen(_listensock);// 2. 创建多路转接对象_poll.CreateEpoll();// 3. 添加listen套接字到服务器中AddConnection(_listensock, std::bind(&TcpServer::Accepter, this, std::placeholders::_1), nullptr, nullptr);// 4. 构建一个获取就绪事件的缓冲区_revs = new struct epoll_event[_revs_num];}// 就绪事件的派发void DisPather(){while (true){//}}~TcpServer(){if (_listensock >= 0)close(_listensock);if (_revs)delete[] _revs;}private:int _listensock;                                    // 监听套接字int _port;                                          // 端口号Epoll _poll;                                        // Epoll三剑客封装std::unordered_map<int, Connection *> _connections; // sock:Connection的映射表struct epoll_event *_revs;int _revs_num;
};

AddConnection函数

未来我们除了要添加_listensock以外，还需要需要添加大量的sock，所以我们的AddConnection函数就需要就需要将读回调，写回调，异常回调全部考虑在内，在进行_listensock添加时将写回调，异常回调设置为空即可；

• 我们再添加sock之间必须先将sock设置为非阻塞，在ET工作模式下，我们需要循环读取，在最后一次数据读取完毕以后，我们还需要进行下一次读取，判断数据是否读取完毕，所以sock必须是非阻塞的；
• 接下来就需要构建conn对象，对sock进行封装；
• 然后就是将sock添加到epoll中，这儿要注意的就是对于任何多路转接服务器，一般只会打开对默认读事件的关心，写入事件会按需进行打开；
• 最后将对应的Connection* 指针添加进_connections映射表中。

void AddConnection(int sock, func_t recv_cb, func_t send_cb, func_t except_cb)
{// 将sock设置为非阻塞Sock::SetNoneBlock(sock);// 未来我们除了要添加_listensock以外，还需要添加大量的sock，而且每一个sock都必须被封装成一个Connection// 服务器就会出现大量的Connection，操作系统就需要对这些Connection进行管理：先描述，在组织// 1. 构建conn对象，封装sockConnection *conn = new Connection(sock);conn->SetCallback(recv_cb, send_cb, except_cb);conn->_svr = this;// 2. 将sock添加到epoll中_poll.AddSockToEpoll(sock, EPOLLIN | EPOLLET); // 任何多路转接的服务器，一般默认只会打开对读取事件的关心，写入事件会按需进行打开// 3. 将对应的Connection* 指针添加进_connections映射表中_connections.insert(std::make_pair(sock, conn));
}

SetNoneBlock

static bool SetNoneBlock(int sock)
{int fl = fcntl(sock, F_GETFL);if (fl < 0)return false;fcntl(sock, F_SETFL, fl | O_NONBLOCK);return true;
}

DisPather函数

DisPather函数就是对我们的就绪事件进行派发，本质就是调用epoll_wait函数将就绪队列中的就绪事件拷贝进我们的就绪事件缓冲区中，然后就绪事件的类型，调用对应的回调函数进行就绪事件处理。

void LoopOnce()
{int n = _poll.WaitEpoll(_revs, _revs_num);for (int i = 0; i < n; i++){int sock = _revs[i].data.fd;uint32_t events = _revs[i].events;if (events | EPOLLIN){if (IsConnectionExists(sock) && _connections[sock]->_recv_cb != nullptr)_connections[sock]->_recv_cb(_connections[sock]);}if (events | EPOLLOUT){if (IsConnectionExists(sock) && _connections[sock]->_send_cb != nullptr)_connections[sock]->_send_cb(_connections[sock]);}}
}// 就绪事件的派发
void DisPather()
{while (true){LoopOnce();}
}bool IsConnectionExists(int sock)
{auto iter = _connections.find(sock);if (iter == _connections.end())return false;elsereturn true;
}

回调函数

Accepter函数

Accepter函数用于处理连接事件，工作流程如下：

• 调用accept函数在底层建立好连接；
• 建立连接完成以后，将客户端产生的sock添加到epoll模型中；
• 将该套接字及其对应需要关心的事件注册到Dispatcher当中。

void Accepter(Connection *conn)
{// logMessage(DEBUG, "Accepter been called");while (true){uint16_t client_port = 0;std::string client_ip;int accept_errno = 0;int sock = Sock::Accept(conn->_sock, &client_port, &client_ip, &accept_errno);if (sock < 0){if (accept_errno == EAGAIN || accept_errno == EWOULDBLOCK) // 并没有读取出错，只是底层没有连接了{break;}else if (accept_errno == EINTR) // 读取的过程被信号中断了{continue;}else // 获取连接失败{logMessage(WARNING, "accept error, %d : %s", accept_errno, strerror(accept_errno));break;}}if (sock > 0){// 将sock托管给TcpserverAddConnection(sock, std::bind(&TcpServer::Recever, this, std::placeholders::_1),std::bind(&TcpServer::Sender, this, std::placeholders::_1),std::bind(&TcpServer::Excepter, this, std::placeholders::_1));logMessage(DEBUG, "accept client %s:%d success, add to epoll&&TcpServer success, sock: %d",client_ip.c_str(), client_port, sock);}}
}

Recever函数

Recever函数用于处理读事件，其工作流程如下：

• 循环调用Recever函数，将读取到的数据添加到Connection结构inbuffer中去；
• 制定协议，保证在inbuffer中切割出来的是一个完整的报文；

报文切割

• 我们以“X”为为一个标志符，对报文之间进行分割，每个报文的最后都会以一个“X”作为报文结束的标志。
• 对inbuffer当中的字符串进行切割，将切割出来的一个个报文放到vector当中，对于最后无法切出完整报文的数据就留在inbuffer当中即可。

void SpliteMessage(std::string &buffer, std::vector<std::string> *out)
{// 100+// 100+19X1// 100+19X100+19while (true){auto pos = buffer.find(SEP);if (std::string::npos == pos)break;std::string message = buffer.substr(0, pos);buffer.erase(0, pos + SEP_LEN);out->push_back(message);}
}

序列化和反序列化

• 序列化是将对象的状态信息转换为可以存储或传输的形式（字节序列）的过程。
• 反序列化是把字节序列恢复为对象的过程。

在网络传输时，序列化目的是为了方便网络数据的发送和接收，无论是何种类型的数据，经过序列化后都变成了二进制序列，此时底层在进行网络数据传输时看到的统一都是二进制序列。

序列化后的二进制序列只有在网络传输时能够被底层识别，上层应用是无法识别序列化后的二进制序列的，因此需要将从网络中获取到的数据进行反序列化，将二进制序列的数据转换成应用层能够识别的数据格式。

如果我们传输的单纯是一个字符串，直接发送到网络中就可以了，但是如果是一些结构化的数据，比如实现一个计算器，他会存在左操作数，右操作数，操作符等，如果一个一个进行发送，就需要一个一个进行接收，此时服务端还需要纠结这些数据如何组合，所以我们可以将这些结构化数据打包。我们可以将协议进行一下封装：

Protocal.hpp

#pragma once#include <iostream>
#include <cstring>
#include <string>
#include <vector>// 1. 报文和报文之间，我们采用特殊字符来进行解决粘报问题
// 2. 获取一个一个独立完整的报文，序列和反序列化 -- 自定义
// 100+19X100+19X100+19std::string Encode(std::string &s)
{return s + SEP;
}class Request
{
public:std::string Serialize(){std::string str;str = std::to_string(x_);str += SPACE;str += op_; // TODOstr += SPACE;str += std::to_string(y_);return str;}bool Deserialized(const std::string &str) // 1 + 1{std::size_t left = str.find(SPACE);if (left == std::string::npos)return false;std::size_t right = str.rfind(SPACE);if (right == std::string::npos)return false;x_ = atoi(str.substr(0, left).c_str());y_ = atoi(str.substr(right + SPACE_LEN).c_str());if (left + SPACE_LEN > str.size())return false;elseop_ = str[left + SPACE_LEN];return true;}public:Request(){}Request(int x, int y, char op) : x_(x), y_(y), op_(op){}~Request() {}public:int x_;   // 是什么？int y_;   // 是什么？char op_; // '+' '-' '*' '/' '%'
};class Response
{
public:// "code_ result_"std::string Serialize(){std::string s;s = std::to_string(code_);s += SPACE;s += std::to_string(result_);return s;}// "111 100"bool Deserialized(const std::string &s){std::size_t pos = s.find(SPACE);if (pos == std::string::npos)return false;code_ = atoi(s.substr(0, pos).c_str());result_ = atoi(s.substr(pos + SPACE_LEN).c_str());return true;}public:Response(){}Response(int result, int code) : result_(result), code_(code){}~Response() {}public:// 约定！// result_ code_int result_; // 计算结果int code_;   // 计算结果的状态码
};

业务处理

业务处理就是服务器拿到客户端发来的数据后，对数据进行数据分析，最终拿到客户端想要的资源。

• 我们这里要做的业务处理非常简单，就是用反序列化后的数据进行数据计算，此时得到的计算结果就是客户端想要的。

void Recever(Connection *conn)
{const int num = 1024;bool err = false;while (true){char buffer[num];ssize_t n = recv(conn->_sock, buffer, sizeof(buffer) - 1, 0);if (n < 0){if (errno == EAGAIN || errno == EWOULDBLOCK){break;}else if (errno == EINTR){continue;}else{logMessage(ERROR, "recv error: %d, %s", errno, strerror(errno));conn->_except_cb(conn);err = true;break;}}else if (n == 0){logMessage(DEBUG, "client[%d] quit, me too!!!", conn->_sock);conn->_except_cb(conn);err = true;break;}else // 读取成功{buffer[n] = 0;conn->_inbuffer += buffer;}}logMessage(DEBUG, "conn->inbuffer[sock:%d]: %s", conn->_sock, conn->_inbuffer.c_str());// 数据读取完毕以后进行业务处理if (!err){std::vector<std::string> message;SpliteMessage(conn->_inbuffer, &message);for (auto &msg : message)_cb(conn, msg);}
}

注意：
在处理读事件时，会出现失败的情况，但是失败会存在多种情况，我们需要对应进行处理：

• 当错误码被设置为EAGAIN或EWOULDBLOCK，说明接受缓冲区中的数据已经被读取完了，此时就直接break；
• 当错误码设置为EINTR，说明读取过程被信号中断了，此时还需要继续调用recv函数进行发送；
• 当以上两种情况都不是时，说明就是读取异常，此时调用_except_cb函数进行异常处理。

Sender函数

sender回调用于处理写事件，其工作流程如下：

• 循环调用send函数发送数据，并将发送出去的数据从该套接字对应Connect结构的_outbuffer中删除；
• 如果循环调用send函数后该套接字对应的_outbuffer当中的数据被全部发送，此时就需要将该套接字对应的写事件关闭，因为已经没有要发送的数据了，如果_outbuffer当中的数据还有剩余，那么该套接字对应的写事件就应该继续打开。

    void Sender(Connection *conn){while(true){ssize_t n = send(conn->_sock, conn->_outbuffer.c_str(), conn->_outbuffer.size(), 0);if(n > 0){conn->_outbuffer.erase(0, n);if(conn->_outbuffer.empty()) break;}else{if(errno == EAGAIN || errno == EWOULDBLOCK){break;}else if(errno == EINTR){continue;}else{logMessage(ERROR, "send error: %d, %s", errno, strerror(errno));conn->_except_cb(conn);break;}}}// 此时并不知道数据是否发完，发完就不写入数据了，没发完就下次发送if(conn->_outbuffer.empty()) EnableWriteRead(conn, true, false);else EnableWriteRead(conn, true, true);}

EnableWriteRead函数

EnableReadWrite函数，用于使能或使能某个文件描述符的读写事件：

• 调用EnableReadWrite函数时需要传入一个文件描述符，表示需要设置的是哪个文件描述符对应的事件。
• 还需要传入两个bool值，分别表示需要使能还是使能读写事件。
• EnableReadWrite函数内部会调用epoll_ctl函数修改将该文件描述符的监听事件。

void EnableWriteRead(Connection* conn, bool readable, bool writeable)
{uint32_t events = (readable ? EPOLLIN : 0) | (writeable ? EPOLLOUT : 0);int res = _poll.CtlEpoll(conn->_sock, events);assert(res);
}

int CtlEpoll(int sock, uint32_t events)
{events |= EPOLLET;struct epoll_event ev;ev.events = events;ev.data.fd = sock;int n = epoll_ctl(_epfd, EPOLL_CTL_MOD, sock, &ev);return n == 0;
}

注意：
在处理写事件时，都会出现失败的情况，但是失败会存在多种情况，我们需要对应进行处理：

• 当错误码设置为EAGAIN或EWOULDBLOCK，说明发送缓冲区中数据已经被写满了，此时就需要将其中已经发送的数据删除在写入；
• 当错误码设置为EINTR，说明发送过程被信号中断了，此时还需要继续调用send函数进行发送；
• 当以上两种情况都不是时，就说明是真的发送出错了，此时我们就需要调用_except_cb函数进行相应异常操作。

Excepter函数

Excepter回调用于处理异常事件：

• 对于异常的事件，我们只需要将其对应的文件描述符关闭即可；
• 首先我们得将该文件描述符epoll模型中移除掉；
• 然后再将该文件描述符从我们建立映射关系的哈希表中移除；
• 最后在关闭文件描述符，delete我们对应的Connection对象。

void Excepter(Connection *conn)
{if (!IsConnectionExists(conn->_sock))return;// 将sock从epoll模型中删除_poll.DelFromEpoll(conn->_sock);// 将对应sock从映射表中移除_connections.erase(conn->_sock);// 关闭文件描述符sockclose(conn->_sock);// ddelete 对应Connection对象delete conn;
}

DelFromEpoll函数

bool DelFromEpoll(int sock)
{int n = epoll_ctl(_epfd, EPOLL_CTL_DEL, sock, nullptr);return n == 0;
}

事件处理

我们在创建一个TcpServer类对象以后，就需要调用Dispather函数进行事件的处理，此时我们设计一个简单是计算任务，封装成一个NetCal函数，他所需要进行的步骤如下：

• 将读取到的数据进行反序列化；
• 进行业务处理；
• 序列化数据，构建应答；
• 将数据交给服务器；
• 让底层的TcpServer开始发送数据；

#include "TcpServer.hpp"
#include "Protocol.hpp"
#include <memory>static Response calculator(const Request &req)
{Response resp(0, 0);switch (req.op_){case '+':resp.result_ = req.x_ + req.y_;break;case '-':resp.result_ = req.x_ - req.y_;break;case '*':resp.result_ = req.x_ * req.y_;break;case '/':if (0 == req.y_)resp.code_ = 1;elseresp.result_ = req.x_ / req.y_;break;case '%':if (0 == req.y_)resp.code_ = 2;elseresp.result_ = req.x_ % req.y_;break;default:resp.code_ = 3;break;}return resp;
}void NetCal(Connection *conn, std::string &request)
{logMessage(DEBUG, "NetCal been called, get request: %s", request.c_str());// 1.反序列化Request req;if (!req.Deserialized(request))return;// 2. 业务处理Response resp = calculator(req);// 3. 序列化，构建应答std::string sendstr = resp.Serialize();sendstr = Encode(sendstr);// 4. 交给服务器conn->_outbuffer += sendstr;// 5. 让底层的TcpServer开始发送数据conn->_svr->EnableWriteRead(conn, true, true);
}
int main()
{std::unique_ptr<TcpServer> tcp_server(new TcpServer());tcp_server->DisPather(NetCal);return 0;
}

注意：

我们的回指指针作用就是在NetCal函数第5步体现出来的，因为我们从始至终都没有调用过Sender函数，一直都是在读取数据，所以我们此时就需要想办法调用一次Sender函数，我们触发发送的动作，一旦我们开启EPOLLOUT，epoll会自动立马触发一次发送事件就绪，如果后续保持发送的开启，epoll会一直发送。

所以我们在此就需要通过我们的回指指针_svr调用EnableWriteRead函数，将读写操作都打开，此时底层就会调用我们的Sender函数，从此以后写操作就会一直执行下去。

套接字相关接口封装

我们将需要用到的套接字接口封装成一个Sock类并设置为静态成员函数，方便后续的调用：

Sock.hpp

#pragma once#include <iostream>
#include <string>
#include <cstring>
#include <cerrno>
#include <cassert>
#include <unistd.h>
#include <memory>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <ctype.h>
#include "log.hpp"class Sock
{
private:const static int gbacklog = 10;public:Sock(){}static int Socket(){int listensock = socket(AF_INET, SOCK_STREAM, 0);if (listensock < 0){logMessage(ERROR, "create socket error:%d:%s", errno, strerror(errno));exit(0);}logMessage(NORMAL, "create socket success, listensock:%d", listensock);int opt = 1;setsockopt(listensock, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt));return listensock;}static void Bind(int sock, uint16_t port, std::string ip = "0.0.0.0"){struct sockaddr_in local;memset(&local, 0, sizeof(local));local.sin_family = AF_INET;local.sin_port = htons(port);inet_pton(AF_INET, ip.c_str(), &local.sin_addr);if (bind(sock, (struct sockaddr *)&local, sizeof(local)) < 0){logMessage(ERROR, "bind error:%d:%s", errno, strerror(errno));exit(1);}}static void Listen(int sock){if (listen(sock, gbacklog) < 0){logMessage(ERROR, "listen error:%d:%s", errno, strerror(errno));exit(2);}logMessage(NORMAL, "init server success...");}static int Accept(int listensock, uint16_t *port, std::string *ip, int *accept_errno){struct sockaddr_in src;socklen_t len = sizeof(src);*accept_errno = 0;int servicesock = accept(listensock, (struct sockaddr *)&src, &len);if (servicesock < 0){logMessage(ERROR, "accept error:%d:%s", errno, strerror);*accept_errno = errno;return -1;}if (port)*port = htons(src.sin_port);if (ip)*ip = inet_ntoa(src.sin_addr);return servicesock;}static bool SetNoneBlock(int sock){int fl = fcntl(sock, F_GETFL);if (fl < 0)return false;fcntl(sock, F_SETFL, fl | O_NONBLOCK);return true;}~Sock(){}
};

同样我们也可以将我们的日志文件引入进来：

log.hpp

#pragma once#include <iostream>
#include <cstdio>
#include <cstdarg>
#include <ctime>
#include <string>// 日志是有日志级别的
#define DEBUG   0
#define NORMAL  1
#define WARNING 2
#define ERROR   3
#define FATAL   4const char *gLevelMap[] = {"DEBUG","NORMAL","WARNING","ERROR","FATAL"
};#define LOGFILE "./threadpool.log"// 完整的日志功能，至少: 日志等级 时间 支持用户自定义(日志内容, 文件行，文件名)
void logMessage(int level, const char *format, ...)
{char stdBuffer[1024]; //标准部分time_t timestamp = time(nullptr);snprintf(stdBuffer, sizeof stdBuffer, "[%s] [%ld] ", gLevelMap[level], timestamp);char logBuffer[1024]; //自定义部分va_list args;va_start(args, format);vsnprintf(logBuffer, sizeof logBuffer, format, args);va_end(args);printf("%s%s\n", stdBuffer, logBuffer);
}

运行Epoll服务器

行服务器后可以看到3号文件描述符被添加到了epoll模型中，这里的3号文件描述符其实就是监听套接字。

当客户端连接服务器后，在服务器端会显示5号文件描述符被添加到了epoll模型当中，因为4号文件描述符已经被epoll模型使用了。

此时客户端就可以向服务器发送一些简单计算任务，这些计算任务之间用“X”隔开，服务器收到计算请求并处理后就会将计算结果发送给客户端，这些计算结果之间也是用“X”隔开的。

此外，由于使用了多路转接技术，虽然当前的epoll服务器是一个单进程的服务器，但它却可以同时为多个客户端提供服务。

当客户端退出后服务器端也会将对应的文件描述符从epoll模型中删除。