腾讯云建站多少钱,网站建设 服务器,wordpress提交500错误,使用jquery的网站前导#xff1a;本文是 I/O 多路复用的升级和实践#xff0c;如果想实现一个类似的服务器的话#xff0c;需要事先学习 epoll 服务器的编写。
友情链接#xff1a; 高级 I/O【Linux】 I/O 多路复用【Linux/网络】#xff08;C实现 epoll、select 和 epoll 服务器#x…前导本文是 I/O 多路复用的升级和实践如果想实现一个类似的服务器的话需要事先学习 epoll 服务器的编写。
友情链接 高级 I/O【Linux】 I/O 多路复用【Linux/网络】C实现 epoll、select 和 epoll 服务器
1. 什么是 Reactor 模式
既然你开始了解 Reactor反应器 模式说明你知道在实现服务端时多线程只能处理少量的客户端请求一旦数量增多维护线程的成本会急剧上升导致服务端性能下降。而 Reactor 模型就是为解决这个问题而诞生的。
Reactor 模式是一种基于事件驱动的设计模式它是 I/O 多路复用在设计模式层面上的体现。Reactor 设计模式的“设计”体现在
将 I/O 事件的处理分为两个阶段 事件分发阶段Dispatcher和事件处理阶段Handler。这种分离可以将 I/O 事件的处理从阻塞 I/O 中解耦出来从而提高系统的并发能力和吞吐量。使用 I/O 多路复用技术 I/O 多路复用技术可以让一个线程或进程监听多个 I/O 事件从而提高系统的资源利用率。使用事件驱动模型 事件驱动模型可以让系统更加灵活和可扩展。 图片来源http://www.dre.vanderbilt.edu/~schmidt/PDF/reactor-siemens.pdf 理解这张图需要你了解 select 服务器的写法。 这就是一个 Reactor 模型以 select 为例它主要包含五个组件
Handle句柄用于标识不同的事件本质是文件描述符。Sychronous Event Demultiplexer同步事件分离器本质是系统调用用于等待事件的发生。对于 Linux 来说同步事件分离器指的就是 I/O 多路复用的接口比如 select、poll、epoll 等。Event Handler事件处理器由多个回调方法构成这些回调方法构成了与应用相关的对于某个事件的处理反馈由用户实现。Concrete Event Handler具体事件处理器事件处理器中各个回调方法的具体实现可以认为是 Event Handler 的函数体。Initiation Dispatcher初始分发器初始分发器就是 Reactor 模型它会通过同步事件分离器来等待事件的发生当对应事件就绪时就调用事件处理器最后调用对应的回调方法来处理这个事件。
2. Reactor 模型的演化
我们知道一个服务器在接收数据后要对这个数据进行解码反序列化处理有必要的话还要序列化然后将处理后的数据返回给客户端。这些操作可以抽象为一个个模块交给线程去做。 首先是单线程 Reator 模型Reactor 模型会利用给定的 selectionKeys 进行派发操作派发到给定的 Handler之后当有客户端连接上来的时候Acceptor 会调用接口 accept()之后将接收到的连接和之前派发的 Handler 进行组合并启动。 然后是接入线程池的 Reactor 模型此模型将读操作和写操作解耦了出来当底层有数据就绪时将原本 Handler 的操作交给线程队列的头部线程来进行极大地提到了整体的吞吐量和处理速度。 图片来源https://gee.cs.oswego.edu/dl/cpjslides/nio.pdf
最后是多 Reactor 模型此模型中有一个主 ReactorMain Reactor和多个从 ReactorSub Reactor。Main Reactor 负责处理客户端的连接请求而 Sub Reactor 则负责处理已经建立的连接的读写事件。这种模型的好处就是整体职责更加明确同时对于多核 CPU 的机器系统资源的利用更加高一些。
3. Reactor 模式的工作流程
Reactor 对象通过 I/O 多路复用接口监听客户端请求事件收到事件后通过 Dispatch 进行分发。如果是建立连接请求则 Acceptor 通过 accept() 处理连接请求然后创建一个 Handler 对象处理完成连接后的各种事件。如果不是连接请求说明是读事件就绪则由 Reactor 分发调用连接对应的 Handler 来处理。Handler 只负责相应事件不做具体的业务处理通过 read() 读取数据后会分发给后面的 Worker 线程池的某个线程处理业务。Worker 线程池如果有的话会分配独立线程完成真正的业务并将结果返回给 Handler。Handler 收到响应后通过 send() 分发将结果返回给客户端。
4. Reactor 服务器
下面是一个 epoll 服务器比较完善的写法它是 I/O 多路复用【Linux/网络】C实现 epoll、select 和 epoll 服务器 中 epoll 服务器的提升版。
4.1 Connection 类
承接上一节中的 epoll 服务器现在的问题是来自用户的数据可能会被 TCP 协议拆分成多个报文那么服务器怎么才能知道什么时候最后一个小报文被接收了呢要保证完整地读取客户端发送的数据服务器需要将这次读取到的数据保存起来对它们进行一定的处理报文可能会有报头以解决粘包问题最后将它们拼接起来再向上层应用程序交付。
问题是 Recver 中的缓冲区 buffer 是一个局部变量每次循环都会重置。而服务端可能会有成百上千个来自客户端建立连接后打开的文件描述符这无法保证为每个文件描述符都保存本轮循环读取的数据。
解决办法是为套接字文件描述符建立独立的接收和发送缓冲区因为套接字是基于连接的所以用一个名为 Connection 的类来保存所有和连接相关的属性例如文件描述符收发缓冲区以及对文件描述符的操作包括读、写和异常操作所以要设置三个回调函数以供后续在不同的分支调用最后还要设置一个回指指针它将会保存服务器对象的地址到后面会介绍它的用处。
class TcpServer;using func_t std::functionvoid(Connection*);// 保存所有和连接/读写相关的属性
class Connection
{
public:Connection(int sock -1): _sock(sock), _tsvr(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; // I/O 文件描述符func_t _recv_cb; // 读事件回调函数func_t _send_cb; // 写事件回调函数func_t _except_cb; // 异常事件回调函数std::string _in_buffer; // 接收缓冲区std::string _out_buffer; // 发送缓冲区TcpServer* _tsvr; // 服务器回指指针
};成员函数 SetCallBack 是用来设置回调函数的地址的。注意成员变量的访问权限设置为 public这么做是测试是就不用写 Get 和 Set 方法了。
4.2 TcpServer 服务器
服务器类已经实现很多遍了这里只是将名字从 EpollServer 换成 TcpServer其中许多逻辑是不变的。不同的是成员变量将 epoll 的文件描述符换成了 epoll 对象因为在服务器类中希望直接调用函数而不进行传参其实不改也可以只是换个地方传参所以对 Epoll 类的封装改进了一下。
Epoll 类
#pragma once#include iostream
using namespace std;#include sys/epoll.hclass Epoll
{const static int g_size 256;const static int g_timeout 5000;public:Epoll(int timeout g_timeout) : _timeout(g_timeout){}void Create(){_epfd epoll_create(g_size);if (_epfd 0)exit(5);}bool Del(int sock){int n epoll_ctl(_epfd, EPOLL_CTL_DEL, sock, nullptr);return n 0;}bool Ctrl(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;}bool Add(int sock, uint32_t events){struct epoll_event ev;ev.events events;ev.data.fd sock;int n epoll_ctl(_epfd, EPOLL_CTL_ADD, sock, ev);return n 0;}int Wait(struct epoll_event revs[], int num){return epoll_wait(_epfd, revs, num, _timeout);}void Close(){if (_epfd 0)close(_epfd);}~Epoll(){}public:int _epfd;int _timeout;
};这里将成员访问限制限定为 public这么做是方便直接访问否则要写一些 Get 或 Set 方法。
服务器类框架
在构造函数中除了创建 listensock 和创建 Epoll 对象等之外要注意不仅 listensock真正运行起来的服务器应该会存在大量的 socket每一个 sock 都要被封装为一个 Connection 对象。那么这些 Connection 对象这么多必须要管理它们先描述后组织。所以在服务器中使用哈希表组织这些 Connection 对象key 是 sockvalue 是 Connection 对象。
以后只要 sock 对应的文件描述符就绪通过哈希表就能找到对应的 Connection 对象这样就能立马调用它的三个回调方法对套接字进行处理。通过这个哈希表就将 Epoll 和应用程序解耦。Epoll 的 AddSockToEpoll 成员函数就是为维护哈希表而设计的。
class TcpServer
{const static int default_port 8080;const static int g_num 128;
public:TcpServer(int port default_port): _port(port), _nrevs(g_num){// 1. 获取 listensock_listensock Sock::Socket();Sock::Bind(_listensock, _port);Sock::Listen(_listensock);// 2. 创建 epoll 实例_poll.Create();// 3. 添加 listensock 到服务器中AddConnection(_listensock, std::bind(TcpServer::Accepter, this, std::placeholders::_1), nullptr, nullptr);// 4. 为保存就绪事件的数组申请空间_revs new struct epoll_event[_nrevs];}void AddConnection(int sock, func_t recv_cb, func_t send_cb, func_t except_cb){// 1. 设置 sock 为非阻塞Sock::SetNonBlock(sock);// 2. 构建 Connection 对象封装 sockConnection *conn new Connection(sock);conn-SetCallBack(recv_cb, send_cb, except_cb);conn-_tsvr this;// 3. 让 epoll 对象监视_poll.Add(sock, EPOLLIN | EPOLLET);// 4. 将 Connection 对象的地址插入到哈希表_connections.insert(std::make_pair(sock, conn));}void Accepter(Connection *conn){}void Recver(Connection *conn){}void Sender(Connection *conn){}void Excepter(Connection *conn){}void LoopOnce(){int n _poll.Wait(_revs, _nrevs);for (int i 0; i n; i){int sock _revs[i].data.fd;uint32_t revents _revs[i].events;// 根据事件类型调用不同的回调函数// ... }}void Dispather(callback_t cb) // 原先的名字是 Start(){_cb cb;while (true){LoopOnce();}}~TcpServer(){if (_listensock 0)close(_listensock);if (_poll._epfd 0)close(_poll._epfd);if (_revs)delete[] _revs;}private:uint16_t _port; // 端口号int _listensock; // 监听套接字文件描述符Epoll _poll; // epoll 实例struct epoll_event *_revs; // 保存从就绪队列中取出的就绪事件的文件描述符的数组int _nrevs; // 数组长度std::unordered_mapint, Connection * _connections; // 用哈希表保存连接
};其中SetNonBlock 在 Sock 类中的实现
static bool SetNonBlock(int sock)
{int fl fcntl(sock, F_GETFL);if (fl 0)return false;fcntl(sock, F_SETFL, fl | O_NONBLOCK);return true;
}注意
在构造函数的第三点添加 listensock 到服务器中实际上就是将它作为 key 值和创建的 Connection 对象的指针绑定插入到哈希表中。用 AddConnection 函数封装这个插入的过程。不过在插入之前首先要创建 Connection 对象然后初始化它的成员。必须设置要插入的文件描述符为非阻塞通过调用 SetNonBlock 实现其次作为一个 I/O 多路复用的服务器一般默认值打开对读事件的关心写入事件会按需打开。除此之外还必须设置服务器为 ET 模式。注意哈希表的插入语法以及绑定函数和参数生成函数对象的用法。为啥要这么干呢因为这个位置上的参数也是函数对象。Epoll 类型的对象取名不取为 epoll而是取做 poll。这是因为在有些项目中poll 这个名字更具有通用性可以通过继承来区分不同类型。Excepter 函数是用来处理异常的例如 recv 和 send 出现了错误不管是什么错误我们都丢到这个函数中统一处理这样另外两个函数就能更关注它们要做的事情上。 std::bind 函数可以将一个可调用对象如函数、函数指针或者函数对象与其参数一起进行绑定生成一个新的可调用对象。其实就是把它们打包在一起 std::bind(TcpServer::Accepter, this, std::placeholders::_1)这行代码的含义是创建一个新的函数对象这个对象绑定了成员函数 TcpServer::Accepter 和对象 this 指针。其中std::placeholders::_1是一个占位符表示这个新的函数对象的第一个参数。 当这个新的函数对象被调用时它会调用 TcpServer::Accepter 这个成员函数同时传入的参数会替换掉占位符std::placeholders::_1。 LoopOnce 函数
在 LoopOnce 函数中执行的是每一次循环要做的事对于 epoll 而言用户进程只需要用一个数组存放已经就绪的文件描述符。然后遍历它。数组的每个元素都是一个事件结构体通过它的成员判断事件的类型是读还是写异常稍后再说。
bool IsConnectionExists(int sock)
{auto iter _connections.find(sock);if (iter _connections.end())return false;elsereturn true;
}
void LoopOnce()
{int n _poll.Wait(_revs, _nrevs);for (int i 0; i n; i){int sock _revs[i].data.fd;uint32_t revents _revs[i].events;// 根据事件类型调用不同的回调函数// 先不管异常事件// 读事件就绪if (revents EPOLLIN){if (IsConnectionExists(sock) _connections[sock]-_recv_cb ! nullptr) // 当回调函数被设置才能调用它_connections[sock]-_recv_cb(_connections[sock]);}// 写事件就绪if (revents EPOLLOUT){if (IsConnectionExists(sock) _connections[sock]-_send_cb ! nullptr)_connections[sock]-_send_cb(_connections[sock]);}}
}注意只有当这个 sock 对应的 Connection 对象在哈希表中才能调用它里面的回调函数而且要保证在调用之前回调函数是被设置过的否则会空指针异常。
判断 Connection 对象在哈希表中通过函数 IsConnectionExists 实现。
Accepter 回调函数
Accepter 函数要做的时和之前一样但是和之前实现的 epoll 服务器不同的是由于服务端要处理不止一个连接所以要在一个死循环中执行逻辑直到文件描述符获取失败。但这次失败不会影响到下次因为设置的 sock 是非阻塞的。
Accept() 函数的返回值小于零并不代表出现了很严重的错误对于本轮循环如果没有连接请求了就说明底层就绪的连接被取完了退出。EAGAIN 和 EWOULDBLOCK 是两个含义相同的错误码因为历史版本而形式不同它们表示数据没有就绪。
这通过一个输出型参数 accept_errno 来获取底层调用 accept() 的错误码因此 Sock::Accept() 也有做相应修改。
当 Accept 成功说明这个文件描述符和服务端建立了连接所以要为这个连接创建一个 Connection 对象以保存它的信息不要忘了注册回调函数。这样就相当于在派发时各自调用了回调函数。
void Accepter(Connection *conn)
{// 服务端需要处理不止一个连接while (true){std::string client_ip;uint16_t client_port;// 输出型参数获取错误码以便判断读取情况int accept_errno 0;int sock Sock::Accept(conn-_sock, client_ip, client_port, accept_errno);// 异常if (sock 0){// 没有新的连接请求取完了if (accept_errno EAGAIN || accept_errno EWOULDBLOCK)break;// 被信号中断else if (accept_errno EINTR)continue;// 失败else{logMessage(WARNING, accept error...code[%d] : %s, accept_errno, strerror(accept_errno));break;}}// 成功// 将 sock 交给 TcpServer 监视并注册回调函数if (sock 0){AddConnection(sock, std::bind(TcpServer::Recver, 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 of TcpServer success, sock: %d,client_ip.c_str(), client_port, sock);}}
}还是那个问题虽然 Accept 函数成功执行了但即使建立了连接服务端也不知道客户端何时会发送数据但现在 sock 的性质从之前的监听套接字变成了一个普通的 I/O 套接字那么继续交给 TcpServer 中的 epoll 实例监视。
这个将文件描述符“交付”给内核的操作在 select 服务器的编写过程中我们通过将文件描述符添加到数组中那么在 epoll 服务器中我们只需要将文件描述符Connection 对象三个回调函数这样的键值对插入到哈希表中就实现了用户和内核之间的解耦和数组的作用是类似的只不过我们不用自己维护哈希表。
Recver 回调函数
由于文件描述符是非阻塞的而且 epoll 被设置为 ET 模式所以要不断地处理本次客户端发送的数据因此在死循环中执行逻辑。差错处理和上面类似不同的是当出现错误时或对端关闭连接时调用 Excepter 函数。
当 recv() 成功时将本轮循环读取的数据尾插到 sock 自己的接收缓冲区中。跳出循环则说明本次客户端发送的数据全部读取完毕测试打印一下接收的数据。
void Recver(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, code:[%d] : %s, errno, strerror(errno));conn-_except_cb(conn);err true;break;}}else if (n 0){logMessage(DEBUG, client disconnected, sock[%d] close..., conn-_sock);conn-_except_cb(conn);err true;break;}// 成功else{buffer[n] \0;// 将本轮读取的数据尾插到接收缓冲区conn-_in_buffer buffer;}}if (!err)logMessage(DEBUG, client[%d] %s, conn-_sock, conn-_in_buffer.c_str());
}再次强调 Recver 函数的作用是进行一次数据的接收因为 eopll 被设置为 ET 模式文件描述符被设置为非阻塞模式由于缓冲区大小可能没那么大所以要通过若干轮读取来获取客户端发送的数据。跳出循环可能是因为读取完毕了可能是真的出错了真正出错的情况下不应该打印所以用一个 bool 类型的标记来限制打印的条件。
简单测试一下 现在可以保证接收数据的逻辑基本没有问题但是有个小细节当客户端断开连接以后这个文件描述符并没有关闭这是因为没有在出错时进行差错处理而是交给 Excepter 函数去做只不过现在还没有实现它。
另外由于是 Telnet 工具所以缓冲区中拼接的内容每次都会有一个回车符暂时不用关心这个问题在代码中已经去除了最后的符号sizeof(buffer) - 1因为这只是一个测试的工具实际上客户端可能是其他软件。
在测试时只是以字节流测试也就是直接读取数组中的内容但实际上客户端发送的数据可能是一个加密的报文所以要进行协议订制这也是解决粘包问题的手段。
服务器不应该和业务强耦合所以应该用回调函数回跳到上层业务设置的逻辑。什么意思呢意思就是在服务器眼里数据仅仅是数据而不关心它是什么具体要对数据做何种事让那个运行服务器的主体去做在这里就是 main 函数。 实际上可以将客户端发来请求封装为一个任务对象然后放到任务队列中让线程池处理这样服务器就只用关心连接和收发数据本身而不关心业务。发送数据只需要将序列化后的数据放到缓冲区中让服务器帮忙转发。 using callback_t std::functionvoid(Connection *, std::string request);
class TcpServer
{// ...
private:// ...callback_t _cb; // 上层设置的业务处理回调函数
};定义一个函数对象叫 callback_t它的参数是 (Connection *, std::string request)这其中包含连接的 sock 以及三个回调函数还有客户端发送的请求这个请求是上层业务需要解析的。
定制协议和解决粘包问题
关于客户端请求的处理在 认识协议【网络基础】 有介绍现在把其中的 Protocol.hpp 中的 Request 类和 Response 类以及 NetCal.hpp 的 calculatorHelper() 的放到这。
Request 类和 Response 类这两个类封装了序列化的反序列化的逻辑用来接收客户端发送的请求Request处理后的数据如果有必要返回的话那就返回应答Response给客户端。calculatorHelper()实现了简单的加减乘除用这个简易的计算机代表上层应用程序的业务。
为了解决粘包问题我们设置一个特殊符号X作为每个完整报文的分隔符例如用户输入两个子请求构成一个请求
1 1X2 * 2X实际上对于业务真正有用的数据应该用 X来拆分它
1 1
和
2 * 2在 Recver 接收到客户端发送的一个完整的请求后调用 SpliteMessage 函数将请求中的X剔除用数组存放每一个有效子请求然后把每个子请求交给上层业务逻辑。
void Recver(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 (!err){logMessage(DEBUG, client[%d] %s, conn-_sock, conn-_in_buffer.c_str());std::vectorstd::string message;// 用容器保存拆分数据中的有效请求SpliteMessage(conn-_in_buffer, message);// 将每个有效请求交给上层业务逻辑for (auto msg : message)_cb(conn, msg);}
}下面是实现拆分报文以及请求和响应中的序列化和反序列化的逻辑
// Protocol.hpp
#pragma once#include iostream
#include cstring
#include string
#include vector#define SEP X
#define SEP_LEN strlen(SEP)
#define SPACE
#define SPACE_LEN strlen(SPACE)void SpliteMessage(std::string buffer, std::vectorstd::string *out)
{while (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);}
}
// 解决粘包问题
std::string Encode(std::string s)
{return s SEP;
}// 请求
class Request
{
public:// 序列化// 1 1 - 1 1// _x _ystd::string Serialize(){std::string str;str std::to_string(_x);str SPACE;str _op;str SPACE;str std::to_string(_y);return str;}// 反序列化// 1 1 - 1 1bool Deserialize(const std::string str){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;else_op 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:// 序列化std::string Serialize(){std::string str;str code:;str std::to_string(_code);str SPACE;str result:;str std::to_string(_result);return str;}// 反序列化bool Deserialize(const std::string str){std::size_t pos str.find(SPACE);if (pos std::string::npos)return false;_code atoi(str.substr(0, pos).c_str());_result atoi(str.substr(pos SPACE_LEN).c_str());return true;}public:Response() {}Response(int result, int code, int x, int y, char op): _result(result), _code(code), _x(x), _y(y), _op(op){}~Response() {}public:int _result; // 结果int _code; // 错误码int _x;int _y;char _op;
};其实就是对字符串剪切拼接的操作
对于客户端发送的数据业务逻辑应该构建 Request 对象然后对数据进行反序列化获取其中真正有效的数据对于服务端返回的响应业务逻辑应该构建 Response 对象然后对处理后的结果进行序列化包装一下每次处理后的结果。这是因为客户端发送的报文中可能不止一个有效数据也就是可能要进行多次处理得到多个结果。为了保证这些结果不被混淆即粘包问题要进行序列化。
返回的响应中可能包含多个结果所以也用X来区分。这部分应该是通信双方协商制定的。
注意对于 SpliteMessage 函数而言它的第一个参数是输入缓冲区它是一个输入输出型参数在函数体内剔除掉特殊字符X后也要对输入缓冲区中的数据做同样的操作。第二个参数是一个输出型参数这个数组用来保存取出的数据。 在服务端处理数据的粘包问题时为什么已经获取到缓冲区中真正有效的数据以后还要将缓冲区中的内容做修改被处理后的缓冲区中的内容似乎没有什么作用了。 虽然内容已经被解码器拷贝出来并交给上层处理了所以它们在缓冲区中就没有存在的意义但是如果不删除或者移动它们那么它们就会占用缓冲区的空间并且可能干扰下一个数据包的读取和解析。所以解码器会将它们清理掉以保持缓冲区的整洁和正确。
// main.cc
#include TcpServer.hpp
#include memorystatic Response calculatorHelper(const Request req)
{Response resp(0, 0, req._x, req._y, req._op);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 (req._y 0)resp._code 1;elseresp._result req._x / req._y;break;case %:if (req._y 0)resp._code 2;elseresp._result req._x % req._y;break;default:resp._code 3;break;}return resp;
}void calculator(Connection *conn, std::string request)
{logMessage(DEBUG, calculator() been called, get a request: %s, request.c_str());// 1. 构建请求Request req;// 2. 反序列化if(!req.Deserialize(request)) return;// 3. 构建应答Response resp;// 4. 业务处理resp calculatorHelper(req);// 5. 序列化std::string sendstr resp.Serialize();// 6. 解决粘包问题sendstr Encode(sendstr);// 7. 将处理后的应答交给服务器conn-_out_buffer sendstr;// 8. 让服务器的 epoll 对象设置对写事件的关心conn-_tsvr-EnableReadWrite(conn, true, true);
}
int main()
{std::unique_ptrTcpServer svr(new TcpServer());svr-Dispather(calculator);return 0;
}[重点] 现在的问题是如何让服务器将处理好的数据发送给客户端呢我们假设 Sender 已经写完了在服务器的构造函数中只关心新到来的连接的文件描述符的读事件而不关心写事件。当一个连接到来后按规则发送了请求但是服务器中的 Epoll 对象并没有打开对写事件的关心这就没办法返回响应给客户端了。
所以在要返回响应给客户端的前提是让服务器打开对写事件的关心但是问题又来了这里已经是服务器上层的业务逻辑了怎么才能回到服务器中设置呢
还记得 Connection 类中有一个 TcpServer 类型的回指指针吗它可以帮助上层业务回到服务器中设置服务器对写事件的关心。
EnableReadWrite 函数
这个函数用来设置服务器的 epoll 对象对连接读写事件的关心与否。它接受三个参数一个是 Connection 类型的指针表示要设置的连接对象两个是 bool 类型的值表示是否允许读取或写入数据。
函数的主要逻辑是根据这两个布尔值来确定要设置的事件类型然后调用_poll.Ctrl() 方法来修改连接的事件监听状态
void EnableReadWrite(Connection *conn, bool readable, bool writeable)
{uint32_t events ((readable ? EPOLLIN : 0) | (writeable ? EPOLLOUT : 0));bool res _poll.Ctrl(conn-_sock, events);if (!res){logMessage(ERROR, EnableReadWrite() error...code:[%d]:%s, errno, strerror(errno));return;}
}除此之外EnableReadWrite 函数将会在 Sender 函数中也发挥作用。
Sender 函数
在 Sender 函数中应该将上层业务处理好后的数据发送给客户端。同样地send() 函数可能没办法一次性将发送缓冲区的数据发送给客户端所以要在一个死循环中执行。
在本轮循环发送的数据必须从发送缓冲区中移除。在差错处理时通过发送缓冲区为空与否来判断数据是否全部发送给客户端。
void Sender(Connection *conn)
{while (true){ssize_t n send(conn-_sock, conn-_out_buffer.c_str(), conn-_out_buffer.size(), 0);if (n 0){// 发送的数据应该在缓冲区中移除conn-_out_buffer.erase(0, n);// 发送完毕缓冲区为空退出发送逻辑if (conn-_out_buffer.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;}}}// 执行到这里并不能保证数据发送完毕// a. 发送完毕缓冲区为空取消服务器 epoll 模型对连接写事件的关心if (conn-_out_buffer.empty())EnableReadWrite(conn, true, false);// b. 缓冲区未空继续发送保持服务器 epoll 模型对连接写事件的关心elseEnableReadWrite(conn, true, true);
}[重要] 当跳出循环时还不能保证全部数据发送完毕这是因为出现错误时也会跳出循环但是缓冲区中依然有数据说明这次发送失败了。在 TCP 协议中对端主机在进行报文的序号校验时会发现这个错误失败信息返回给服务端进而发送给上层上层会重新调用 Sender 函数来发送。不过这里并未实现重发的逻辑。
当跳出循环时可能有以下几种情况
缓冲区为空表示用户空间的所有数据都已经拷贝到内核空间但是还不能保证内核空间的所有数据都已经发送到对端。这时候需要取消服务器 epoll 模型对连接写事件的关心避免频繁触发写事件浪费 CPU 资源。同时需要依赖 TCP 协议的可靠性机制确保数据最终能够到达对端。缓冲区不为空表示用户空间还有部分数据没有拷贝到内核空间可能是因为内核空间的发送缓冲区已满或者遇到其他错误。这时候需要继续保持服务器 epoll 模型对连接写事件的关心等待下一次可写时再次尝试发送数据。 为什么在 epoll 实现的服务器中服务器在发送数据之后要关闭 epoll 对写事件的关心呢 是因为 EPOLLOUT 事件是一个高频率触发的事件也就是说在大多数情况下文件描述符都是可写的除非缓冲区满了或者出现异常。如果不关闭 epoll 对写事件的关心那么每次 epoll_wait 返回时都会返回大量的 EPOLLOUT 事件占用了服务器的 CPU 资源并且可能干扰其他更重要的事件的处理。因此在发送数据之后如果数据已经发送完毕就应该关闭 epoll 对写事件的关心只保留对读事件或者其他事件的关心。这样可以提高服务器的性能和效率。
Excepter 函数
Excepter 函数是当内核检测到异常事件就绪时触发的回调函数。它的作用是在 Recver 函数或 Sender 函数中当 recv() 或 send() 系统调用出现致命性错误时进行移除不需要的文件描述符以及资源回收等操作。
致命性错误指的是系统调用真的出错了导致这个文件描述符没有再维护和监视的意义例如在使用 revc() 来接收数据时对端关闭了连接那么服务端也就没有必要再监视这个文件描述符了。
在此想强调的是要熟悉这些系统调用返回值和错误码的意义返回值小于零并不代表它真的出错了还要进一步通过错误码来判断真实的错误。 例如我在调试时我没有在初始化列表中为 timeout 参数初始化所以当 LoopOnce() 调用 EpollWait() 函数时其中的 epoll_wait 函数执行失败了但刚好这里面没有进行差错处理为此花了不少时间。 我的收获是不仅要根据返回值还要根据错误码定位错误原因。在 Recver 函数和 Sender 函数中也是这么做的。 void Excepter(Connection *conn)
{ // 0. 判断连接是否存在if (!IsConnectionExists(conn-_sock))return;// 1. 从 epoll 模型中删除bool res _poll.Del(conn-_sock);if (!res){logMessage(ERROR, DelFromEpoll() error...code:[%d]:%s, errno, strerror(errno));}// 2. 从服务器的哈希表中删除_connections.erase(conn-_sock);// 3. 关闭文件描述符close(conn-_sock);// 4. 释放空间delete conn;logMessage(DEBUG, Excepter() OK...);
}除了 Recver 函数和 Sender 函数LoopOnce 函数也需要差错处理如果事件的类型是 EPOLLERR文件描述符发生错误 或 EPOLLHUP对端将文件描述符关闭那么应该调用 Excepter 函数。但是 LoopOnce 函数中应该进行的是服务器一次循环应该做的事也就是根据就绪事件的类型来调用 Recver 函数还是 Sender 函数。
而事件的状态是通过事件掩码保存的那么将“出错”这个状态通过|运算设置进事件掩码中是不影响其他状态的因为出错后都要调用 Excepter 函数直接让 Sender 和 Recver 函数去做就好了。 void LoopOnce(){int n _poll.Wait(_revs, _nrevs);for (int i 0; i n; i){int sock _revs[i].data.fd;uint32_t revents _revs[i].events;// 根据事件类型调用不同的回调函数// 错误if (revents EPOLLERR)revents | (EPOLLIN | EPOLLOUT); // 将读写事件添加到就绪事件中// 对端关闭连接if (revents EPOLLHUP)revents | (EPOLLIN | EPOLLOUT);// 读事件就绪if (revents EPOLLIN){if (IsConnectionExists(sock) _connections[sock]-_recv_cb ! nullptr) // 当回调函数被设置才能调用它_connections[sock]-_recv_cb(_connections[sock]);}// 写事件就绪if (revents EPOLLOUT){if (IsConnectionExists(sock) _connections[sock]-_send_cb ! nullptr)_connections[sock]-_send_cb(_connections[sock]);}}}通过代码来看就是当事件出错时将它们的状态设置为可读可写然后进入后面两个分支这样就能通过调用 Recver 函数或 Sender 函数中的差错处理来调用 Excepter 函数。 为什么当文件描述符出错时将它们的状态设置为可读可写不主动调用 Excepter 函数来进行处理而是设置出错的文件描述符状态为可读可写然后进入 Recver 函数或 Sender 函数中的差错处理来调用 Excepter 函数呢 这样做的目的是能够及时发现和处理各种错误情况并且避免不必要的错误处理。什么意思呢就是说 Excepter 函数只处理致命的错误言外之意是文件描述符的状态无法真正表征错误的严重性。
和上面的系统调用出错时的返回值问题类似有些错误不能仅仅通过文件描述符的状态来判断因为一个错误状态可能对应多个问题。而且有些错误并不是立即可知的而是需要通过 recv 或 send 函数 read 或 write 函数来检测到。例如如果对方正常关闭了连接那么 read 函数会返回 0如果对方异常关闭了连接那么 recv 或 send 函数 read 或 write 函数会返回 -1并设置 errno 为 ECONNRESET 或 EPIPE。这些情况都需要通过 recv 或 send 函数 read 或 write 函数来发现并进行相应的处理。因此在事件出错时将它们的状态设置为可读可写可以让 Recver 或 Sender 函数在尝试读或写时发现错误并调用 Excepter 函数来处理。
同样地有些错误并不是致命的而是可以通过重试或忽略来解决的注意代码中的 break 和 continue 对应着不同的错误。如代码中写的如果 recv 或 send 函数 read 或 write 函数返回 -1并设置 errno 为 EAGAIN 或 EWOULDBLOCK那么表示缓冲区已满或者没有数据可读只需要等待下一次可读或可写时再次尝试即可如果 recv 或 send 函数 read 或 write 函数返回 -1并设置 errno 为 EINTR那么表示被信号中断了只需要继续尝试即可。这些情况都不需要调用 Excepter 函数来处理。因此在事件出错时将它们的状态设置为可读可写可以让 Recver 或 Sender 函数在遇到这些错误时进行重试或忽略。
4.3 测试 4.4 扩展
如果有恶意节点和服务端建立大量连接并且保持长时间不发送数据这种无意义的连接会占用服务端大量资源解决办法是设置一个超时时间。记录当前时间和客户端最近一次发送数据的时间当它们的差值超过设定的超时时间就断开连接。
class Connection
{// ...
public:// ...time_t _last_time; // 这个连接上次就绪的时间
};
class TcpServer
{// ...const static int link_timeout 10;public:void AddConnection(int sock, func_t recv_cb, func_t send_cb, func_t except_cb){// ...// 2. 构建 Connection 对象封装 sockconn-_last_time time(nullptr);// ...}void Recver(Connection *conn){// 记录这个连接最近发送数据的时间conn-_last_time time(nullptr);// ...}void ConnectAliveCheck(){for (auto connection : _connections){time_t current_time time(NULL);if (connection.second-_sock ! connection.first) break;if (current_time - connection.second-_last_time link_timeout)continue;else{if (connection.first ! _listensock connection.second ! nullptr (_connections.find(connection.first) ! _connections.end()))Excepter(connection.second);}}}void Dispather(callback_t cb){_cb cb;while (true){LoopOnce();ConnectAliveCheck();}}// ...
}思路是在检查函数 ConnectAliveCheck 中遍历哈希表中的所有连接超时且文件描述符不是监听套接字的话则调用 Excepter 函数。
值得注意的是当超时后调用 Excepter 函数删除超时连接在哈希表中的键值对而哈希表的 erase 函数并不是真正的删除而是将这个键值对设置为默认值以表示它的空闲状态这么做的目的是避免下次相同的键值要插入时重复计算哈希位置。
因此在 Excepter 函数中抹除了连接在哈希表中的值但这个位置仍然是占用内存的而且它在被再次设置之前是不能够访问它的。所以在枚举哈希表的元素时必须判断键值对中的 keysock和 valueConnection的 fd 是否相同因为理论上它们是相同的。如果不这么做会出现非法访问内存即段错误使得服务器程序崩溃。
除此之外由于哈希表在构造函数中将监听套接字文件描述符插入到了哈希表中这个文件描述符不应该被计时否则会直接断开连接。
当然这只是一个简单的计时逻辑它很粗糙精度很低但可以是一个很好的引入。实际上在多路转接主要是 epoll服务器中通常使用以下几种方案实现定时器
基于升序链表的定时器这种方案是将所有的定时器按照超时时间从小到大排序存放在一个链表中。每次调用 epoll_wait 时将链表头部的定时器的超时时间作为超时参数这样可以保证最先到期的定时器能够及时被处理。处理完一个定时器后将其从链表中删除并检查下一个定时器是否也已经超时如果是则继续处理直到没有超时的定时器为止。这种方案的优点是实现简单添加和删除定时器的时间复杂度都是 O (1)缺点是每次调用 epoll_wait 都需要遍历链表查找最近的超时时间时间复杂度是 O (n)。基于时间轮的定时器这种方案是将所有的定时器分配到一个环形数组中数组的每个元素对应一个时间槽每个时间槽可以存放多个定时器。数组有一个指针指向当前的时间槽每隔一段固定的时间称为槽间隔指针就向前移动一格并处理该槽中的所有定时器。这样可以保证定时器的精度不低于槽间隔并且不需要每次都遍历所有的定时器。这种方案的优点是添加和删除定时器的时间复杂度都是 O (1)并且可以支持长时间的定时任务缺点是需要额外的空间存储时间轮并且对于短时间的定时任务可能会有较大的误差。基于最小堆的定时器这种方案是将所有的定时器按照超时时间从小到大排序存放在一个最小堆中。最小堆的特点是堆顶元素根节点是最小的元素每次调用 epoll_wait 时将堆顶元素的超时时间作为超时参数这样可以保证最先到期的定时器能够及时被处理。处理完一个定时器后将其从堆中删除并重新调整堆结构使其满足最小堆性质。这种方案的优点是添加和删除定时器的时间复杂度都是 O (logn)并且可以支持任意精度的定时任务缺点是实现相对复杂并且需要额外的空间存储最小堆。 来自网络以后会填坑。 参考资料
【死磕 NIO】— Reactor 模式就一定意味着高性能吗Reactor pattern–wikireactor-siemens
源码Reactor
