【计网】实现reactor反应堆模型 --- 框架搭建

在这里插入图片描述

没有一颗星,
会因为追求梦想而受伤,
当你真心渴望某样东西时,
整个宇宙都会来帮忙。
--- 保罗・戈埃罗 《牧羊少年奇幻之旅》---

实现Reactor反应堆模型

  • 1 前言
  • 2 框架搭建
  • 3 准备工作
  • 4 Reactor类的设计
  • 5 Connection连接接口
  • 6 回调方法

1 前言

到目前为止,我学习了计算机网络,了解了网络传输的过程,理解网络协议栈的层与层之间的关系。实践了使用TCP进行的网络编程,也了解了协议的编写,实际了http协议下的通信过程。

最近学习了五种IO模型,可以通过多路转接EPOLL提高读取效率。

那么现在是否可以将多路转接与网络结合,编写一个高效处理网络请求的反应堆模型Reactor。今天我们搭建基础的结构。

2 框架搭建

在这里插入图片描述
我们想要搭建的是这样的结构:

  1. 最底层是Reactor:负责事件派发,管理connection套接字连接。可以添加监听套接字与普通套接字,其中都有对应的回调方法。可以通过套接字类型赋予连接对应的回调方法。通过多路转接IO获取就绪事件,找到对应connection执行事件。
  2. Connection连接:管理文件描述符的连接对象,内部有这个文件描述符的输入输出缓冲区,回调函数,客户端信息,就绪事件集。等待Reactor调用回调方法。
  3. Listener监听:这是专门管理监听套接字的对象,里面有对于监听套接字的方法,可以获取新连接。作为监听套接字connection的回调方法
  4. HandlerConnection普通套接字 :这是针对普通套接字的对象,里面有对于普通套接字事件就绪的处理方法类。

最底层的就是这三层结构。下面我们来实现这三层结构。

3 准备工作

在实现三层结构之前,我们先对多路转接IO进行封装,让代码尽可能解耦:

对于多路转接,我们设计一个基类,作为上层调用的统一接口。然后继承出子类Epoll poll select,在子类中分别实现对应的方法。

这里只提供了Epoll的封装:

  1. 构造函数:构造时创建EPOLL模型,获得EPOLLfd。
  2. AddEvent:添加事件,调用epoll_ctl_add方法即可。
  3. Wait:获取底层就绪事件,直接使用epoll_wait即可
#pragma once
#include <iostream>
#include <stdlib.h>
#include <sys/epoll.h>
#include "Log.hpp"
#include "Comm.hpp"using namespace log_ns;// 多路复用基类
class Mutliplex
{
public:Mutliplex(/* args */){}virtual bool AddEvent(int fd, uint32_t events) = 0;virtual int Wait(struct epoll_event revs[], int num, int timeout) = 0;~Mutliplex(){}
};// epoll poll select基类
class Epoller : public Mutliplex
{
private:static const int size = 128;public:Epoller(){_epollfd = ::epoll_create(size);if (_epollfd < 0){LOG(ERROR, "epoll create failed!\n");exit(EPOLL_CREATE);}}std::string EventToString(uint32_t revents){std::string ret;if (revents & EPOLLIN)ret += "EPOLLIN";if (revents & EPOLLOUT)ret += "| EPOLLOUT";return ret;}bool AddEvent(int fd, uint32_t events){struct epoll_event ev;ev.data.fd = fd;ev.events = events;int n = ::epoll_ctl(_epollfd, EPOLL_CTL_ADD, fd, &ev);if (n < 0){LOG(ERROR, "epoll_ctl add failed , errno:%d", errno);return -1;}LOG(INFO, "epoll_ctl add fd:%d , events:%s\n", fd, EventToString(events).c_str());return true;}int Wait(struct epoll_event revs[], int num, int timeout){return ::epoll_wait(_epollfd, revs, num, timeout);}~Epoller(){}private:int _epollfd;
};

4 Reactor类的设计

之前的TcpServer等服务端都要在内部封装_listensock。如果封装了监听套接字那么代码结构就定型了,就必须要有对监听套接字的处理。而这里我们想将Reactor设计一个管理connection连接的类,不需要针对监听套接字进行特殊处理

成员变量:

  1. 通过fd映射Connection*对象的哈希表_conn
  2. 判断是否启动bool isrunning
  3. 构建一个Multipex对象 , 构造时建立epoll指针,负责处理多路转接IO
  4. 就绪事件组struct epoll_event revs[gnum]
  5. 针对监听套接字的方法集,在添加连接时可以将方法设置进入connection
  6. 针对普通套接字的方法集

回调方法的类型为using handler_t = std::function<void(Connection *conn)>;

#pragma once
#include <string>
#include <iostream>
#include <memory>
#include <unordered_map>#include "Connection.hpp"
#include "Epoller.hpp"using namespace log_ns;class Reactor
{
private:static const int gnum = 128;public:Reactor() : _epoller(std::make_unique<Epoller>()), _isrunning(false){}void SetOnNormal(handler_t OnRecver, handler_t OnSender, handler_t OnExcepeter){_OnRecver = OnRecver;_OnSender = OnSender;_OnExcepeter = OnExcepeter;}void SetOnConnect(handler_t OnConnect){_OnConnect = OnConnect;}// 加入连接void AddConnection(int fd, uint32_t events, const InetAddr &addr, int type){}void Dispatcher(){}~Reactor(){}private:// fd 映射连接表std::unordered_map<int, Connection *> _conn;// 是否启动bool _isrunning;std::unique_ptr<Mutliplex> _epoller;// 事件数组struct epoll_event revs[gnum];//_listen新连接到来handler_t _OnConnect;// 处理普通fd IOhandler_t _OnRecver;handler_t _OnSender;handler_t _OnExcepeter;
};
  • Addconnection接口 :首先通过 fd events 与客户端信息和连接类型建立connection , 进行设置对应的事件集, 然后根据type判断类型,设置connection的上层处理回调方法。注意这里要对 connReactor进行关联 !后续connection的模块进行讲解,设置addr方便打印日志(可以知道是哪一个客户端);然后通过fd events 托管给epoll 进行添加事件 。最后将连接放入哈希表中。
  • IsConnExists判断当前连接是否存在
  • Dispatch()事件派发接口:进行while循环,获取底层哪些事件就绪 储存在成员变量struct epoll_event revs[gnum],根据返回值 n 对n个事件进行处理!这里只处理 ERR HUP IN OUT 使用if语句ERR HUP直接设置为IN OUT后续统一处理IN事件就绪 事件派发 通过_conn[fd]找到对应连接 执行对应事件的回调函数(注意保证连接存在 且 回调方法存在)。

完整代码如下:

#pragma once
#include <string>
#include <iostream>
#include <memory>
#include <unordered_map>#include "Connection.hpp"
#include "Epoller.hpp"using namespace log_ns;class Reactor
{
private:static const int gnum = 128;public:Reactor() : _epoller(std::make_unique<Epoller>()), _isrunning(false){}void SetOnNormal(handler_t OnRecver, handler_t OnSender, handler_t OnExcepeter){_OnRecver = OnRecver;_OnSender = OnSender;_OnExcepeter = OnExcepeter;}void SetOnConnect(handler_t OnConnect){_OnConnect = OnConnect;}// 加入连接void AddConnection(int fd, uint32_t events, const InetAddr &addr, int type){// 1. 通过 fd  构建一个 connection指针 set对应的事件集Connection *conn = new Connection(fd);conn->SetReactor(this);conn->SetEvents(events);conn->SetConnectionType(type);conn->SetAddr(addr);// 2. TODO 设置对connection的上层处理 设置回调方法if (conn->Type() == ListenConnection){conn->RegisterHandler(_OnConnect, nullptr, nullptr); // 设置方法}else{conn->RegisterHandler(_OnRecver, _OnSender, _OnExcepeter); // 设置方法}// 3. fd 与 events 托管给epoll 添加事件 出错直接 return;int n = _epoller->AddEvent(fd, events);// 4. 托管给_connection_conn.insert(std::make_pair(fd, conn));// 添加连接成功}// 判断连接是否存在bool IsConnExist(int fd){return _conn.find(fd) != _conn.end();}void LoopOnce(int timeout){// 获取底层事件int n = _epoller->Wait(revs, gnum, -1);for (int i = 0; i < n; i++){// 文件描述符int fd = revs[i].data.fd;// 就绪事件uint32_t revents = revs[i].events;// 处理IN OUT ERR HUPif (revents & EPOLLERR)revents |= (EPOLLIN | EPOLLOUT);if (revents & EPOLLHUP)revents |= (EPOLLIN | EPOLLOUT);if (revents & EPOLLIN){// 调用回调方法if (IsConnExist(fd) && _conn[fd]->_handler_recver)_conn[fd]->_handler_recver(_conn[fd]);}if (revents & EPOLLOUT){// 调用回调方法if (IsConnExist(fd) && _conn[fd]->_handler_sender)_conn[fd]->_handler_sender(_conn[fd]);}}}void Dispatcher(){_isrunning = true;int timeout = -1;while (true){LoopOnce(timeout);PrintDebug();//打印托管的fd列表}_isrunning = false;}void PrintDebug(){std::string s = "已建立的连接:";for (auto &conn : _conn){s += std::to_string(conn.first) + ' ';}LOG(DEBUG, "epoll 管理的fd列表: %s\n", s.c_str());}~Reactor(){}private:// fd 映射连接表std::unordered_map<int, Connection *> _conn;// 是否启动bool _isrunning;std::unique_ptr<Mutliplex> _epoller;// 事件数组struct epoll_event revs[gnum];//_listen新连接到来handler_t _OnConnect;// 处理普通fd IOhandler_t _OnRecver;handler_t _OnSender;handler_t _OnExcepeter;
};

5 Connection连接接口

  1. 成员变量
    • 文件描述符fd
    • 需要关心的事件集 events
    • 输入缓冲区 输出缓冲区
    • 三种事件的回调方法
    • 设置一个Reactor* _R
  2. SetEvents接口:通过传入events 初始化 events
  3. Events接口返回事件集
  4. Sockfd返回对应fd
  5. RegisterHandler接口快速设置回调方法
  6. SetReactor(Reactor* R)接口 connection与Reactor进行绑定,执行自己属于的Reactor

对于这个Reactor* _R 指针,是监听套装字获取到连接时发挥作用。当监听套接字的事件就绪,在回调方法中可以通过参数Connection取出内部的_R指针,找到对应的Reactor,进行AddConnection操作

#pragma once#include <iostream>
#include <string>
#include <functional>#include "InetAddr.hpp"class Connection;
class Reactor;using handler_t = std::function<void(Connection *conn)>;#define ListenConnection 0
#define NormalConnection 1class Connection
{
public:Connection(int fd) : _sockfd(fd){}void RegisterHandler(handler_t recver, handler_t sender, handler_t excepter){_handler_recver = recver;     // 处理读取_handler_sender = sender;     // 处理写入_handler_excepter = excepter; // 处理异常}void SetEvents(uint32_t events){_events = events;}void SetAddr(const InetAddr &addr){_addr = addr;}int Sockfd(){return _sockfd;}uint32_t Events(){return _events;}int Type(){return _type;}void SetReactor(Reactor *R){_R = R;}void SetConnectionType(int type){_type = type;}Reactor *GetReactor(){return _R;}InetAddr GerInetAddr(){return _addr;}void AppendInbuffer(const std::string &in){_inbuffer += in;}std::string &Inbuffer(){return _inbuffer;}~Connection(){}private:int _sockfd;            // 套接字fduint32_t _events;       // 事件集std::string _inbuffer;  // 输入缓冲区std::string _outbuffer; // 输出缓冲区Reactor *_R;int _type;InetAddr _addr;public:handler_t _handler_recver;   // 处理读取handler_t _handler_sender;   // 处理写入handler_t _handler_excepter; // 处理异常
};

6 回调方法

这里需要两种回调方法类,一种针对监听套接字,一种针对普通套接字。

Listener
  1. Listener统一管理Tcp连接模块,管理_listensock
  2. 成员变量 :
    • std::unique_ptr _listensock Tcp套接字对象
    • int _port;端口号
  3. 通过端口号进行构造TcpSocket
  4. ListenSock接口返回_listensock的fd。
  5. Accepter(conn* , int* code)方法获取连接并得到文件描述符 (这里采用ET模式)首先将listensockfd 读取设置为非阻塞读取,然后进行while(true)进行非阻塞读取 ,根据Accepter返回的错误码通过code返回 通过错误码进行判断,当读取到一个新的fd时,通过conn的Reactor指针调用AddConnection 加入新连接!
#pragma once
#include <memory>
#include <iostream>
#include "Socket.hpp"
#include "Connection.hpp"using namespace log_ns;
using namespace socket_ns;// 处理listen套接字的读写
class Listener
{
public:Listener(uint16_t port) : _port(port), _listensock(std::make_unique<TcpSocket>(port)){_listensock->BuildListenSocket(_port);}int ListenSockfd(){return _listensock->GetSockfd();}void Accepter(Connection *conn){LOG(DEBUG, "%d socket ready\n", conn->Sockfd());// 非阻塞式读取while (true){errno = 0;int code = 0;InetAddr addr;int sockfd = _listensock->Accepter(&addr, &code);if (sockfd > 0){LOG(INFO, "成功获取连接, 客户端:%s sockfd:%d\n", addr.AddrStr().c_str(), sockfd);conn->GetReactor()->AddConnection(sockfd, EPOLLIN | EPOLLET, addr, NormalConnection);}else{if (code == EWOULDBLOCK){// 读取完毕LOG(INFO, "底层数据全部读取完毕!\n");break;}// 信号中断else if (code == EINTR){continue;}else{LOG(ERROR, "获取连接失败!\n");break;}}}}~Listener(){}private:uint16_t _port;std::unique_ptr<Socket> _listensock;
};
HandlerConnection
  1. 处理普通连接读写问题,这个的设计就比较简单了,注意其只复杂数据的读取,协议解析需要交给上层进行处理!
  2. HandlerRecver(conn*):我们先实现读取的逻辑!
  3. HandlerSender(conn*):后续实现
  4. HandlerExcepter(conn*):后续实现
#include <sys/types.h>
#include <sys/socket.h>
// 不应该让HandlerConnection处理报文
class HandlerConnection
{
private:const static int buffersize = 512;public:HandlerConnection(handler_t process) : _process(process){}void Recver(Connection *conn){// LOG(DEBUG , "client发送信息: %d\n" , conn->Sockfd());// 进行正常读写 --- 非阻塞读取while (true){char buffer[buffersize];int n = ::recv(conn->Sockfd(), buffer, sizeof(buffer) - 1, 0);if (n > 0){// buffer是一个数据块 添加到conn的输入缓冲区中buffer[n] = 0;conn->AppendInbuffer(buffer);// 数据交给上层处理}else if (n == 0){// 连接断开LOG(INFO, "客户端[%s]退出, 服务器准备关闭fd: %d\n", conn->GerInetAddr().AddrStr().c_str(), conn->Sockfd());conn->_handler_excepter(conn); // 统一执行异常处理}else{// 本轮数据读完了if (errno == EWOULDBLOCK){// 这是唯一出口break;}// 信号中断else if (errno == EINTR){continue;}// 出现异常else{conn->_handler_excepter(conn);return;}}}// 读取完毕,我们应该处理数据了!// 加入协议std::cout << "Inbuffer 内容:" << conn->Inbuffer() << std::endl;_process(conn);}void Sender(Connection *conn){}void Excepter(Connection *conn){}~HandlerConnection(){}private:handler_t _process;
};

至此,Reactor反应堆模型的框架已经搭建好了,下一篇文章我们将在这个的基础之上进行协议解析与数据处理!并设计如何将数据发回。这里只是简单的实现读取数据的逻辑!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/diannao/59772.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

外包干了2年,快要废了。。。

先说一下自己的情况&#xff0c;普通本科毕业&#xff0c;在外包干了2年多的功能测试&#xff0c;这几年因为大环境不好&#xff0c;我整个人心惊胆战的&#xff0c;怕自己卷铺盖走人了&#xff0c;我感觉自己不能够在这样蹉跎下去了&#xff0c;长时间呆在一个舒适的环境真的会…

linux驱动-i2c子系统框架学习(2)

linux驱动-i2c子系统框架学习(1) 在这篇博客里面已经交代了i2c设备驱动层&#xff0c;主要的功能就是编写具体i2c的外设驱动&#xff0c;和创建设备接点给上层使用 &#xff0c;按之前学习的字符设备&#xff0c;有了设备节点&#xff0c;就可以对硬件操作了&#xff0c;在i2c…

Webserver(4.6)poll和epoll

目录 pollclient.cpoll.c epollepoll.cclient.c epoll的两种工作模式水平触发边沿触发 poll poll是对select的一个改进 select的缺点在于每次都需要将fd集合从用户态拷贝到内核态&#xff0c;开销很大。每次调用select都需要在内核遍历传递进来的所有fd&#xff0c;这个开销也…

Stable Diffusion的解读(一)

Stable Diffusion的解读&#xff08;一&#xff09; 文章目录 Stable Diffusion的解读&#xff08;一&#xff09;摘要Abstract一、机器学习部分1. Stable Diffusion的早期工作1.1 从编码器谈起1.2 第一条路线&#xff1a;VAE和DDPM1.3 第二条路线&#xff1a;VQVAE1.4 路线的交…

计算机网络——TCP篇

TCP篇 基本认知 TCP和UDP的区别? TCP 和 UDP 可以使用同一个端口吗&#xff1f; 可以的 传输层中 TCP 和 UDP在内核中是两个完全独立的软件模块。可以根据协议字段来选择不同的模块来处理。 TCP 连接建立 TCP 三次握手过程是怎样的&#xff1f; 一次握手:客户端发送带有 …

ROS话题通信机制理论模型的学习

话题通信是ROS&#xff08;Robot Operating System&#xff0c;机器人操作系统&#xff09;中使用频率最高的一种通信模式&#xff0c;其实现模型主要基于发布/订阅模式。 一、基本概念 话题通信模型中涉及三个主要角色&#xff1a; ROS Master&#xff08;管理者&#xff0…

【Android】名不符实的Window类

1.“名不符实”的Window类 Window 是一个窗口的概念&#xff0c;是所有视图的载体&#xff0c;不管是 Activity&#xff0c;Dialog&#xff0c;还是 Toast&#xff0c;他们的视图都是附加在 Window 上面的。例如在桌面显示一个悬浮窗&#xff0c;就需要用到 Window 来实现。Wi…

后端java——如何为你的网页设置一个验证码

目录 1、工具的准备 2.基本方法 3.实现类 4.实践 HTML文件&#xff1a; Java文件1:创建验证码 Java文件2:验证验证码 本文通过HUTOOL实现&#xff1a;Hutool参考文档Hutool&#xff0c;Java工具集https://hutool.cn/docs/#/ 1、工具的准备 如果我们通过hutool来实现这个…

【go从零单排】Strings and Runes 字符串和字符

Don’t worry , just coding! 内耗与overthinking只会削弱你的精力&#xff0c;虚度你的光阴&#xff0c;每天迈出一小步&#xff0c;回头时发现已经走了很远。 概念 在Go语言中&#xff0c;rune 是一个内置的数据类型&#xff0c;用于表示一个Unicode字符。它实际上是一个别名…

如何在本地Linux服务器搭建WordPress网站结合内网穿透随时随地可访问

文章目录 前言1. 安装WordPress2. 创建WordPress数据库3. 安装相对URL插件4. 安装内网穿透发布网站4.1 命令行方式&#xff1a;4.2. 配置wordpress公网地址 5. 配置WordPress固定公网地址 前言 本文主要介绍如何在Linux Ubuntu系统上使用WordPress搭建一个本地网站&#xff0c…

vue data变量之间相互赋值或进行数据联动

摘要&#xff1a; 使用vue时开发会用到data中是数据是相互驱动&#xff0c;经常会想到watch,computed&#xff0c;总结一下&#xff01; 直接赋值&#xff1a; 在 data 函数中定义的变量可以直接在方法中进行赋值。 export default {data() {return {a: 1,b: 2};},methods: {u…

在 Java 中使用脚本语言

在 Java 中使用脚本语言&#xff0c;特别是在 Java 平台上集成如 Python、JavaScript 或 Ruby 等语言&#xff0c;通常可以通过 Java 的 Scripting API 来实现。这个 API 基于 JSR 223&#xff08;“Scripting for the Java Platform”&#xff09;&#xff0c;提供了一种标准方…

大数据-212 数据挖掘 机器学习理论 - 无监督学习算法 KMeans 基本原理 簇内误差平方和

点一下关注吧&#xff01;&#xff01;&#xff01;非常感谢&#xff01;&#xff01;持续更新&#xff01;&#xff01;&#xff01; 目前已经更新到了&#xff1a; Hadoop&#xff08;已更完&#xff09;HDFS&#xff08;已更完&#xff09;MapReduce&#xff08;已更完&am…

微服务系列六:分布式事务与seata

目录 实验环境说明 前言 一、分布式事务问题与策略 1.1 分布式事务介绍 1.2 分布式事务解决策略分析 二、分布式事务解决方案 Seata 2.1 认识Seata 2.2 Seata的工作原理 2.3 部署Seata微服务 2.3.1 准备数据库表 2.3.2 准备配置文件 2.3.3 docker部署 2.4 微服务集…

Java 上机实践2(基础数据类型与数组)

&#xff08;大家好&#xff0c;今天分享的是Java的相关知识&#xff0c;大家可以在评论区进行互动答疑哦~加油&#xff01;&#x1f495;&#xff09; 目录 实验一&#xff1a;输出希腊字母表 一、实验目的 二、实验要求 三、程序代码 四、实验结果 实验二&#xff1a;…

w024基于SpringBoot的企业客户管理系统的设计与实现

&#x1f64a;作者简介&#xff1a;拥有多年开发工作经验&#xff0c;分享技术代码帮助学生学习&#xff0c;独立完成自己的项目或者毕业设计。 代码可以查看文章末尾⬇️联系方式获取&#xff0c;记得注明来意哦~&#x1f339;赠送计算机毕业设计600个选题excel文件&#xff0…

并发编程volatile精解

多线程下变量的不可见性 在多线程并发执行的情况下&#xff0c;多个线程修改共享的成员变量&#xff0c;会出现一个线程修改了共享变量的值后&#xff0c;另一个线程不能直接看到该线程修改后的变量最新值。(多线程下修改共享变量会出现变量修改值后的不可见性) 可见性问题…

十款外贸软件盘点,专注企业订单业务管理

在当今全球化的市场环境中&#xff0c;外贸企业的发展面临着诸多挑战与机遇。如何高效管理企业业务&#xff0c;提升运营效率&#xff0c;成为外贸企业在激烈竞争中脱颖而出的关键。外贸业务管理ERP软件作为一种强大的工具&#xff0c;能够整合企业资源、优化管理流程、实现数据…

yaml文件编写

Kubernetes 支持YAML和JSON格式管理资源 JSON 格式:主要用于 api 接口之间消息的传递 YAML 格式;用于配置和管理,YAML是一种简洁的非标记性语言,内容格式人性化容易读懂 一&#xff0c;yaml语法格式 1.1 基本语法规则 使用空格进行缩进&#xff08;不使用制表符&#xff0…

Node.js 全栈开发进阶篇

​&#x1f308;个人主页&#xff1a;前端青山 &#x1f525;系列专栏&#xff1a;node.js篇 &#x1f516;人终将被年少不可得之物困其一生 依旧青山,本期给大家带来node.js篇专栏内容:node.js- 全栈开发进阶篇 前言 大家好&#xff0c;我是青山。在上一篇文章中&#xff0c;…