网站建设以及推广提案书,英文 网站 源码,网络设计方案的组成部分,php 做资讯网站#x1f525;个人主页#x1f525;#xff1a;孤寂大仙V #x1f308;收录专栏#x1f308;#xff1a;计算机网络 #x1f339;往期回顾#x1f339;#xff1a;【计算机网络】非阻塞IO——epoll 编程与ET模式详解——(easy)高并发网络服务器设计 #x1f516;流水不… 个人主页孤寂大仙V 收录专栏计算机网络 往期回顾【计算机网络】非阻塞IO——epoll 编程与ET模式详解——(easy)高并发网络服务器设计 流水不争争的是滔滔不息 一、Reactor模式简介Reactor 模式整体过程Reactor 模式优点 二、Reactor 模式网络框架实现详解Reactor整体结构Reactor反应堆Epollerepoll封裝Listener监听器Channel每条 TCP 连接数据流转时序解读举个例子 代码实现Reactor.hppEpoller.hppConnection.hppListener.hpp 和 Channel.hppMain.cc 一、Reactor模式简介
Reactor 模式是网络编程中最为典型且最实用的一种事件处理模式广泛应用于高速网络服务器中比如 Nginx、Redis、Netty 等旨在实现事件驱动下的非阻塞 I/O从而用少量线程处理海量连接大幅提升并发能力。
Reactor 模式整体过程
Reactor 模式主要分为以下几个环节 事件源Event Source 由操作系统提供如socket、文件、时钟等。这些事件源发生时会准备好数据或者发生条件。 事件分离器Demultiplexer, Epoll/Select 等 负责监控一组事件源若其中有准备好进行I/O的数据时会将它们分离出去告知Reactor进行处理。 本实现中我们借助 epoll 进行事件检测epoll为我们提供高速且灵活的I/O事件通知。 事件分发器Reactor Reactor根据事件发生时哪个文件描述符准备好分发给事先为每个文件准备好的事件处理器Handlers进行处理。 本实现中Reactor为每一个连接绑定了对应的handler比如说 Connection 类由handler进行数据读写。 事件处理器Handlers 事件处理器负责对发生的事件进行具体的数据操作。比如说 读缓冲区的数据。 进行业务逻辑处理。 再写回到缓冲区中。 本实现中下面代码Handlers 就是 Connection 的子类如 Listener负责接受新连接、Channel负责读写数据等。
Reactor 模式的数据流 整体的数据流可以简述为 事件源 - epoll 等事件分离器 - Reactor - 事件处理器 - 读/写缓冲区 - 业务处理 - 发送出去
Reactor 模式优点
资源消耗少 单个Reactor中只需要少量线程即可处理数千甚至数万个并发连接。扩展性强 适用从中、小到大型网络服务。解耦 事件分离器、事件分发器、事件处理器之间高度解耦方便代码扩展和测试。非阻塞 依赖非阻塞I/O避免资源浪费和时间消耗。
有兴趣可以了解这篇文章——Reactor模式 英文原版——reactor模式
二、Reactor 模式网络框架实现详解
近几年来随着并发量增加Reactor 模式凭借非阻塞 I/O 事件通知成为实现高速网络服务时最优且最稳定的方案。这种模式主体上依赖 Linux 的 epoll 等内核机制进行事件分发做到用少量线程处理很多连接。 让我带你看看你实现出的这个网络框架是怎样工作的。
Reactor整体结构
Reactor反应堆
负责事件循环poll、分发和管理每条连接。
class Reactor {// 负责// 1. epoll_create// 2. epoll_wait 等待事件发生// 3. 依序分发给每条 Connection 进行处理// 4. 新连接时进行添加// 5. 断线时进行清理
}Epollerepoll封裝
对 epoll_create、epoll_ctl、epoll_wait 进行封裝为 Reactor 减少实现负担
class Epoller {// 负责// 1. epoll_create// 2. epoll_ctl ADD/MOD/DEL// 3. epoll_wait 等待事件发生
}Listener监听器
负责对listen socket进行准备绑定、listen、非阻塞设置有新连接时为每条连接封裝为 Connection再交给 Reactor。
class Listener : public Connection {// 负责// 1. 绑定、listen// 2. Accept 新连接// 3. 设置为非阻塞// 4. 交由 Reactor 管理
}Channel每条 TCP 连接
封裝每条 TCP 连接负责数据缓冲、读写和异常处理。
class Channel : public Connection {// 负责// 1. recv 读数据到缓冲区// 2. send 发送缓冲区的数据出去// 3. EPOLLIN 就读// 4. EPOLLOUT 就写
}数据流转时序解读举个例子
比如说 有一个Client连到Server。 Reactor 通过 epoll_wait 等到此时有一个 EPOLLIN 事件。 这时 Reactor 就会检查是不是 Listening socket。 - 若是则由 Listener 进行 Accept()。 - 若不是则由对应 Connection (比如 Channel) 进行 recv()。
Connection 读到数据后会进入自己的缓冲区。 再由你为每条 Connection 设置好handler进行业务处理。 处理完后若需要发送出去则设置为 EPOLLOUT由 Reactor 继续关注此 socket 的写事件。
Reactor 就是事件分发器它自己不干活比如说不存在自己去读数据、写数据而是根据事件去通知对应的数据源比如说 socket由它自己进行处理。 让我举个例子你可以把它想象为高速公路收费站
收费站Reactor)负责分流每一条道路上发生的小汽车。每一条道路socket)可以发生不同的事件有車到达需要付费。收费员handler)为每一条道路提供服务比如收钱、找零。
Reactor 典型工作过程:
epoll_wait 等待事件
Reactor 通过 epoll_wait 等待哪个 socket 有事件发生。可能发生的事件包括 有新连接到达listen socket 有现有连接有数据可以读client socket 有缓冲区可以写出去client socket 有异常发生比如对方断掉
分发事件 Reactor 检查哪个文件发生了哪个事件。 若是listen socket 有 EPOLLIN 事件 - 就由 Listener 执行 Accept() 获取到 新的 Connection。 - 新 Connection 就被加入到 Reactor 的关注表中。 若是现有 Connection 有 EPOLLIN 事件 - 就由对应 Connection 执行 recv() 读出数据到缓冲区。 - 再由 Connection 绑定的数据处理器handler进行业务逻辑处理。 若是需要发送出去的数据准备好 - 就可以为 Connection 设置 EPOLLOUT 由 Reactor 关注到时进行 send() 发送出去。
代码实现
Reactor.hpp
#pragma once
#include iostream
#include memory
#include unordered_map
#include Epoller.hpp
#include Connection.hpp
#include Log.hppusing namespace std;
using namespace LogModule;class Reactor // 基座 Reactor反应堆模式高并发网络服务器
{static const int size 128;
private:bool IsAddConnectionHelper(int sockfd) // 判断_connections中是否存在connection{auto iter _connections.find(sockfd);if (iter _connections.end())return false;elsereturn true;}bool IsAddConnectionExist(const shared_ptrConnection conn){return IsAddConnectionHelper(conn-GetSockFd());}bool IsAddConnectionExist(int sockfd){return IsAddConnectionHelper(sockfd);}bool IsConnectionEmpty() //判断_connections是否为空{return _connections.empty();}int LoopOnce(int timeout){int n _epoller_ptr-WaitEvents(_revs, size, timeout); // 获取就绪队列中的就绪事件return n;} void Dispatcher(int n){for (int i 0; i n; i){int sockfd _revs[i].data.fd; // 拿到就绪队列中就绪的文件描述符uint32_t events _revs[i].events; // 拿到就绪队列中的事件if (events EPOLLERR) // 就绪事件这里不管出什么错都进EPOLLIN|EPOLLOUT让具体的异常处理进行处理events | (EPOLLIN | EPOLLOUT);if (events EPOLLHUP)events | (EPOLLIN | EPOLLOUT);if (events EPOLLIN){// 读事件就绪if (IsAddConnectionExist(sockfd))_connections[sockfd]-Recver();//索引_connection中connection对象}if (events EPOLLOUT){// 写事件就绪if (IsAddConnectionExist(sockfd))_connections[sockfd]-Sender();}}}public:Reactor(): _epoller_ptr(make_uniqueEpoller()),_isrunning(false){memset(_revs, 0, sizeof(_revs)); // 初始化 _revs}void Loop(){if (IsConnectionEmpty())return;_isrunning true;int timeout-1;while (_isrunning){PrintConnection();int nLoopOnce(timeout); //获取就绪队列就绪事件个数Dispatcher(n); //派发读事件||写事件}_isrunning false;}// 把所有新连接添加到_connections并把事件写到内核void AddConnection(shared_ptrConnection conn){if (IsAddConnectionExist(conn)){LOG(LogLevel::WARNING) connection is exist conn-GetSockFd();return;}uint32_t events conn-GetEvent(); // 一个connection中的事件int sockfd conn-GetSockFd(); // 一个connection中的文件描述符_epoller_ptr-AddEvent(sockfd, events); // 写入内核conn-SetOwner(this); //设置回指指针把reactor对象设置到connection基类_connections[sockfd] conn; //把connection放到_connections中}//关心写事件关写事件的connection放进_connections中就又走那个派发逻辑了然后再去发数据(妙)void EnableReadWrite(int sockfd,bool enableread,bool enablewrit){if(!IsAddConnectionExist(sockfd)){LOG(LogLevel::WARNING) connection is not exist sockfd;return;}uint32_t new_event(EPOLLET | (enableread ? EPOLLIN : 0) | (enablewrit ? EPOLLOUT : 0));_connections[sockfd]-SetEvent(new_event);_epoller_ptr-ModEvent(sockfd,new_event);}//异常处理(关闭文件描述符)void DelConnection(int sockfd) {_epoller_ptr-DelEvent(sockfd); //移除关心事件_connections.erase(sockfd); //_connections移除自己close(sockfd); //关闭文件描述符LOG(LogLevel::DEBUG)client quit;}void PrintConnection(){std::cout 当前Reactor正在进行管理的fd List:;for(auto conn : _connections){std::cout conn.second-GetSockFd() ;}std::cout \r\n;}~Reactor(){}private:unique_ptrEpoller _epoller_ptr; // 构造epoll模型unordered_mapint, shared_ptrConnection _connections; // 管理每一个connectionbool _isrunning; // 判断是否启动struct epoll_event _revs[size]; // 就绪队列就绪事件
};_epoll_ptr是对epoll类型的封装,Reactor是负责统筹兼顾的事件分发器。获取就绪队列中的就绪事件然后根据事件的不同去进行派发。 整个reactor模式有connection管理一套fd及其相对应的事件。为了在reactor中进行管理让文件描述符和connection建立哈希映射是_connections。listen套接字或者普通套接字的关心事件都会写入connection所以在添加到_connection之前要拿到这个文件描述符关心的事件并写入内核然后放入hash表中。后面进行事件派发的时候是根据就绪队列中的文件描述符对_connections哈希表中的对应的写事件就绪或者读事件就绪的connection(listen或者channel)进行索引。然后根据不同的事件去走不同的逻辑。 若 sockfd 对应 Listening 则 Recv() 就是去 Accept() 新连接。 若 sockfd 对应 Client 则 Recv() 就是去读数据。 若需要 Sender() 则是对 Client 进行写缓冲的数据发送出去。
异常处理这里就是移除关心的事件然后在_connections中移除索引然后关闭文件描述符。 Epoller.hpp
#pragma once
#include iostream
#include sys/epoll.h
#include Log.hpp
#include Common.hppusing namespace std;
using namespace LogModule;class Epoller
{
public:Epoller():_epfd(-1){_epfdepoll_create(256); //构造epoll模型if(_epfd0) {LOG(LogLevel::FATAL)epoll_create error;exit(EPOLL_CREATE_ERR);}LOG(LogLevel::INFO)epoll_create success;}void AddEvent(int sockfd,uint32_t events) //添加关心的事件{struct epoll_event ev;ev.eventsevents;ev.data.fdsockfd;int nepoll_ctl(_epfd,EPOLL_CTL_ADD,sockfd,ev);if(n0){LOG(LogLevel::FATAL)epoll_ctl error;exit(EPOLL_CTL_ERR);}LOG(LogLevel::INFO)epoll_ctl success;}int WaitEvents(struct epoll_event revs[],int size,int timeout) //就绪队列中就绪事件个数{int nepoll_wait(_epfd,revs,size,timeout);if(n0){LOG(LogLevel::WARNING)epoll_wait error;return n;}else if(n0){LOG(LogLevel::WARNING)epoll_wait timeout;return n;}else{LOG(LogLevel::INFO)读事件就绪;return n;}}void ModEvent(int sockfd,uint32_t events){struct epoll_event ev;ev.data.fdsockfd;ev.eventsevents;int nepoll_ctl(_epfd,EPOLL_CTL_MOD,sockfd,ev);if(n0){LOG(LogLevel::FATAL)mov_event error;exit(EPOLL_CTL_ERR);}LOG(LogLevel::INFO)mov_event success;}void DelEvent(int sockfd){int nepoll_ctl(_epfd,EPOLL_CTL_DEL,sockfd,nullptr);if(n0){LOG(LogLevel::FATAL)del_event error;exit(EPOLL_CTL_ERR);}LOG(LogLevel::INFO)del_event success;}~Epoller(){}
private:int _epfd;
};根据epoll的函数进行封装。
Connection.hpp
#pragma once
#include iostream
#include string
#include InetAddr.hppusing namespace std;
using namespace LogModule;using handler_tfunctionstring(string);class Reactor;
class Connection;//封装fd保证每个fd一套缓冲区
//基类
class Connection
{
public:Connection(){}virtual void Recver()0;virtual void Sender()0;virtual void EXcepter()0; //异常处理virtual int GetSockFd() 0;void SetEvent(const uint32_t events)//获取事件{_eventsevents;}uint32_t GetEvent()//返回事件{return _events; }void SetOwner(Reactor* owner)//设置回指指针{_ownerowner;}Reactor* GetOwner()//返回回指指针{return _owner;}void RegisterHandler(handler_t handler){_handlerhandler;}~Connection(){}
private: Reactor *_owner; //回指指针uint32_t _events; //关系的事件
public:handler_t _handler;
};
作为基类纯虚方法给Listener和Channel。事件管理和回指指针为什么用回指指针后面Listener中聊。注册回调也在后面用到的时候聊。
Listener.hpp 和 Channel.hpp
Listener.hpp
#pragma once
#include iostream
#include memory.h
#include Connection.hpp
#include Common.hpp
#include Socket.hpp
#include Log.hpp
#include Channel.hppusing namespace std;
using namespace SocketModule;
using namespace LogModule;//获取新连接
class Listener : public Connection
{
public:Listener(int portdefaultport):_port(port),_listensock(make_uniqueTcpSocket()){_listensock-BuildTcpSocketMethod(_port); //注意这里已经获取了listen套接字SetEvent(EPOLLIN | EPOLLET); //关心事件设置到connection ET模式 SetNonBlock(_listensock-Fd()); //文件描述符设置为非阻塞(ET模式)}void Recver() override //收{InetAddr client;while(true)//ET模式循环读文件描述符{ //这个文件描述符是正常文件描述符int sockfd_listensock-Accept(client);if(sockfdACCEPTDONE)break;else if(sockfdACCEPTCONTINUE)continue;else if(sockfdACCEPTERROR)break;else{//成功获取连接shared_ptrConnection connmake_sharedChannel(sockfd,client);//构造channel对象conn-SetEvent(EPOLLIN | EPOLLET);if(_handler!nullptr)conn-RegisterHandler(_handler);GetOwner()-AddConnection(conn);//继承基类函数 拿到reactor指针进而把channel对象添加到connection}}}void Sender() override{}void EXcepter() override{}int GetSockFd() override{return _listensock-Fd();}~Listener(){}
private:unique_ptrSocket _listensock;int _port;
};Channel.hpp
#pragma once
#include iostream
#include string
#include sys/types.h
#include sys/socket.h
#include memory
#include functional
#include Common.hpp
#include Connection.hpp
#include Log.hpp
#include InetAddr.hppusing namespace std;
using namespace LogModule;#define SIZE 1024
// 普通文件描述符的IO
class Channel : public Connection
{
public:Channel(int sockfd, InetAddr client): _sockfd(sockfd), _client_addr(client){SetNonBlock(sockfd);}void Recver() override // 收客户端数据{char buffer[SIZE];while (true){buffer[0] 0;ssize_t n recv(_sockfd, buffer, sizeof(buffer) - 1, 0);if (n 0){buffer[n] 0;_inbuffer buffer;}else if (n 0) // 读到数据为0{EXcepter(); // 进入异常处理return;}else // 读取失败{if (errno EAGAIN || errno EWOULDBLOCK){break;}else if (errno EINTR){continue;}else{EXcepter();return;}}}LOG(LogLevel::DEBUG) Channel inbuffer- _inbuffer;if (!_inbuffer.empty()){_outbuffer _handler(_inbuffer); // 处理完的数据给outbuffer}if (!_outbuffer.empty()) // 发数据{Sender();}}void Sender() override{while (true){ssize_t n send(_sockfd, _outbuffer.c_str(), _outbuffer.size(), 0);if (n 0){_outbuffer.erase(0, n); //发送缓冲区收到多少数据就清空多少个if (_outbuffer.empty())break;}else if (n 0){break;}else{if (errno EAGAIN || errno EWOULDBLOCK){break;}else if (errno EINTR){continue;}else{EXcepter();return;}}}//注意什么时候可以设置关系写事件写事件默认就是就绪的只有当发送缓冲区数据满了的时候才设置关系写事件//发送缓冲区已经满了outbuffer的数据发不到发送缓冲区中了,所以判断outbuffer不为空 //这时候就要关系写事件什么时候发送缓冲区有空位了outbuffer再往发送缓冲区中写if (!_outbuffer.empty()){GetOwner()-EnableReadWrite(_sockfd,true,true);}else{GetOwner()-EnableReadWrite(_sockfd,true,false);}}void EXcepter() override //异常处理{GetOwner()-DelConnection(_sockfd);}int GetSockFd() override // 获取文件描述符{return _sockfd;}~Channel(){}private:int _sockfd; // 套接字string _inbuffer;string _outbuffer;InetAddr _client_addr;
};Listener继承connection构造的时候获取连接把设置关系事件到connection中然后因为是ET模式边缘触发要把文件描述符设置为非阻塞的。 继承connection要实现connection的虚函数Recver()是接收连接创建普通套接字。拿到普通套接字要进行数据的收发了这时候Channel就要出场了。设置关心事件到connection这个Channel对象也要放到_connections中吧不放到_connections中的话在Reactor中对文件描述符进行索引如果是派发读数据写数据那还咋读咋写所以基类connection中的回指指针就派上用场了通过Reactor的指针用类内的AddConnection方法把Channel对象也放进去。
Channel主要是进行数据的收发收数据放到_inbuffer中读到数据为零进入异常处理读取错误进行错误处理。注册回调把_inbuffer通过回调给另一个模块进行处理然后返回后放到_outbuffer(发数据)。这里的回调function这些放到基类connection中这样Listener和Channel都能继承到。Listener的if(_handler!nullptr) conn-RegisterHandler(_handler);让注册回调更加灵活。如果_outbuffer中不为空那么就发数据。发数据这里要注意什么时候关心写事件呢写事件默认就是就绪的只有当发送缓冲区数据满了的时候才设置关系写事件发送缓冲区已经满了outbuffer的数据发不到发送缓冲区中了,所以判断outbuffer不为空 这时候就要关系写事件什么时候发送缓冲区有空位了outbuffer再往发送缓冲区中写。所以判断_outbuffer不为空说明发送缓冲区已经满了要关心写事件。
这时候要让Reactor进行统一管理了说白了就是epoll中的epoll_ctl关系事件的方式改一下去关心写事件具体不多说了看代码吧。 Main.cc
#include iostream
#include string
#include Reactor.hpp
#include Listener.hpp
#include Common.hpp
#include Log.hpp
#include NetCal.hpp
#include Protocol.hppusing namespace std;
using namespace LogModule;static void Usage(std::string proc)
{std::cerr Usage: proc port std::endl;
}int main(int argc, char *argv[])
{if (argc ! 2){Usage(argv[0]);exit(USAGE_ERR);}uint16_t port stoi(argv[1]);Enable_Console_Log_Strategy(); // 启用控制台输出// 顶层业务模块shared_ptrCal cal make_sharedCal();// 构造协议对象shared_ptrProtocol protocol make_sharedProtocol([cal](Request req) - Response{ return cal-Execute(req); });// connection管理listen连接模块 间接创建channel(进行io)模块shared_ptrConnection conn make_sharedListener(port);conn-RegisterHandler([protocol](string inbuffer) - string{string response_str;while (true){string package;if (!protocol-Decode(inbuffer, package)) // 判断报文完不完整break;response_str protocol-Execute(package);}LOG(LogLevel::DEBUG) 结束匿名函数中...: response_str;return response_str;});// rector基座 事件派发unique_ptrReactor tsvr make_uniqueReactor();tsvr-AddConnection(conn);tsvr-Loop();return 0;
}应用层模块就不多说了已经做到完美解耦合了。顶层应用层业务模块然后协议对象去调用业务模块connection管理listen连接模块在listener中又创建了channel对象。这里进行回调(这个回调是用RegisterHandler在这里就体现出来了是拿Listener对象调用的然后进行回调)把channel中的_inbuffer的数据回调给协议模块去进行处理。
Reactor模式高并发网络服务器——源码
