网上做网站赚钱,劳务派遣好还是外包好,小网站,网站建设介绍ppt模板这个结局是我的期待#xff0c;我会一直为你祝福。 项目实现目标: 仿muduo库One Thread One Loop式主从Reacto模型实现高并发服务器。通过实现高并发服务器组件#xff0c;简洁快速完成搭建一个高性能服务器。并且#xff0c;通过组件内提供的不同应⽤层协议⽀… 这个结局是我的期待我会一直为你祝福。 项目实现目标: 仿muduo库One Thread One Loop式主从Reacto模型实现高并发服务器。通过实现高并发服务器组件简洁快速完成搭建一个高性能服务器。并且通过组件内提供的不同应⽤层协议⽀持也可以快速完成⼀个⾼性能应⽤服务器的搭建。本次项目支持HTTP应用层协议但仅限于消息收发而不包括任何业务处理内容。
前置背景:
认识HTTP服务器 概念: HTTP(HyperTextTransferProtocol)超⽂本传输协议是应⽤层协议是⼀种简单的使用最广泛的请求-响应协议客⼾端根据⾃⼰的需要向服务器发送请求服务器针对请求提供服务完毕后通信结束。 熟悉网络分层模型的友子们肯定知道HTTP协议是一个运行在TCP协议之上的上层应用协议因此所谓的HTTP服务器其本质就是TCP服务器。只不过在应⽤层 基于HTTP协议格式 进⾏ 数据的组织和解析 来明确客⼾端的请求并完成业务处理。 认识Reactor模型 概念: Reactor模式是指通过 ⼀个或多个输⼊ 同时传递给服务器进⾏请求处理时的事件驱动处理模式。 服务端程序处理传⼊ 多路请求 并将它们 同步分派给请求对应的处理线程 Reactor模式也叫Dispatcher模式。
Reactor模型分类
① 单Reactor单线程单I/O多路复⽤业务处理
● 通过IO多路复⽤模型进⾏客⼾端请求监控 ● 触发事件后进⾏事件处理: a. 如果是新建连接请求则获取新建连接并添加⾄多路复⽤模型进⾏事件监控。 b. 如果是数据通信请求则进⾏对应数据处理(接收数据处理数据发送响应)。 优点: 所有操作在同⼀线程中完成思想流程较为简单不涉及进程/线程间通信及资源争抢问题。 缺点: 适⽤于客⼾端数量较少且处理速度较为快速的场景。 (处理较慢或活跃连接较多会导致串⾏处理的情况下后处理的连接⻓时间⽆法得到响应) ② 单Reactor多线程单I/O多路复⽤线程池业务处理
● Reactor线程通过I/O多路复⽤模型进⾏客⼾端请求监控 ● 触发事件后进⾏事件处理: a. 如果是新建连接请求则获取新建连接并添加⾄多路复⽤模型进⾏事件监控。 b. 如果是数据通信请求则接收数据后分发给线程池里的业务线程进⾏业务处理。 c. ⼯作线程处理完毕后将响应交给Reactor线程进⾏数据响应。 优点: 充分利⽤CPU多核资源。 缺点: 多线程间的数据共享访问控制较为复杂。 单个Reactor承担所有事件的监听和响应在单线程中运⾏⾼并发场景下容易成为性能瓶颈。 ③ 多Reactor多线程多I/O多路复⽤线程池业务处理 基于单Reactor多线程的缺点考虑如果该Reactor进行IO时此时又来一个新连接事件又无法立即处理。因此将Reactor处理IO模块单独拎出来。让一个Reactor仅仅做事件派发而让其他Reactor进行IO事件的派发将数据分发给业务线程。 因此这种多Reactor多线程的模式又被称为主从Reactor模型。主Reactor负责新连接的监控而从Reactor负责对IO事件进行监控线程池里的线程则负责处理由从Reactor派发下来的数据。 优点: 充分利⽤CPU多核资源主从Reactor各司其职。 可是执行流不是越多越好反而会增加CPU的调度成本。 目标定位:One Thread One Loop主从Reactor模型: 主Reactor线程仅仅监控监听描述符获取新建连接保证获取新连接的⾼效性提⾼服务器的并发性能而⼦Reactor线程监控各⾃的描述符的读写事件进⾏数据读写以及业务处理。 OneThreadOneLoop的思想: 把所有的操作都放到⼀个线程中进⾏⼀个线程对应⼀个事件处理的循环。 功能模块: 基于以上的理解我们要实现的是⼀个带有协议⽀持的Reactor模型⾼性能服务器由此可以把项目分成两个大的模块
Server模块: 实现Reactor模型的TCP服务器
协议模块: 对当前的Reactor模型服务器提供应⽤层协议⽀持
SERVER 模块 server模块就是对所有的连接以及线程进⾏管理,让它们各司其职在合适的时候做合适的事最终完成⾼性能服务器组件的实现。 具体分为以下三个⽅⾯:
● 监听连接管理对监听连接进⾏管理.
● 通信连接管理对通信连接进⾏管理.
● 超时连接管理对超时连接进⾏管理. 为实现上面的管理思想将三个模块细致地划分为以下多个⼦模块:
Buffer模块: Buffer模块是⼀个缓冲区模块⽤于实现通信中⽤⼾态的接收缓冲区和发送缓冲区功能。
Socket模块: Socket模块是对套接字操作封装的⼀个模块主要实现的socket的各项操作。
Channel模块: Channel模块是对⼀个 “描述符” 需要进⾏的 ”IO事件管理的模块实现对描述符可读可写错误...事件的管理操作,以及当事件就绪时调用由上层设置的回调函数。 Acceptor模块: Acceptor模块是对Socket模块Channel模块的⼀个整体封装实现了对⼀个监听套接字的整体的管理。
● Acceptor模块内部包含有⼀个Socket对象实现监听套接字的操作.
● Acceptor模块内部包含有⼀个Channel对象实现监听套接字IO事件就绪的处理. Poller模块: Poller模块是对epoll进⾏封装的⼀个模块主要实现epoll的IO事件添加修改移除获取活跃连接功能。 Connection模块: Connection模块是对Buffer模块Socket模块Channel模块的⼀个整体封装。 实现了对⼀个通信套接字的整体的管理,每⼀个进⾏数据通信的套接字也就是accept获取到的新连接都会使⽤Connection进⾏管理。 TimerQueue模块: TimerQueue模块是实现固定时间定时任务的模块任务将在固定时间后被执⾏同时也可以通过刷新定时任务来延迟任务的执⾏。 这个模块主要是对Connection对象的⽣命周期管理简单来说一个连接是否是短连接还是长连接。如果是长连接一旦在一定的时间内没有发生任何数据通信该定时任务就会触发对该Connection对象进行资源清理、回收。 EventLoop模块 EventLoop模块可以理解就是我们上边所说的Reactor模块它是对Poller模块TimerQueue模块Socket模块的⼀个整体封装进⾏所有描述符的事件监控。 EventLoop模块必然是⼀个对象对应⼀个线程的模块线程内部的⽬的就是运⾏EventLoop的启动函数。 EventLoop模块为了保证整个服务器的线程安全问题因此要求Connection的所有操作⼀定要在其对应的EventLoop线程内完成不能在其他线程中进⾏(例如: Connection发送数据以及关闭连接这种操作。 TcpServer模块 这个模块是⼀个整体Tcp服务器模块的封装内部封装了Acceptor模块EventLoopThread模块。
● TcpServer中包含有⼀个 EventLoop对象以备在超轻量使⽤场景中不需要EventLoop线程池只需要在主线程中完成所有操作的情况。
● TcpServer模块内部包含有⼀个 EventLoopThreadPool对象其实就是EventLoop线程池也就是⼦Reactor线程池。
● TcpServer模块内部包含有⼀个Acceptor对象⼀个TcpServer服务器必然对应有⼀个监听套接 字能够完成获取客⼾端新连接并处理的任务。
● TcpServer模块内部包含有⼀个std::shared_ptrConnection的hash表,保存了所有的新建连接 对应的Connection。所有的Connection使⽤shared_ptr进⾏管理在shared_ptr计数器为0的情况下完成对Connection资源的释放操作。 协议模块:
Util模块 这个模块是⼀个⼯具模块主要提供HTTP协议模块所⽤到的⼀些⼯具函数⽐如url编解码⽂件读写等等。 HttpRequest模块 这个模块是HTTP请求数据模块⽤于保存HTTP请求数据被解析后的各项请求元素信息。 HttpResponse模块 这个模块是HTTP响应数据模块⽤于业务处理后设置并保存HTTP响应数据的的各项元素信息最终会被按照HTTP协议响应格式组织成为响应信息发送给客⼾端。 HttpContext模块 这个模块是⼀个HTTP请求接收的上下⽂模块主要是为了防⽌在⼀次接收的数据中不是⼀个完整的HTTP请求则解析过程并未完成⽆法进⾏完整的请求处理需要在下次接收到新数据后继续根据上下⽂进⾏解析。 HttpServer模块 这个模块是最终给组件使⽤者提供的HTTP服务器模块了⽤于以简单的接⼝实现HTTP服务器的搭建。
● HttpServer模块内部包含有⼀个TcpServer对象TcpServer对象实现服务器的搭建。
● HttpServer模块内部包含有两个提供给TcpServer对象的接⼝连接建⽴成功设置上下⽂接⼝数据处理接⼝。
● HttpServer模块内部包含有⼀个hash-map表存储请求与处理函数的映射表。使⽤者向HttpServer设置哪些请求应该使⽤哪些函数进⾏处理等TcpServer收到对应的请求就会使⽤对应的函数进⾏处理。 前置知识:
(1) C中的bind函数: bind (Fn fn, Args... args); 我们可以将bind接⼝看作是⼀个通⽤的函数适配器它接受⼀个函数对象以及函数的各项参数然后返回⼀个新的函数对象。 基于bind的作⽤当我们在设计⼀些线程池或者任务池的时候就可以将将任务池中的任务设置为函数类型函数的参数由添加任务者直接使⽤bind进⾏适配绑定设置⽽任务池中的任务被处理只需要取出⼀个个的函数进⾏执⾏即可。 (2) Linux中的定时器: 对于当前高并发服务器我们不得不考虑⼀个问题那就是连接的超时关闭问题。我们需要避免⼀个连接⻓时间不通信但是也不关闭空耗资源的情况。 这时候我需要借助Linux提供的定时器定时的将超时过期的连接进⾏释放。 int timerfd_create(int clockid, int flags); clockid: CLOCK_REALTIME - 系统实时时间如果修改了系统时间就会出问题 CLOCK_MONOTONIC - 从开机到现在的时间是⼀种相对时间; flags: TFD_NONBLOCK- 默认阻塞属性0 TFD_CLOEXEC - 关闭文件描述符继承 RETURN VAL: timerfd_create() returns a new file descriptor int timerfd_settime(int fd, int flags,const struct itimerspec *new_value, struct itimerspec *old_value); fd: timerfd_create返回的⽂件描述符. flags: 0-相对时间 1-绝对时间默认设置为0即可. new ⽤于设置定时器的新超时时间. old ⽤于接收原来的超时时间. 定时器会在每次超时时⾃动给fd中写⼊8字节的数据表⽰在上⼀次读取数据到当前读取数据期间超时了多少次。 上边例⼦是⼀个定时器的使⽤⽰例是每隔3s钟触发⼀次定时器超时否则就会阻塞在read读取数据这⾥。
如果超时触发了不止一次而上层这才想起进行读取呢 我们先让进程休眠3s并且将原先的超时时间3s更新为1s. 此时我们就会发现如果触发超时为1时读取到的tmp是1.因为进程先进行了睡眠了3s,这个期间超时触发已经发生了3次了。由此read函数调用不仅会处理一次超时触发也可以处理多次的超时触发并且进行清空。 时间轮思想 现在我们大抵是了解到了Linux为用户提供的定时器接口。基于上述例子我们可以设置服务器连接时间为n秒每隔n秒就可以去检测管理的连接里面是否存在长时间未通信超时的连接并把它们释放掉。 但这里存在一个很大的问题。每次超时都要将所有的连接遍历⼀遍连接数少还能接收可是如果有上万个连接呢效率⽆疑是较为低下的。当然提升效率的解决方法很多比如说你还可以根据每一个连接最近一次通信的时间构建一个小根堆这样每次只需要对堆顶部分的连接逐个释放直到没有超时的连接为⽌。不过这里我们采用一种时间轮转的方案。 什么是时间轮呢它的起源其实是来自于钟表。 比如我们现如今要设定一个5点钟的闹铃当钟表盘上的时针指向5时此时这个闹铃就会开始叮叮当当地躁动起来代表现在已经到了我们预设的时间。 我们将原型表盘抽象成计算机语言的符号那无非就是一个数组数组的下标就等于钟表盘上的数字而tick指针就等于那盘上不停顺时针旋转的时针当tick指向某一个下标对应的位置也就意味着该定时任务应当被执行了 不过这样以秒划分的时间盘如果只是执行短时间定时任务那开辟的空间是有限的。那如果是如果设置⼀⼩时后的定时任务呢则需要定义3600个元素的数组这样⽆疑是⽐较⿇的。因此可以采⽤多层级的时间轮有秒针轮分针轮时针轮60time3600则time/60就是分针轮对应存储的位置当tick/3600等于对应位置的时候将其位置的任务向分针秒针轮进⾏移动。 但是我们也得考虑⼀个问题当前的设计是时间到了则主动去执⾏定时任务释放连接。可是这种但凡需要人记住的操作难免会让人头疼总会有那一天忘记释放连接的操作。那有没有什么方法能够自动执行定时任务呢这时候我们就想到⼀个操作——类的析构函数。 如果仅仅是因为想搭上自动调用的析构函数这一条捷径就去设计出一个额外的类出来似乎不是很划算。并且我们又得考虑如果我们设置的是连接超时时间是30s在经历第10s的时候对端这时候又来进行通信那么此时也就意味着我们得延迟对该定时任务的执行也就是需要当tick遍历到40s的时候如果期间没有发生任何数据通信该连接应当被释放掉。所以当tick走到30s的时候我们就应该取消该执行的超时任务。 综上既满足能够在对象销毁时自动执行超时任务又能处理好延迟任务执行的问题我们这里就用到了智能指针shared_ptr。shared_ptr有个计数器当计数为0的时候才会真正释放⼀个对象那么如果连接在第10s进⾏了⼀次通信。则我们继续向定时任务中添加⼀个30s后(40s),的任务类对象的shared_ptr,则这时候两个任务shared_ptr计数为2。只有在第40s的时候shared_ptr计数器变为0了这个任务才会被真正释放。 Timer:
// 定时器任务
using OnTimerCallback std::functionvoid();
using RleaseCallback std::functionvoid();
class Timer
{
private:uint64_t _timerfd; // 定时器IDint _timeout; // 定时时间bool _cancealed; // false-任务正常执⾏ true-任务被取消// 定时任务、释放函数OnTimerCallback _timer_callback;RleaseCallback _release_callback;
public:Timer(uint64_t timerfd, int timeout) : _timerfd(fimerfd), _timeout(timeout),_cancealed(false) {}~Timer(){// 清理资源if (_release_callback)_release_callback();// 定时器任务if(_timer_callback !_cancealed) _timer_callback();}uint64_t get_id() { return _timerfd; }int delay_time() { return _timeout; }void canceled() { _canceled true; } // 取消定时任务// 设置回调void set_on_time_callback(const OnTimerCallback cb) { _timer_callback cb; }void set_release_callback(const RleaseCallback cb) { _release_callback cb; }
}; TimeQueue: 因为本次项目需要的超时等待时间不是很长因此使用单层轮转盘即可。
#define MAX_TIMEOUT 60
using PtrTimer std::shared_ptrTimer;
using WeakPtr std::weak_ptrTimer;
class TimeQueue
{
private:int _tick; // 任务指针int _capacity; // 轮盘时间std::vectorstd::vetorPtrTimer _wheel; // 时间轮盘// 判断定时任务是否存在 遍历_wheel是耗费时间的 因此需要建立 timerfd 和 PtrTimer 的映射// 但是如果使用PtrTimer -- 因为是shared_ptr 所以建立映射反而会多增加其计数器// 避免计数器增加所以这里会使用weak_ptrstd::unordered_mapuint64_t, WeakPtr _timers;private:void remove_weaktimer_from_timequeue(int timerfd){auto iter _timers.find(timerfd);if (iter ! _timers.end()){_timers.erase(iter);}}public:TimeQueue(int capacity MAX_TIMEOUT) : _tick(0), _capacity(capacity), _wheel(_capacity) {}void run_ontime_queue(){_tick (_tick 1) % _capacity;// 销毁_wheel[_tick].clear();}// 判断timer是否存在bool has_timer(int timerfd){auto iter _timers.find(timerfd);if (iter _timers.end())return false;return true;}// 添加定时任务void timer_add(const OnTimerCallback cb, int delay, uint64_t timerfd){if (delay 0 || delay _capacity)return;PtrTimer new_timer(new Timer(timerfd, delay));// 设置定时任务对象要执⾏的定时任务--会在对象被析构时执⾏new_timer-set_on_time_callback(cb);// _timers中保存了⼀份定时任务对象的weak_ptr因此希望在析构的同时进⾏移除new_timer-set_release_callback(std::bind(TimeQueue::remove_weaktimer_from_timequeue, this, timerfd));// 添加仅_wheel之中int pos (_tick delay) % _capacity;_wheel[pos].push_back(new_timer);_timers[timerfd] WeakPtr(new_timer);}// 延迟任务void timer_refresh(uint64_t timerfd){auto iter _timers.find(timerfd);assert(iter ! _timers.end());int delay iter-second.lock()-delay_time();int pos (_tick delay) % _capacity;// 刷新_wheel[pos].push_back(PtrTimer(iter-second));}// 取消任务void timer_cancealed(uint64_t timerfd){auto iter _timers.find(timerfd);assert(iter ! _timers.end());// lock()获取shared_ptrPtrTimer timer iter-second.lock();if (timer)timer-canceled(); // 取消任务}
};
测试: (3) 正则库的简单使用: 正则表达式(regularexpression)描述了⼀种字符串匹配的模式pattern,正则表达式(regularexpression)描述了⼀种字符串匹配的模式pattern。 正则表达式的使⽤可以使得HTTP请求的解析更加简单、更灵活但这并不意味着会比直接处理字符串的效率要快。 C11中为开发人员提供了Regex库: 这儿有详细的正则表达式通配符表需要的话可以了解一下。点我 解析HTTP请求行: 一个正常的http请求报头含有下字段: 请求行头部字段空行正文(有时不存在)。
① 获取method(GET|POST|HEAD|PUT|DELETE): 表示匹配并提取其中任意一个字符串.
② 获取资源路径([^?]*): [^?] 匹配非问号字符 后边的*表示0次或多次.
③ 提交参数(?:\\?(.*))?空格: ?:表示匹配某个字符串但不提取这里是要匹配?但不提取。 “\\?”:表示原始字符 ?。(.*)表示提取?之后的一个或多个字符直到遇到空格. 末尾?:表示匹配前一个表达式 0次或多次。
④ 提取协议版本(HTTP/1.\\[10]):表示匹配以HTTP/1.开始 后边有个0或1的字符串.
⑤ 空行(?:\n\r|\n)?: (?: ...)表示匹配某个格式字符串但是不提取. 最后的表示的是匹配前边的表达式0次或1次. (4) 同用类型Any: 每⼀个Connection对连接进⾏管理最终都不可避免需要涉及到应⽤层协议的处理因此在 Connection中需要设置协议处理的上下⽂来控制处理节奏。可是协议有千千万万种为了降低数据与协议的耦合度我们一定得让这个协议接收解析上下⽂就不能有明显的协议倾向它可以是任意协议的上下⽂信息也可以不是。所以我们需要一种特殊的数据结构用来保存不同数据。 在C语⾔中通⽤类型可以使⽤void*来管理但是在C中boost库和C17给我们提供了⼀个通⽤类型any来灵活使⽤。
C17标准库any容器使用: 自主实现Any类: 当然我们自己也可以实现一份简单的Any类。
● 首先any类一定不是一个模板类否则编译的时候 Anyint a, Anyfloatb,需要传类型作为模板参数也就是说在使⽤的时候才确定其类型。
● 我们也不知道支持的是什么协议它的上下文数据是什么类型的数据就更不知道应该传递什么模板参数。
● 所以我们需要在Any内部设计⼀个 模板容器holder类 可以保存各种类型数据但Any类中⽆法定义这个holder对象或指针因为any也不知道这个类要保存什么类型的数据因而⽆法传递类型参数。
● 定义⼀个基类placehoder让holder继承于placeholde⽽Any类保存⽗类指针即可。当需要保存数据时则new⼀个带有模板参数的⼦类holder对象出来保存数据然后让Any类中的⽗类指针指向这个⼦类对象就搞定了。
#pragma once
#include iostream
#include cassert
// 通过使用模板构造函数擦除模板类的参数类型。
class Any
{
public:Any() : _content(nullptr) {}~Any(){if (_content)delete _content;}template typename TAny(const T val) : _content(new holderT(val)) {}// 拷贝构造、赋值Any(const Any other) : _content(other._content ? other._content-clone() : nullptr) {}void swap(Any other){std::swap(_content, other._content);}template typename TAny operator(const T val){// 为val构建⼀个临时对象出来然后进⾏交换.// 这样临时对象销毁的时候顺带原先,保存的placeholder也会被销毁Any(val).swap(*this);return *this;}Any operator(Any other){// 这⾥要注意形参只是⼀个临时对象进⾏交换后就会释放// 所以交换后原先保存的placeholder指针也会被销毁other.swap(*this);return *this;}public:template typename T // anyT.get()T *get(){assert(typeid(T) _content-type());return ((holderT *)_content)-_val;}private:// 模板类编译时就会确定类型class placeholder{public:virtual ~placeholder() {}virtual const std::type_info type() 0;virtual placeholder *clone() 0;};// 声明⼀个holder模板类出来使⽤holder类来管理传⼊的对象// ⽽Any类只需要管理holder对象即可template typename Tclass holder : public placeholder{public:holder(const T v) : _val(v) {}~holder() {}virtual const std::type_info type() { return typeid(T); }virtual placeholder *clone() { return new holder(_val); }T _val;};// Any只需要用一个父类指针管理placeholder *_content;
};
测试 SERVER模块实现: 我们的所有实现都放在.hpp一个文件下。
日志宏实现: 我们使用宏函数完成 日志打印功能的实现,这个日志可以根据设置等级打印或不打印等级较低的日志。
#define INF 0
#define DBG 1
#define ERR 2
#define DEFAULT_LOG_LEVEL INF
#define LOG(level, format, ...) \do \{ \if (level DEFAULT_LOG_LEVEL) \break; \time_t times time(nullptr); \struct tm *t localtime(times); \char ts[32] {0}; \strftime(ts, sizeof(ts), %H:%M:%S, t); \fprintf(stdout, [%s:%d] [%s] format \n, __FILE__, __LINE__, ts, ##__VA_ARGS__); \} while (0)#define INF_LOG(format, ...) LOG(INF, format, ##__VA_ARGS__)
#define DBG_LOG(format, ...) LOG(DBG, format, ##__VA_ARGS__)
#define ERR_LOG(format, ...) LOG(ERR, format, ##__VA_ARGS__) Buffer类: 用于实现⽤⼾态缓冲区提供数据缓冲取出等功能.
// 实现⽤⼾态缓冲区提供数据缓冲
const int buffer_default_size 1024;
class Buffer
{
private:// 选用vector而不选用string 是考虑到 传输的数据含0字符的情况std::vectorchar _buffer;// 记录buffer内数据读取和写入位置uint64_t _reader_idx; // 读偏移uint64_t _writer_idx; // 写偏移// 管理读写位置
public:Buffer() : _reader_idx(0), _writer_idx(0), _buffer(buffer_default_size) {}char *Begin() { return *(_buffer.begin()); }// 获取当前写⼊起始地址char *WritePosition() { return Begin() _writer_idx; }// 获取当前读取起始地址char *ReadPosition() { return Begin() _reader_idx; }// 获取缓冲区末尾空闲空间⼤⼩uint64_t TailIdleSize() { return _buffer.size() - _writer_idx; }// 获取缓冲区起始空闲空间⼤⼩ -- 这是获取可覆盖空间uint64_t HeadIdleSize() { return _reader_idx; }// 获取可读数据⼤⼩ 写偏移 - 读偏移uint64_t ReadAbleSize() { return _writer_idx - _reader_idx; }// 将 读偏移 向后移动void MoveReadOffset(uint64_t len){if (len 0)return;// 读偏移向后移动 len不能超过可读数据大小assert(len ReadAbleSize());_reader_idx len;}// 将写偏移向后移动void MoveWriteOffset(uint64_t len){if (len 0)return;// 写偏移向后移动 len必须⼩于当前后边的空闲空间⼤⼩assert(len TailIdleSize());_writer_idx len;}// 管理读取、写入
public:void EnsureWriteSpace(uint64_t len){// 可写空间的总大小为: TailIdleSize() HeadIdleSize()// 1.如果末尾如果够插入 直接返回if (len TailIdleSize())return;// 2.如果不超过 可写空间的总大小 把原有数据向前挪动if (len TailIdleSize() HeadIdleSize()){// 原先数据大小uint64_t res ReadAbleSize();// 向前拷贝合并空间std::copy(ReadPosition(), ReadPosition() res, Begin());// 更新偏移量_reader_idx 0;_writer_idx res;}else{// 3.总体空间不够则需要扩容不移动数据直接给写偏移之后扩容⾜够空间即可_buffer.resize(_writer_idx len);}}// 真正的写入数据void Write(const void *data, size_t len){// 1.保证空间足够if (len 0)return;EnsureWriteSpace(len);// 2.数据拷贝const char *d (const char *)data;std::copy(d, d len, WritePosition());}// 写入数据 - string类型void WriteString(const std::string data){WriteAndPush(data.c_str(), data.size());}// 写入数据 - buffer类型void WriteBuffer(Buffer buf){WriteAndPush(buf.ReadPosition(), buf.ReadAbleSize());}// 写入数据移动偏移量 —— 最好使用这个void WriteAndPush(const void *data, size_t len){Write(data, len);MoveWriteOffset(len);}// 真正的取出数据void Read(void *data, size_t len){if (len 0)return;// 要获取的数据⼤⼩必须⼩于可读数据⼤⼩assert(len ReadAbleSize());std::copy(ReadPosition(), ReadPosition() len, (char *)data);}// 按照字符串方式取出 数据std::string ReadAsString(size_t len){if (len 0)return ;assert(len ReadAbleSize());std::string str;str.resize(len);ReadAndPop(str[0], len);return str;}// 取出数据移动偏移量 —— 最好使用这个void ReadAndPop(void *data, size_t len){Read(data, len);MoveReadOffset(len);}// 清空缓冲区void Clear(){_reader_idx 0;_writer_idx 0;}// HTTP处理char *FindCRLF(){char *res (char *)memchr(ReadPosition(), \n, ReadAbleSize());return res;}// 通常用于获取一行数据std::string GetOneLine(){char *pos FindCRLF();if (pos nullptr)return ;// 这里的1 是将\n一并取出来return ReadAsString(pos - ReadPosition() 1);}
}; 基础套接字Socket:
// 为避免服务器向已经关闭的文件描述符输入
// OS会发送SIGPIPE信号终止程序
// 大多数服务器都会选择将这个信号忽略掉
class NetWork
{
public:NetWork(){INF_LOG(SIGPIPIE INIT);signal(SIGPIPE, SIG_IGN);}
};
// 定义静态全局是为了保证构造函数中的信号忽略处理能够在程序启动阶段就被直接执⾏
static NetWork nw;#define MAX_LISTEN 1024
class Socket
{
private:int _sockfd;public:Socket() : _sockfd(-1) {}Socket(int fd) : _sockfd(fd) {}~Socket() { Close(); }int get_fd() { return _sockfd; }void Close(){if (_sockfd 0){close(_sockfd);_sockfd -1;}}// 套接字创建
private:bool Create(){_sockfd socket(AF_INET, SOCK_STREAM, IPPROTO_IP);if (_sockfd 0){ERR_LOG(create socket faild...);return false;}return true;}bool Bind(const std::string ip, uint16_t port){struct sockaddr_in local;local.sin_family AF_INET;local.sin_port htons(port);local.sin_addr.s_addr INADDR_ANY;if (bind(_sockfd, (const sockaddr *)local, sizeof(local)) 0){ERR_LOG(bind socket faild...);return false;}return true;}bool Listen(int backlog MAX_LISTEN){if (listen(_sockfd, backlog) 0){ERR_LOG(listen socket faild...);return false;}return true;}void ReuseAddr(){int flag 1;setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, flag, sizeof(flag));int val 1;setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, val, sizeof(val));}// 获取连接、建立连接
public:bool Connect(const std::string ip, uint16_t port){struct sockaddr_in peer;peer.sin_family AF_INET;peer.sin_port htons(port);peer.sin_addr.s_addr inet_addr(ip.c_str());int ret connect(_sockfd, (const sockaddr *)peer, sizeof(peer));if (ret 0){ERR_LOG(connect socket faild...);return false;}return true;}int Accept(){// 这里不关心 发起连接一方的信息int newfd accept(_sockfd, nullptr, nullptr);if (newfd 0){ERR_LOG(accept socket faild...);return -1;}return newfd;}void NonBlock(){// 获取_sockfd模式int flag fcntl(_sockfd, F_GETFL, 0);// 设置非阻塞fcntl(_sockfd, F_SETFL, flag | O_NONBLOCK);}// 套接字的读和写 -- 真正的读写操作// buffer -- 只提供策略
public:ssize_t Recv(void *buf, size_t len, int flag 0){ssize_t s recv(_sockfd, buf, len, flag);if (s 0){// 1.没有出错 只是缓冲区没数据或者被信号中断if (errno EAGAIN || errno EINTR){// 表⽰这次接收没有接收到数据return 0;}// 2.真的出错了ERR_LOG(socket recv faild...);return -1;}// 实际接收的数据⻓度return s;}size_t Recv_NonBlock(void *buf, size_t len){int size Recv(buf, len, MSG_DONTWAIT);return size;}int Send(const void *buf, size_t len, int flag 0){ssize_t ret send(_sockfd, buf, len, flag);if (ret 0){if (errno EAGAIN || errno EINTR){return 0;}ERR_LOG(socket send faild...);return -1;}return ret;}ssize_t Send_NonBlock(const void *buf, size_t len){return Send(buf, len, MSG_DONTWAIT);}// 构建服务端\客户端
public:bool CreateServer(uint16_t port, const std::string ip 0.0.0.0, bool block_flag false){if (Create() false)return false;if (block_flag)NonBlock();if (Bind(ip, port) false)return false;if (Listen() false)return false;ReuseAddr();return true;}bool CreateClient(uint16_t port, const std::string ip){if (Create() false)return false;if (Connect(ip, port) false)return false;return true;}
}; 事件监控管理Channel:
// 每一个socket都对应一个 Channel
// 该Channel关心这个描述符上的设置的事件
// 当事件就绪时 就会调用被设置进的 回调函数
class Poller;
class EventLoop;
class Channel
{
private:int _fd;uint32_t _events; // 该事件需要关心的 事件uint32_t _revents; // 就绪事件// One thread One Loop// 当一个线程 去处理一个开启监控的描述符上的事件// 它不是通过Channel子模块而是通过EventLoop这个整合 事件监控、管理、修改等等的大模块// 找到对应描述符上关心的Channel事件这里设置_loop是一种会指机制// 该Channel事件的处理 是放在这一个Loop这个thread之中的EventLoop *_loop;using EventCallback std::functionvoid();EventCallback _read_callback; // 可读事件被触发的回调函数EventCallback _write_callback; // 可写事件被触发的回调函数EventCallback _error_callback; // 错误事件被触发的回调函数EventCallback _close_callback; // 连接断开事件被触发的回调函数EventCallback _event_callback; // 任意事件被触发的回调函数
public:// 回调函数设置void SetReadCallback(const EventCallback cb) { _read_callback cb; }void SetWriteCallback(const EventCallback cb) { _write_callback cb; }void SetErrorCallback(const EventCallback cb) { _error_callback cb; }void SetCloseCallback(const EventCallback cb) { _close_callback cb; }void SetEventCallback(const EventCallback cb) { _event_callback cb; }public:Channel(EventLoop *loop, int fd) : _fd(fd), _events(0), _revents(0), _loop(loop){}int get_fd() { return _fd; }// 获取想要监控的事件uint32_t get_events() { return _events; }// 设置实际就绪的事件void set_revents(uint32_t events) { _revents events; }// 事件监控bool ReadAble() { return _events EPOLLIN; }bool WriteAble() { return _events EPOLLOUT; }// 启动\关闭读写事件// 这里的update和 Poller(修改事件监控)相关 但我们可以通过回指loop指针 调用Poller里的内容void EnableRead(){_events | EPOLLIN;Update();}void EnableWrite(){_events | EPOLLOUT;Update();}void DisableRead(){_events ~EPOLLIN;Update();}void DisableWrite(){_events ~EPOLLOUT;Update();}void DisableAll(){_events 0;Update();}void Update();void Remove();public:// 事件处理⼀旦连接触发了事件就调⽤这个函数⾃⼰触发了什么事件如何处理⾃⼰决定void HandlerEvent(){// 这些都与 读事件相关if ((_revents EPOLLIN) || (_revents EPOLLRDHUP) || (_revents EPOLLPRI)){if (_read_callback)_read_callback();}/*有可能会释放连接的操作事件⼀次只处理⼀个*/if (_revents EPOLLOUT){if (_write_callback)_write_callback();}else if (_revents EPOLLERR){if (_error_callback)_error_callback();}else if (_revents EPOLLHUP){if (_close_callback)_close_callback();}if (_event_callback)_event_callback();}
};#define MAX_EPOLL_EVENTS 1024
class Poller
{
private:int _epfd;struct epoll_event _evs[MAX_EPOLL_EVENTS]; // 通过就绪队列 获取的就绪事件信息// [描述符,Channel]// 记录有多少描述符的Channel需要被管控std::unordered_mapint, Channel * _channels;private:// 真正修改监控// 增删改void Update(Channel *channel, int op){int fd channel-get_fd();struct epoll_event ev;ev.data.fd fd;ev.events channel-get_events();int ret epoll_ctl(_epfd, op, fd, ev);if (ret 0){ERR_LOG(epoll ctl error:%s\n, strerror(errno));}return;}bool HasChannel(Channel *channel){auto iter _channels.find(channel-get_fd());if (iter _channels.end())return false;return true;}public:Poller(){_epfd epoll_create(MAX_EPOLL_EVENTS);if (_epfd 0){ERR_LOG(epoll create error:%s\n, strerror(errno));abort();}}void UpdateEvent(Channel *channel){bool ret HasChannel(channel);if (ret false){// 不存在 就添加_channels.insert(std::make_pair(channel-get_fd(), channel));return Update(channel, EPOLL_CTL_ADD);}return Update(channel, EPOLL_CTL_MOD);}void RemoveEvent(Channel *channel){auto iter _channels.find(channel-get_fd());if (iter ! _channels.end()){_channels.erase(iter);}return Update(channel, EPOLL_CTL_DEL);}// 输出型参数带出就绪事件的Channelvoid Epoll(std::vectorChannel * *active){// -1: 阻塞等待int nfds epoll_wait(_epfd, _evs, MAX_EPOLL_EVENTS, -1);if (nfds 0){if (errno EINTR){return;}ERR_LOG(epoll wait error:%s\n, strerror(errno));abort();}// 事件就绪for (int i 0; i nfds; i){auto iter _channels.find(_evs[i].data.fd);assert(iter ! _channels.end());iter-second-set_revents(_evs[i].events); // 设置事件就绪active-push_back(iter-second); // 插入就绪事件数组}return;}
};// Channel才保存着 监控哪些事件的信息因此参数传this
void Channel::Remove() { return _loop-RemoveEvent(this); }
void Channel::Update() { return _loop-UpdateEvent(this); } 描述符控制更改Poller:
#define MAX_EPOLL_EVENTS 1024
class Poller
{
private:int _epfd;struct epoll_event _evs[MAX_EPOLL_EVENTS]; // 通过就绪队列 获取的就绪事件信息// [描述符,Channel]// 记录有多少描述符的Channel需要被管控std::unordered_mapint, Channel * _channels;private:// 真正修改监控// 增删改void Update(Channel *channel, int op){int fd channel-get_fd();struct epoll_event ev;ev.data.fd fd;ev.events channel-get_events();int ret epoll_ctl(_epfd, op, fd, ev);if (ret 0){ERR_LOG(epoll ctl error:%s\n, strerror(errno));}return;}bool HasChannel(Channel *channel){auto iter _channels.find(channel-get_fd());if (iter _channels.end())return false;return true;}public:Poller(){_epfd epoll_create(MAX_EPOLL_EVENTS);if (_epfd 0){ERR_LOG(epoll create error:%s\n, strerror(errno));abort();}}void UpdateEvent(Channel *channel){bool ret HasChannel(channel);if (ret false){// 不存在 就添加_channels.insert(std::make_pair(channel-get_fd(), channel));return Update(channel, EPOLL_CTL_ADD);}return Update(channel, EPOLL_CTL_MOD);}void RemoveEvent(Channel *channel){auto iter _channels.find(channel-get_fd());if (iter ! _channels.end()){_channels.erase(iter);}return Update(channel, EPOLL_CTL_DEL);}// 输出型参数带出就绪事件的Channelvoid Epoll(std::vectorChannel * *active){// -1: 阻塞等待int nfds epoll_wait(_epfd, _evs, MAX_EPOLL_EVENTS, -1);if (nfds 0){if (errno EINTR){return;}ERR_LOG(epoll wait error:%s\n, strerror(errno));abort();}// 事件就绪for (int i 0; i nfds; i){auto iter _channels.find(_evs[i].data.fd);assert(iter ! _channels.end());iter-second-set_revents(_evs[i].events); // 设置事件就绪active-push_back(iter-second); // 插入就绪事件数组}return;}
}; 定时任务管理TimerWheel类实现:
// 定时任务/清理资源函数
using TaskFunc std::functionvoid();
using ReleaseFunc std::functionvoid();
class TimerTask
{
private:int _timerfd; // 定时器对象iduint32_t _timeout; // 定时任务超时时间bool _canceled; // 定时任务是否被取消TaskFunc _task_cb; // 定时器对象要执⾏的定时任务ReleaseFunc _release_cb; // ⽤于删除TimerWheel中保存的定时器对象信息
public:TimerTask(int timerfd, uint32_t timeout, const TaskFunc cb) : _timerfd(timerfd), _timeout(timeout),_task_cb(cb), _canceled(false) {}~TimerTask(){if (_canceled false)_task_cb();_release_cb();}void SetReleaseCallback(const ReleaseFunc cb) { _release_cb cb; }void Cancel() { _canceled true; }uint32_t get_timeout() { return _timeout; }
};class TimerWheel
{
private:using WeakTask std::weak_ptrTimerTask;using PtrTask std::shared_ptrTimerTask;int _tick; // 当前的秒针⾛到哪⾥释放哪⾥释放哪⾥执行该任务int _capacity; // 表盘容量std::vectorstd::vectorPtrTask _TimerWheels; // 表盘std::unordered_mapuint64_t, WeakTask _timers; // 已经存在TimerTaskEventLoop *_loop;// 定时任务通过Channel进行读监控// Channel _timer_channel;// 定时器描述符--可读事件回调就是读取计数器执⾏定时任务int _timerfd;std::unique_ptrChannel _timer_channel;private:void RemoveTimer(uint64_t id){// 清理weak_ptr的 对象auto it _timers.find(id);if (it ! _timers.end()){_timers.erase(it);}}int CreateTimerfd(){int timerfd timerfd_create(CLOCK_MONOTONIC, 0);if (timerfd 0){ERR_LOG(TIMERFD CREATE FAILED!);abort();}struct itimerspec spec;spec.it_value.tv_sec 1;spec.it_value.tv_nsec 0;spec.it_interval.tv_sec 1;spec.it_interval.tv_nsec 0;timerfd_settime(timerfd, 0, spec, nullptr);return timerfd;}// 每次超时会向_fd写入数据 触发读事件// 回调Ontime 处理tick指针int ReadTImerfd(){// 每秒向_timer写入int times;int ret read(_timerfd, times, 8);if (ret 0){ERR_LOG(READ TIMEFD FAILED!);abort();}return times;}void RunTimerTick(){_tick (_tick 1) % _capacity;// 这里清空数组内容 保存在里面的对象 会自动调用析构函数 - 回调设置的超时任务_TimerWheels[_tick].clear();}void OnTime(){// 根据实际超时的次数执⾏对应的超时任务int times ReadTImerfd();// 每读取一次 就移动_tickfor (int i 0; i times; i){RunTimerTick();}}void TimerAddInLoop(uint64_t id, int delay, const TaskFunc cb){PtrTask ptr(new TimerTask(id, delay, cb));// 这里就设置 Release的callbackptr-SetReleaseCallback(std::bind(TimerWheel::RemoveTimer, this, id));int pos (_tick delay) % _capacity;_TimerWheels[pos].push_back(ptr);_timers[id] WeakTask(ptr);}void TimerRefreshInLoop(uint64_t id){auto iter _timers.find(id);if (iter _timers.end()){return;}// lock 获取weakptr中的shared_ptrPtrTask ptr iter-second.lock();// 重新计算位置插入int delay ptr-get_timeout();int pos (_tick delay) % _capacity;_TimerWheels[pos].push_back(ptr);}void TimerCancelInLoop(uint64_t id){auto iter _timers.find(id);assert(iter ! _timers.end());PtrTask ptr iter-second.lock();if (ptr)ptr-Cancel();}public:TimerWheel(EventLoop *loop) : _tick(0), _capacity(60), _TimerWheels(_capacity), _loop(loop),_timerfd(CreateTimerfd()), _timer_channel(new Channel(_loop, _timerfd)){// 设置读事件回调_timer_channel-SetReadCallback(std::bind(TimerWheel::OnTime, this));_timer_channel-EnableRead();}// 定时任务(添加\刷新\取消)void TimerAdd(uint64_t id, uint32_t delay, const TaskFunc cb);void TimerRefresh(uint64_t id);void TimerCancel(uint64_t id);bool HasTimer(uint64_t id){auto iter _timers.find(id);if (iter _timers.end())return false;return true;}
};// 定时器中有个_timers成员定时器信息的操作有可能在多线程中进⾏因此需要考虑线程安全问题
// 如果不想加锁 One thread one loop 都放到⼀个线程中进⾏
void TimerWheel::TimerAdd(uint64_t id, uint32_t delay, const TaskFunc cb)
{_loop-RunInLoop(std::bind(TimerWheel::TimerAddInLoop, this, id, delay, cb));
}
void TimerWheel::TimerRefresh(uint64_t id)
{_loop-RunInLoop(std::bind(TimerWheel::TimerRefreshInLoop, this, id));
}
void TimerWheel::TimerCancel(uint64_t id)
{_loop-RunInLoop(std::bind(TimerWheel::TimerCancelInLoop, this, id));
} Reactor-EventLoop实现 Eventloop集成Channel、Poller、TimerWheel三个模块彻底对文件描述符事件进行统一监控管理数据的收发并展开业务处理。 什么是eventfd? 这是一个用来唤醒event事件的文件描述符我们下来看看如何对它进行使用以及它能呈现什么样的效果出来。
为什么需要eventfd? 创建eventfd进行可读可写是为了当向EventLoop里的任务队列中塞任务时可以通过eventd进行读写事件从而来唤醒底层的epoll_wait()处理任务池中的任务。 一些不可以用可读可写进行监控的操作例如释放连接当压入任务池中的时候就需要借助eventfd唤醒epoll_wait()对任务池里的任务继续处理。 设计任务池的好处在于一个Loop对应一个线程避免因为资源竞争带来的负面影响影响处理性能。 如何理解任务队列 EventLoop对描述符进行事件的监控以及事件的处理。而每一个EventLoop对应唯一的一个线程。 在多线程环境下因为描述符资源是共享的一旦一个描述符上的连接事件就绪多线程都会触发这个事件进行处理那么一定会引起线程安全问题。所以我们需要做的是让一个描述的监控、事件的处理以及对该描述符的其他操作都放在同一个线程中进行。 所以我们设计一个任务队列(池)对一个描述符的所有操作进行一次封装对连接的操作并不立即执行而是将任务添加到任务队列中。
EventLoop处理流程
① 在线程中对该文件描述符进行时间监控。
② 有描述符就绪则对事件进行处理。(将这些回调函数压入任务池当中)。
③ 所有就绪事件处理完成后再将任务池中的所有任务一一一去执行。 这里的效率并非体现在压入任务上因为我们也得在操作任务队列时给这个队列加锁保证线程安全。它的高效在于压入的任务能够并发进行处理。这种思想就像生产消费者模型一样。 具体实现:
// 被压入进EventLoop进行处理
class EventLoop
{
private:// 可以是任意回调函数!using Functor std::functionvoid();// one thread one loopstd::thread::id _thread_id;// 唤醒IO事件监控有可能导致的阻塞int _event_fd;std::unique_ptrChannel _event_channel;Poller _poller;std::vectorFunctor _tasks; // 任务池std::mutex _mtx; // 任务池线程安全TimerWheel _timer_wheel; // 定时器模块
private:static int CreateEventFd(){int efd eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);if (efd 0){ERR_LOG(CREATE EVENTFD FAILED!!);abort(); // 让程序异常退出}return efd;}// 执行任务池里的所有任务void RunAllTask(){std::vectorFunctor functor;{std::unique_lockstd::mutex _lock(_mtx);_tasks.swap(functor);}for (auto f : functor){f();}return;}// 设置eventfd唤醒的 回调函数// 触发读事件void ReadEventfd(){uint64_t res 0;int ret read(_event_fd, res, sizeof(res));if (ret 0){// EINTR -- 被信号打断 EAGAIN -- 表⽰⽆数据可读if (errno EINTR || errno EAGAIN){return;}ERR_LOG(READ EVENTFD FAILED!);abort();}return;}void WeakUpEventFd(){uint64_t val 1;int ret write(_event_fd, val, sizeof(val));if (ret 0){if (errno EINTR){return;}ERR_LOG(READ EVENTFD FAILED!);abort();}return;}public:EventLoop() : _thread_id(std::this_thread::get_id()), _event_fd(CreateEventFd()),_event_channel(new Channel(this, _event_fd)),_timer_wheel(this){// 给eventfd添加可读事件回调函数读取eventfd事件通知次数_event_channel-SetReadCallback(std::bind(EventLoop::ReadEventfd, this));_event_channel-EnableRead();}void Start(){// 循环运行:// 1.事件监控 2.事件处理 3.执行任务while (1){std::vectorChannel * actives;_poller.Epoll(actives); // 监控waitfor (auto channel : actives){// 执行回调 处理事件就绪channel-HandlerEvent();}// 执行任务池里的任务RunAllTask();}}// 判断将要执⾏的任务是否处于当前线程中如果是则执⾏不是则压⼊队列void RunInLoop(const Functor cb){if (IsInLoop()){return cb();}return QueueInLoop(cb);}void QueueInLoop(const Functor cb){{std::unique_lockstd::mutex _lock(_mtx);_tasks.push_back(cb);}// 唤醒有可能因为没有事件就绪⽽导致的epoll阻塞// 其实就是给eventfd写⼊⼀个数据eventfd就会触发可读事件WeakUpEventFd();}// ⽤于判断当前线程是否是EventLoop对应的线程bool IsInLoop() { return _thread_id std::this_thread::get_id(); }void AssertInLoop() { assert(_thread_id std::this_thread::get_id()); }// 控制描述符void UpdateEvent(Channel *channel) { return _poller.UpdateEvent(channel); }void RemoveEvent(Channel *channel) { return _poller.RemoveEvent(channel); }// 管理超时任务void TimerAdd(uint64_t id, uint32_t delay, const TaskFunc cb) { return _timer_wheel.TimerAdd(id, delay, cb); }void TimerRefresh(uint64_t id) { return _timer_wheel.TimerRefresh(id); }void TimerCancel(uint64_t id) { return _timer_wheel.TimerCancel(id); }bool HasTimer(uint64_t id) { return _timer_wheel.HasTimer(id); }
}; LoopThread\LoopThreadPool: 既然是要奉行one thread one loop的设计精髓我们如何管理线程和loops的关系呢
法一: 先创建loops实例化后分配给线程
法二: 先创建线程并在线程内部实例化loops 对于法子一而言看似这样行得通可是当你用主线程先去实例化loops的时候其他线程是没有被创建出来的也就是即便你之后创建出来如何让loops和线程对应进行分配也是难以控制的由此loops中的操作反而完完全全只能在主线程中执行。所以我们先得把线程创建好再在每个线程的内部实例化loop(从属Reactor)。
LoopThread:
// 避免线程创建了但是_loop还没有实例化
class LoopThread
{
private:std::mutex _mtx;std::condition_variable _cond;// EventLoop指针变量这个对象需要在线程内实例化// 线程对应的_loopEventLoop *_loop;// EventLoop对应的线程std::thread _thread;private:void ThreadEntry(){EventLoop loop;{std::unique_lockstd::mutex lock(_mtx); // 加锁_loop loop;_cond.notify_all();}loop.Start();}public:LoopThread() : _loop(NULL), _thread(std::thread(LoopThread::ThreadEntry, this)) {}EventLoop *GetLoop(){EventLoop *loop NULL;{std::unique_lockstd::mutex lock(_mtx); // 加锁_cond.wait(lock, [](){ return _loop ! NULL; }); // loop为NULL就⼀直阻塞loop _loop;}return loop;}
}; LoopThreadPool:
class LoopThreadPool
{
private:int _thread_count;int _next_idx; // 轮询控制 取出Loop池中的 从属Reactor// 主Reactor: 仅仅用于监听套接字EventLoop *_baseloop;// 线程与从属Reactorstd::vectorLoopThread * _threads;std::vectorEventLoop * _loops;public:LoopThreadPool(EventLoop *baseloop) : _thread_count(0), _next_idx(0), _baseloop(baseloop) {}void SetThreadCount(int count) { _thread_count count; }void Create(){if (_thread_count 0){// one thread one loop_threads.resize(_thread_count);_loops.resize(_thread_count);for (int i 0; i _thread_count; i){// 从LoopThread获取_threads[i] new LoopThread();// one thread one loop_loops[i] _threads[i]-GetLoop();}}return;}// 获取EventLoopEventLoop *NextLoop(){if (_thread_count 0){return _baseloop;}_next_idx (_next_idx 1) % _thread_count;return _loops[_next_idx];}
}; 通信连接管理Connection类实现:
class Connection;
typedef enum
{DISCONNECTED,CONNECTING,CONNECTED,DISCONNECTING
} ConnStatu;using PtrConnection std::shared_ptrConnection;
class Connection : public std::enable_shared_from_thisConnection
{
private:uint64_t _conn_id; // 连接的唯⼀ID便于连接的管理和查找// 这里简便让 _conn_id作为定时器IDint _sockfd; // 连接关联的⽂件描述符bool _enable_inactive_release; // 连接是否启动⾮活跃销毁的判断标志默认为falseConnStatu _status; // 连接状态EventLoop *_loop; // 这个连接关联的loopSocket _socket; // 套接字管理Channel _channel; // 连接的事件管理Buffer _in_buffer; // 输⼊缓冲区---存放从socket中读取到的数据Buffer _out_buffer; // 输出缓冲区---存放要发送给对端的数据Any _context; // 请求的接收处理上下⽂// 这四个回调函数是让服务器模块来设置的using ConnectedCallback std::functionvoid(const PtrConnection );using MessageCallback std::functionvoid(const PtrConnection , Buffer *);using ClosedCallback std::functionvoid(const PtrConnection );using AnyEventCallback std::functionvoid(const PtrConnection );ConnectedCallback _connected_callback;MessageCallback _message_callback;ClosedCallback _closed_callback;AnyEventCallback _event_callback;// 服务器会进行Connection的管理 这里是 从服务器删除该connection信息ClosedCallback _server_closed_callback;private:// 五个channel的事件回调函数// 描述符可读事件触发后调⽤的函数接收socket数据放到接收缓冲区中然后调⽤_message_callbackvoid HandlerRead(){// 1.接收socket的数据放到缓冲区char buf[65536];ssize_t ret _socket.Recv_NonBlock(buf, 65535);if (ret 0){// 出错了,不能直接关闭连接return ShutdownInLoop();}// 0表示的是没有读取到数据// -1表示连接断开// 更新缓冲区数据_in_buffer.WriteAndPush(buf, ret);// 调⽤message_callback进⾏业务处理if (_in_buffer.ReadAbleSize() 0){return _message_callback(shared_from_this(), _in_buffer);}}// 描述符可写事件触发后调⽤的函数将发送缓冲区中的数据进⾏发送void HandlerWrite(){//_out_buffer中保存的数据就是要发送的数据ssize_t ret _socket.Send_NonBlock(_out_buffer.ReadPosition(), _out_buffer.ReadAbleSize());if (ret 0){// 发送错误就该关闭连接了if (_in_buffer.ReadAbleSize() 0){// 数据没有发送完 就把剩余的发送_message_callback(shared_from_this(), _in_buffer);}return Release(); // 这时候就是实际的关闭释放操作了}// 注: 这里读偏移_out_buffer.MoveReadOffset(ret);if (_out_buffer.ReadAbleSize() 0){// 要发送的缓冲区没数据了 关闭监控_channel.DisableWrite();// 如果当前是连接待关闭状态则有数据发送完数据释放连接没有数据则直接释放if (_status DISCONNECTING){return Release();}}return;}// 描述符触发挂断事件void HandleClose(){// ⼀旦连接挂断了套接字就什么都⼲不了了因此有数据待处理就处理⼀下完毕关闭连接if (_in_buffer.ReadAbleSize() 0){_message_callback(shared_from_this(), _in_buffer);}return Release();}// 描述符触发出错事件void HandlerError(){HandleClose();}// 描述符触发任意事件: 1. 刷新连接的活跃度--延迟定时销毁任务 2. 调⽤组件使⽤者的任意事件回调void HandlerEvent(){if (_enable_inactive_release true){_loop-TimerRefresh(_conn_id);}if (_event_callback)_event_callback(shared_from_this());}// 启动⾮活跃连接超时释放规则void EnableInactiveReleaseInLoop(int sec){_enable_inactive_release true;// 如果当前定时销毁任务已经存在那就刷新延迟⼀下即可if (_loop-HasTimer(_conn_id))return _loop-TimerRefresh(_conn_id);// 如果不存在定时销毁任务则新增_loop-TimerAdd(_conn_id, sec, std::bind(Connection::Release, this));}void CancelInactiveReleaseInLoop(){_enable_inactive_release false;if (_loop-HasTimer(_conn_id)){_loop-TimerCancel(_conn_id);}}// 连接获取之后所处的状态下要进⾏各种设置启动读监控,调⽤回调函数void EstablishedInLoop(){assert(_status CONNECTING); // 当前一定是处于半连接状态下_status CONNECTED;// ⼀旦启动读事件监控就有可能会⽴即触发读事件如果这时候启动了⾮活跃连接销毁_channel.EnableRead();if (_connected_callback){_connected_callback(shared_from_this());}}// 这个关闭操作并⾮实际的连接释放操作需要判断还有没有数据待处理待发送void ShutdownInLoop(){// 设置连接为半关闭状态_status DISCONNECTING;// 处理业务逻辑没有处理的数据if (_in_buffer.ReadAbleSize() 0){if (_message_callback){_message_callback(shared_from_this(), _in_buffer);}}// 要么就是写⼊数据的时候出错关闭要么就是没有待发送数据直接关闭// 这里是真正发送数据if (_out_buffer.ReadAbleSize() 0){if (_channel.WriteAble() false){_channel.EnableWrite();}}if (_out_buffer.ReadAbleSize() 0){Release();}}// 实际释放窗口void ReleaseInLoop(){_status DISCONNECTED;_channel.Remove();_socket.Close();// 如果当前定时器队列中还有定时销毁任务则取消任务if (_loop-HasTimer(_conn_id)){CancelInactiveReleaseInLoop();}// 调⽤关闭回调函数避免先移除服务器管理的连接信息导致Connection被释// 此时去处理会出错因此先调⽤⽤⼾的回调函数if (_closed_callback)_closed_callback(shared_from_this());// 移除服务器内部管理的连接信息if (_server_closed_callback)_server_closed_callback(shared_from_this());}// 这个接⼝并不是实际的发送接⼝⽽只是把数据放到了发送缓冲区启动了可写事件监控void SendInLoop(Buffer buf){if (_status DISCONNECTED)return;_out_buffer.WriteBuffer(buf);if (_channel.WriteAble() false){_channel.EnableWrite();}}// 切换协议 重新设置函数void UpgradeInLoop(const Any context,const ConnectedCallback conn,const MessageCallback msg,const ClosedCallback closed,const AnyEventCallback event){_context context;_connected_callback conn;_message_callback msg;_closed_callback closed;_event_callback event;}public:Connection(EventLoop *loop, uint64_t conn_id, int sockfd) : _conn_id(conn_id), _sockfd(sockfd),_enable_inactive_release(false), _loop(loop),_status(CONNECTING), _socket(_sockfd),_channel(loop, _sockfd){_channel.SetReadCallback(std::bind(Connection::HandlerRead, this));_channel.SetWriteCallback(std::bind(Connection::HandlerWrite, this));_channel.SetCloseCallback(std::bind(Connection::HandleClose, this));_channel.SetErrorCallback(std::bind(Connection::HandlerError, this));_channel.SetEventCallback(std::bind(Connection::HandlerEvent, this));}~Connection() { DBG_LOG(RELEASE CONNECTION:%p, this); }// 获取文件描述符int get_fd() { return _sockfd; }// 获取iduint64_t get_id() { return _conn_id; }// 连接状态bool Connected() { return _status CONNECTED; }// 设置上下⽂--连接建⽴完成时进⾏调⽤void SetContext(const Any context) { _context context; }// 获取上下⽂返回的是指针Any *GetContext() { return _context; }// using ConnectedCallback std::functionvoid(const PtrConnection );// using MessageCallback std::functionvoid(const PtrConnection , Buffer *);// using ClosedCallback std::functionvoid(const PtrConnection );// using AnyEventCallback std::functionvoid(const PtrConnection );// ConnectedCallback _connected_callback;// MessageCallback _message_callback;// ClosedCallback _closed_callback;// AnyEventCallback _event_callback;void SetConnectedCallback(const ConnectedCallback cb) { _connected_callback cb; }void SetMessageCallback(const MessageCallback cb) { _message_callback cb; }void SetAnyEventCallback(const AnyEventCallback cb) { _event_callback cb; }void SetClosedCallback(const ClosedCallback cb) { _closed_callback cb; }void SetSrvClosedCallback(const ClosedCallback cb) { _server_closed_callback cb; }// 发送数据将数据放到发送缓冲区启动写事件监控void Send(const char *data, size_t len){// 外界传⼊的data可能是个临时的空间我们现在只是把发送操作压⼊了任务池有可能并没有被⽴即执⾏// 因此有可能执⾏的时候data指向的空间有可能已经被释放了Buffer buf;buf.WriteAndPush(data, len);// 右值引用 -- 提升效率_loop-RunInLoop(std::bind(Connection::SendInLoop, this, std::move(buf)));}// 连接建⽴就绪后进⾏channel回调设置 启动读监控void Established(){_loop-RunInLoop(std::bind(Connection::EstablishedInLoop, this));}// 提供给组件使⽤者的关闭接⼝--并不实际关闭需要判断有没有数据待处理void Shutdown(){_loop-RunInLoop(std::bind(Connection::ShutdownInLoop, this));}void Release(){_loop-QueueInLoop(std::bind(Connection::ReleaseInLoop, this));}// 启动⾮活跃销毁并定义多⻓时间⽆通信就是⾮活跃添加定时任务void EnableInactiveRelease(int sec){_loop-RunInLoop(std::bind(Connection::EnableInactiveReleaseInLoop, this, sec));}void CancelInactiveRelease(){_loop-RunInLoop(std::bind(Connection::CancelInactiveReleaseInLoop, this));}void Upgrade(const Any context,const ConnectedCallback conn,const MessageCallback msg,const ClosedCallback closed,const AnyEventCallback event){// 这是一个非线程安全的// 当底层拿到数据上层进行协议切换这个行为被压入队列之中// 如果此时遇到新的事件发生读取的数据仍然会按照切换协议之前的格式进行解读_loop-AssertInLoop();_loop-RunInLoop(std::bind(Connection::UpgradeInLoop, this, context, conn, msg, closed, event));}
};监听描述符管理Acceptor类实现:
// 获取连接管理
class Acceptor
{
private:// ⽤于创建监听套接字Socket _socket;// ⽤于对监听套接字进⾏事件监控EventLoop *_loop;// ⽤于对监听套接字进⾏事件管Channel _channel;// 读取新连接的回调函数 -- 这里的回调是创建Connectionusing AcceptCallback std::functionvoid(int);AcceptCallback _accept_callback;public:// 监听套接字的读事件回调处理函数---获取新连接调⽤_accept_callback函数进⾏新连接处理void HandlerRead(){int newfd _socket.Accept();if (newfd 0){return;}if (_accept_callback)_accept_callback(newfd);}int CreateServer(uint16_t port){bool ret _socket.CreateServer(port);assert(ret true);return _socket.get_fd();}public:Acceptor(EventLoop *loop, uint16_t port) : _socket(CreateServer(port)), _loop(loop),_channel(loop, _socket.get_fd()){_channel.SetReadCallback(std::bind(Acceptor::HandlerRead, this));}void SetAcceptCallback(const AcceptCallback cb) { _accept_callback cb; }// 启动监控void Listen() { _channel.EnableRead(); }
};服务器TcpServer类实现
class TcpServer
{
private:// 自增长iduint64_t _next_id;uint16_t _port;int _timeout; // 保存管理所有连接对应的shared_ptr对象bool _enable_inactive_release; // 是否启动了⾮活跃连接超时销毁的判断标志// 这是主线程的EventLoop对象负责监听事件的处理// 这是监听套接字的管理对象EventLoop _baseloop;Acceptor _acceptor;LoopThreadPool _pool; // 这是从属EventLoop线程池std::unordered_mapuint64_t, PtrConnection _conns; // 保存管理所有连接对应的shared_ptr对象// 回调函数using ConnectedCallback std::functionvoid(const PtrConnection );using ClosedCallback std::functionvoid(const PtrConnection );using MessageCallback std::functionvoid(const PtrConnection , Buffer *);using AnyEventCallback std::functionvoid(const PtrConnection );using Functor std::functionvoid();ConnectedCallback _connected_callback;ClosedCallback _closed_callback;MessageCallback _message_callback;AnyEventCallback _event_callback;private:// ⽤于添加⼀个定时任务void RunAfterInLoop(const Functor task, int delay){_next_id;_baseloop.TimerAdd(_next_id, delay, task);}// 为新连接构造⼀个Connection进⾏管理void NewConnection(int fd){_next_id;PtrConnection conn(new Connection(_pool.NextLoop(), _next_id, fd));conn-SetMessageCallback(_message_callback);conn-SetClosedCallback(_closed_callback);conn-SetConnectedCallback(_connected_callback);conn-SetAnyEventCallback(_event_callback);conn-SetSrvClosedCallback(std::bind(TcpServer::RemoveConnection, this, std::placeholders::_1));// 启动⾮活跃超时销毁if (_enable_inactive_release){conn-EnableInactiveRelease(_timeout);}// 启动监控conn-Established();_conns.insert(std::make_pair(_next_id, conn));}// 从管理Connection的_conns中移除连接信息 --- 需要bind到Connection内void RemoveConnectionInLoop(const PtrConnection conn){int id conn-get_id();auto iter _conns.find(id);if (iter ! _conns.end()){_conns.erase(iter);}}void RemoveConnection(const PtrConnection conn){_baseloop.RunInLoop(std::bind(TcpServer::RemoveConnectionInLoop, this, conn));}public:TcpServer(int port) : _port(port),_next_id(0),_enable_inactive_release(false),_acceptor(_baseloop, _port),_pool(_baseloop){_acceptor.SetAcceptCallback(std::bind(TcpServer::NewConnection, this, std::placeholders::_1));// 监听事件启动 将监听套接字挂到baseloop上_acceptor.Listen();}void SetThreadCount(int count) { return _pool.SetThreadCount(count); }void SetConnectedCallback(const ConnectedCallback cb) { _connected_callback cb; }void SetMessageCallback(const MessageCallback cb) { _message_callback cb; }void SetClosedCallback(const ClosedCallback cb) { _closed_callback cb; }void SetAnyEventCallback(const AnyEventCallback cb) { _event_callback cb; }void EnableInactiveRelease(int timeout){_timeout timeout;_enable_inactive_release true;}// ⽤于添加⼀个定时任务void RunAfter(const Functor task, int delay){_baseloop.RunInLoop(std::bind(TcpServer::RunAfterInLoop, this, task, delay));}void Start(){_pool.Create();_baseloop.Start();}
};
基于TcpServer实现回显服务器:
#pragma once
#include ../server.hppclass EchoServer
{
private:TcpServer _server;// 回调函数void OnConnected(const PtrConnection conn){DBG_LOG(NEW CONNECTION:%p, conn.get());}void OnClosed(const PtrConnection conn){DBG_LOG(CLOSE CONNECTION:%p, conn.get());}void OnMessage(const PtrConnection conn, Buffer *buf){// DBG_LOG(OnMessage:%p:%s, conn.get(), buf-Begin());conn-Send(buf-ReadPosition(), buf-ReadAbleSize());// Send没有支持任何read指针移动buf-MoveReadOffset(buf-ReadAbleSize());conn-Shutdown();}public:EchoServer(int port) : _server(port){_server.SetThreadCount(3);_server.EnableInactiveRelease(5);_server.SetConnectedCallback(std::bind(EchoServer::OnConnected, this, std::placeholders::_1));_server.SetMessageCallback(std::bind(EchoServer::OnMessage, this, std::placeholders::_1, std::placeholders::_2));_server.SetClosedCallback(std::bind(EchoServer::OnClosed, this, std::placeholders::_1));}void Start(){_server.Start();}
}; EchoServer关系图: EchoServer测试: 我们让客户端发送五条消息后不再进行通信经过timeout后服务端将该连接关闭。 HTTP协议模块:
Util工具类:
状态码描述字段:
std::unordered_mapint, std::string _statu_msg {{100, Continue},{101, Switching Protocol},{102, Processing},{103, Early Hints},{200, OK},{201, Created},{202, Accepted},{203, Non-Authoritative Information},{204, No Content},{205, Reset Content},{206, Partial Content},{207, Multi-Status},{208, Already Reported},{226, IM Used},{300, Multiple Choice},{301, Moved Permanently},{302, Found},{303, See Other},{304, Not Modified},{305, Use Proxy},{306, unused},{307, Temporary Redirect},{308, Permanent Redirect},{400, Bad Request},{401, Unauthorized},{402, Payment Required},{403, Forbidden},{404, Not Found},{405, Method Not Allowed},{406, Not Acceptable},{407, Proxy Authentication Required},{408, Request Timeout},{409, Conflict},{410, Gone},{411, Length Required},{412, Precondition Failed},{413, Payload Too Large},{414, URI Too Long},{415, Unsupported Media Type},{416, Range Not Satisfiable},{417, Expectation Failed},{418, Im a teapot},{421, Misdirected Request},{422, Unprocessable Entity},{423, Locked},{424, Failed Dependency},{425, Too Early},{426, Upgrade Required},{428, Precondition Required},{429, Too Many Requests},{431, Request Header Fields Too Large},{451, Unavailable For Legal Reasons},{501, Not Implemented},{502, Bad Gateway},{503, Service Unavailable},{504, Gateway Timeout},{505, HTTP Version Not Supported},{506, Variant Also Negotiates},{507, Insufficient Storage},{508, Loop Detected},{510, Not Extended},{511, Network Authentication Required}
}; 类型后缀字段:
std::unordered_mapstd::string, std::string _mime_msg {{.aac, audio/aac},{.abw, application/x-abiword},{.arc, application/x-freearc},{.avi, video/x-msvideo},{.azw, application/vnd.amazon.ebook},{.bin, application/octet-stream},{.bmp, image/bmp},{.bz, application/x-bzip},{.bz2, application/x-bzip2},{.csh, application/x-csh},{.css, text/css},{.csv, text/csv},{.doc, application/msword},{.docx, application/vnd.openxmlformats-officedocument.wordprocessingml.document},{.eot, application/vnd.ms-fontobject},{.epub, application/epubzip},{.gif, image/gif},{.htm, text/html},{.html, text/html},{.ico, image/vnd.microsoft.icon},{.ics, text/calendar},{.jar, application/java-archive},{.jpeg, image/jpeg},{.jpg, image/jpeg},{.js, text/javascript},{.json, application/json},{.jsonld, application/ldjson},{.mid, audio/midi},{.midi, audio/x-midi},{.mjs, text/javascript},{.mp3, audio/mpeg},{.mpeg, video/mpeg},{.mpkg, application/vnd.apple.installerxml},{.odp, application/vnd.oasis.opendocument.presentation},{.ods, application/vnd.oasis.opendocument.spreadsheet},{.odt, application/vnd.oasis.opendocument.text},{.oga, audio/ogg},{.ogv, video/ogg},{.ogx, application/ogg},{.otf, font/otf},{.png, image/png},{.pdf, application/pdf},{.ppt, application/vnd.ms-powerpoint},{.pptx, application/vnd.openxmlformats-officedocument.presentationml.presentation},{.rar, application/x-rar-compressed},{.rtf, application/rtf},{.sh, application/x-sh},{.svg, image/svgxml},{.swf, application/x-shockwave-flash},{.tar, application/x-tar},{.tif, image/tiff},{.tiff, image/tiff},{.ttf, font/ttf},{.txt, text/plain},{.vsd, application/vnd.visio},{.wav, audio/wav},{.weba, audio/webm},{.webm, video/webm},{.webp, image/webp},{.woff, font/woff},{.woff2, font/woff2},{.xhtml, application/xhtmlxml},{.xls, application/vnd.ms-excel},{.xlsx, application/vnd.openxmlformats-officedocument.spreadsheetml.sheet},{.xml, application/xml},{.xul, application/vnd.mozilla.xulxml},{.zip, application/zip},{.3gp, video/3gpp},{.3g2, video/3gpp2},{.7z, application/x-7z-compressed}
};
字符串切分: // 1.字符串分割函数static int Split(const std::string src, const std::string sep, std::vectorstd::string *arr){// 上一次sep位置size_t offset 0;while (offset src.size()){size_t pos src.find(sep,offset);if (pos std::string::npos){// 将剩余的部分当作⼀个字串放⼊arry中arr-push_back(src.substr(offset));return arr-size();}// 出现连续sep的情况// acbedif (pos offset){offset pos sep.size();continue;}arr-push_back(src.substr(offset, pos - offset));offset pos sep.size();}return arr-size();} 读取\写入文件内容:
static bool ReadFile(const std::string filename, std::string *buf)
{std::ifstream ifs(filename, std::ios::binary);if (ifs.is_open() false){std::cout open file: filename.c_str() faild... std::endl;return false;}size_t fsize 0;// 跳转读写位置到末尾ifs.seekg(0, ifs.end);// 获取当前读写位置相对于起始位置的偏移量fsize ifs.tellg();ifs.seekg(0,ifs.beg);buf-resize(fsize);ifs.read((*buf)[0],fsize);if(ifs.good() false){std::cout read file: filename.c_str() faild... std::endl;ifs.close();return false;}ifs.close();return false;
} // 向⽂件写⼊数据static bool WriteFile(const std::string filename, const std::string buf){std::ofstream ofs(filename, std::ios::binary | std::ios::trunc);if (ofs.is_open() false){std::cout open file: filename.c_str() faild... std::endl;return false;}ofs.write(buf.c_str(), buf.size());if (ofs.good() false){std::cout write file: filename.c_str() faild... std::endl;ofs.close();return false;}ofs.close();return true;} 响应码\文件类型Type
// 响应状态码的描述信息获取static std::string StatuDesc(int statu){uto it _statu_msg.find(statu);if (it ! _statu_msg.end()){return it-second;}return Unknow;}// 根据⽂件后缀名获取⽂件mimestatic std::string ExtMime(const std::string filename){// a.b.txt 先获取⽂件扩展名size_t pos filename.find_last_of(.);if (pos std::string::npos){return application/octet-stream;}std::string ext filename.substr(pos);auto it _mime_msg.find(ext);if (it _mime_msg.end()){// 默认返回二进制流return application/octet-stream;}return it-second;} 判断目录\普通文件: 当我们拿到这个文件信息获取st_mode可以调用这个宏函数来判断类别。 // 判断⼀个⽂件是否是⼀个⽬录static bool IsDirectory(const std::string filename){struct stat st;int ret stat(filename.c_str(), st);if (ret 0){return false;}return S_ISDIR(st.st_mode);}// 判断⼀个⽂件是否是⼀个普通⽂件static bool IsRegular(const std::string filename){struct stat st;int ret stat(filename.c_str(), st);if (ret 0){return false;}return S_ISREG(st.st_mode);} 有效路径判断: 访问服务器资源本质上就是访问服务器上存在的目录文件。当我们访问一台服务器的首页目录一般是这样/index.html前面的/仅仅是相对目录。这里的意思表达式客⼾端只能请求相对根⽬录中的资源其他地⽅的资源都不予理会。如果出现这样的不合理情况就应当制止这种不安全的行为。
static bool ValidPath(const std::string path){std::vectorstd::string subdir;Split(path, /, subdir);int level 0; // 当前层为0for (auto dir : subdir){// 访问上级目录if (dir ..){level--;if (level 0)return false;continue;}level;}return true;} 编码与解码:
如何理解编码解码 当我们打开百度首页网址值得我们注意的是这个页面的url是我们能明眼看懂的。紧接着我们在搜索栏输入“C”关键字再敲回车。 此时我们发现本应看到wdC反而被解析成了wdC%20%20。那么为什么需要这样做呢 为什么要进行URL编码
● 防止出现语法错误:比如空格、中文等这些字符也许会造成URL的语法错误。在URL编码之后这些特殊字符会被转码为相应的ASCII码避免了出现语法上的错误。
● 防止乱码:如果URL中含有一些非ASCII字符如中文等由于不同编码之间的差异可能会造成在浏览器端乱码的情况。在URL编码之后这些非ASCII字符会被转码为% %和十六进制编码确保浏览器能正常读取。
● 安全性:URL中含有敏感信息时如密码等需要进行URL编码以防止被网络攻击者截取和恶意篡改。 总的来说URL编码能够避免语法上的错误、防止乱码以及增强安全性保障了网络传输的正常进行和网络信息的安全传输。 RFC3986 协议对 URL 的编解码问题做出了详细的建议指出了哪些字符需要被编码才不会引起 URL 语义的转变以及对为什么这些字符需要编码做出了相应的解释。 编码格式将特殊字符的ascii值转换为两个16进制字符%HH,前缀% C - C%2B%2B 不编码的特殊字符 RFC3986⽂档规定: . - _ ~ 字⺟数字 属于绝对不编码字符。 W3C标准中规定查询字符串中的空格需要编码为 解码则是转空格 在线URL解码编码工具 // URL编码避免URL中资源路径与查询字符串中的特殊字符与HTTP请求中特殊字符产⽣歧义static std::string UrlEncode(const std::string url, bool convert_space_to_plus){std::string res;for (auto ch : url){if (ch . || ch _ || ch - || ch ~ || isalnum(ch)){// 绝不进行编码的字符res ch;continue;}// w3c转换if (ch convert_space_to_plus true){res ;continue;}// 需要进行编码字符// 格式为: %HHchar tmp[4] {0};// 格式化字符// %: %%// 十六进制: %X// 02:两个字符 左边补0snprintf(tmp, sizeof(tmp), %%%02X, ch);res tmp;}return res;}static char HEXTOI(char c){if (c 0 c 9){return c - 0;}else if (c a c z){return c - a 10;}else if (c A c Z){return c - A 10;}return -1;}static std::string UrlDecode(const std::string url, bool convert_plus_to_space){// 遇到了%则将紧随其后的2个字符转换为数字// 数字: 16进制 -- 10进制// :43(D) 2B(H)// 2*16 B*1: 3211 43// 让第一个数 左移4位 第二个数std::string res;for (int i 0; i url.size(); i){if (url[i] convert_plus_to_space true){res ;continue;}// 需要解码if (url[i] % (i 2) url.size()){char v1 HEXTOI(url[i 1]);char v2 HEXTOI(url[i 2]);char tmp v1 * 16 v2;res tmp;i 2;continue; // 这里的continue是跳转到判断位置 而不会进行}// 普通字符res url[i];}return res;} Http请求响应类:
// HttpRequest请求类实现
class HttpRequest
{
public:std::string _method; // 请求⽅法std::string _path; // 资源路径std::string _version; // 协议版本std::string _body; // 请求正⽂std::smatch _matches; // 资源路径的正则提取数据std::unordered_mapstd::string, std::string _headers; // 头部字段std::unordered_mapstd::string, std::string _params; // 查询字符串HttpRequest() : _version(DEFAULT_VERSION) {}void ReSet(){// 资源清理_method.clear();_path.clear();_version DEFAULT_VERSION;_body.clear();std::smatch match;_matches.swap(match);_headers.clear();_params.clear();}// 插⼊头部字段void SetHeader(const std::string key, const std::string val){_headers.insert(std::make_pair(key, val));}// 判断是否存在指定头部字段bool HasHeader(const std::string key) const{auto iter _headers.find(key);if (iter _headers.end()){return false;}return true;}// 获取指定头部字段的值std::string GetHeader(const std::string key) const{auto it _headers.find(key);if (it _headers.end()){return ;}return it-second;}void SetParam(const std::string key, const std::string val){_params.insert(std::make_pair(key, val));}// 判断是否有某个指定的查询字符串bool HasParam(const std::string key) const{auto it _params.find(key);if (it _params.end()){return false;}return true;}// 获取指定的查询字符串std::string GetParam(const std::string key) const{auto it _params.find(key);if (it _params.end()){return ;}return it-second;}// 获取正⽂⻓度size_t ContentLength() const{bool ret HasHeader(Content-Length);if (ret false){return 0;}std::string clen GetHeader(Content-Length);return std::stoi(clen);}// 判断是否是短链接bool Close() const{// 没有Connection字段或者有Connection但是值是close则都是短链接否则就是⻓连接if (HasHeader(Connection) true GetHeader(Connection) keep-alive){return false;}return true;}
};class HttpResponse
{
public:int _status_code; // 状态码bool _redirect_flag; // 重定向标志std::string _redirect_url; // 重定向urlstd::string _body; // 正文std::unordered_mapstd::string, std::string _headers;HttpResponse() : _redirect_flag(false), _status_code(200) {}HttpResponse(int status_code) : _redirect_flag(false), _status_code(status_code) {}void Reset(){_status_code 200;_redirect_flag false;_body.clear();_redirect_url.clear();_headers.clear();}// 插⼊头部字段void SetHeader(const std::string key, const std::string val){_headers.insert(std::make_pair(key, val));}bool HasHeader(const std::string key){auto it _headers.find(key);if (it _headers.end()){return false;}return true;}// 获取指定头部字段的值std::string GetHeader(const std::string key){auto it _headers.find(key);if (it _headers.end()){return ;}return it-second;}void SetContent(const std::string body, const std::string type text/html){_body body;SetHeader(Content-Type, type);}void SetRedirect(const std::string url, int status_code 302){_status_code status_code;_redirect_flag true;_redirect_url url;}// 判断是否是短链接bool Close(){// 没有Connection字段或者有Connection但是值是close则都是短链接否则就是⻓连接if (HasHeader(Connection) true GetHeader(Connection) keep-alive){return false;}return true;}
}; HttpContext上下⽂类实现: // 五种接收状态
// 可能出现数据 但报文不完整的情况
typedef enum
{RECV_HTTP_ERROR,RECV_HTTP_LINE,RECV_HTTP_HEAD,RECV_HTTP_BODY,RECV_HTTP_OVER
} HttpRecvStatu;// 800m
#define MAX_LINE 8192*1024*100
class HttpContext
{
private:int _status_code; // 响应状态码HttpRecvStatu _recv_status; // 当前接收及解析的阶段状态HttpRequest _request; // 已经解析得到的请求信息
private:// 解析请求行bool ParseHttpLine(const std::string line){std::smatch matches;std::regex reg((GET|HEAD|POST|PUT|DELETE) ([^?]*)(?:\\?(.*))?(HTTP/1\\.[01])(?:\n|\r\n)?);bool ret std::regex_match(line, matches, reg);if (ret false){_recv_status RECV_HTTP_ERROR;_status_code 400; // BAD REQUESTreturn false;}// 0 : GET /wwwroot/login?userxiaomingpass123123 HTTP/1.1// 1 : GET --- 请求方法// 2 : /wwwroot/login --- 资源路径// 3 : userxiaomingpass123123 --- 提交参数params// 4 : HTTP/1.1 --- 协议版本_request._method matches[1];std::transform(_request._method.begin(), _request._method.end(), _request._method.begin(), ::toupper);// 资源路径的获取需要进⾏URL解码操作但是不需要转空格_request._path Util::UrlDecode(matches[2], false);// 协议版本的获取_request._version matches[4];// 查询字符串的获取与处理// 3 : userxiaomingpass123123 --- 提交参数paramsstd::vectorstd::string query_string_arry;std::string query_string matches[3];Util::Split(query_string, , query_string_arry);for (auto str : query_string_arry){size_t pos str.find();// 协议切换if (pos std::string::npos){_recv_status RECV_HTTP_ERROR;_status_code 400; // BAD REQUESTreturn false;}std::string key Util::UrlDecode(str.substr(0, pos), true);std::string val Util::UrlDecode(str.substr(pos 1), true);_request.SetParam(key, val);}return true;}// 解析头部字段bool ParseHttpHead(std::string line){// key: val\r\nkey: val\r\n....if (line.back() \n)line.pop_back(); // 末尾是换⾏则去掉换⾏字符if (line.back() \r)line.pop_back(); // 末尾是回⻋则去掉回⻋字符size_t pos line.find(: );if (pos std::string::npos){_recv_status RECV_HTTP_ERROR;_status_code 400;return false;}std::string key line.substr(0, pos);std::string val line.substr(pos 2);_request.SetHeader(key, val);return true;}public:// 获取请求行bool RecvHttpLine(Buffer *buf){if (_recv_status ! RECV_HTTP_LINE)return false;// 1. 获取⼀⾏数据带有末尾的换⾏std::string line buf-GetOneLine();// 需要考虑的⼀些要素缓冲区中的数据不⾜⼀⾏ 获取的⼀⾏数据超⼤if (line.size() 0){// 缓冲区中的数据不⾜⼀⾏则需要判断缓冲区的可读数据⻓度如果很⻓了或者都不⾜⼀⾏这是有问题if (buf-ReadAbleSize() MAX_LINE){_recv_status RECV_HTTP_ERROR;_status_code 414; // URI TOO LONGreturn false;}// 缓冲区中数据不⾜⼀⾏但是也不多就等等新数据的到来return true;}if (line.size() MAX_LINE){_recv_status RECV_HTTP_ERROR;_status_code 414; // URI TOO LONGreturn false;}// 解析请求行bool ret ParseHttpLine(line);if (ret false){return false;}// ⾸⾏处理完毕进⼊头部获取阶段_recv_status RECV_HTTP_HEAD;return true;}// 获取头部字段bool RecvHttpHead(Buffer *buf){if (_recv_status ! RECV_HTTP_HEAD)return false;// ⼀⾏⼀⾏取出数据直到遇到空⾏为⽌ 头部的格式 key: val\r\nkey:val\r\n....while (1){// 需要考虑的⼀些要素缓冲区中的数据不⾜⼀⾏ 获取的⼀⾏数据超⼤std::string line buf-GetOneLine();if (line.size() 0){// 缓冲区中的数据不⾜⼀⾏则需要判断缓冲区的可读数据⻓度如果很⻓了都不⾜⼀⾏这是有问题的if (buf-ReadAbleSize() MAX_LINE){_recv_status RECV_HTTP_ERROR;_status_code 414; // URI TOO LONGreturn false;}}if (line.size() MAX_LINE){_recv_status RECV_HTTP_ERROR;_status_code 414; // URI TOO LONGreturn false;}if (line \n || line \r\n){break;}bool ret ParseHttpHead(line);if (ret false){return false;}}// 头部处理完毕进⼊正⽂获取阶段_recv_status RECV_HTTP_BODY;return true;}bool RecvHttpBody(Buffer *buf){if (_recv_status ! RECV_HTTP_BODY)return false;// 获取正⽂⻓度size_t content_length _request.ContentLength();DBG_LOG(RecvHttpBody content_length:%d,content_length);if (content_length 0){// 没有正⽂则请求接收解析完毕_recv_status RECV_HTTP_OVER;return true;}// 当前已经接收了多少正⽂,其实就是往 _request._body 中放了多少数据了// 实际还需要接收的正⽂⻓度size_t real_len content_length - _request._body.size();DBG_LOG(RecvHttpBody real_len:%d,real_len);DBG_LOG(RecvHttpBody _request._body:%d, _request._body.size());// 1.接收正⽂放到body中但是也要考虑当前缓冲区中的数据是否是全部的正⽂if (buf-ReadAbleSize() real_len){// 缓冲区中数据包含了当前请求的所有正⽂则取出所需的数据_request._body.append(buf-ReadPosition(), real_len);buf-MoveReadOffset(real_len);_recv_status RECV_HTTP_OVER;return true;}// 2.缓冲区中数据⽆法满⾜当前正⽂的需要数据不⾜取出数据然后等待新数据到来_request._body.append(buf-ReadPosition(), buf-ReadAbleSize());buf-MoveReadOffset(buf-ReadAbleSize());return true;}public:HttpContext() : _status_code(200), _recv_status(RECV_HTTP_LINE) {}void ReSet(){_status_code 200;_recv_status RECV_HTTP_LINE;_request.ReSet();}int get_status_code() { return _status_code; }HttpRecvStatu get_recv_status() { return _recv_status; }HttpRequest Request() { return _request; }// 接收并解析HTTP请求void RecvHttpRequest(Buffer *buf){// 不同的状态做不同的事情但是这⾥不要break 因为处理完请求⾏后应该⽴即处理头部⽽不是退出等新数据switch (_recv_status){case RECV_HTTP_LINE:RecvHttpLine(buf);case RECV_HTTP_HEAD:RecvHttpHead(buf);case RECV_HTTP_BODY:RecvHttpBody(buf);}return;}
}; HttpSever搭建:
class HttpServer
{
private:using Handler std::functionvoid(const HttpRequest , HttpResponse *);// 功能性请求:请求方法 映射 对应方法// 这里不能使用 unordered_map容器,因为regex没有重载// operator比较using Handlers std::vectorstd::pairstd::regex, Handler;Handlers _get_route;Handlers _post_route;Handlers _put_route;Handlers _delete_route;std::string _base_dir;TcpServer _server;private:// 访问静态资源bool IsFileHandler(const HttpRequest req){// 1. 必须设置了静态资源根⽬录if (_base_dir.empty()){return false;}// 2. 请求⽅法必须是GET / HEAD请求⽅法if (req._method ! GET req._method ! HEAD){return false;}// 3.请求的资源路径必须是⼀个合法路径if (Util::ValidPath(req._path) false){return false;}// 4.请求的资源必须存在,且是⼀个普通⽂件// 但如果访问的某个目录 那么这种情况下默认追加index.html// image/ -- image/index.html// 这是相对路径绝对路径需要带上base目录 ./wwwroot/image/a.png// 为了避免直接修改请求的资源路径因此定义⼀个临时对象std::string req_path _base_dir req._path;if (req._path.back() /){// 访问目录req_path index.html;}if (Util::IsRegular(req_path) false){// 不是普通文件return false;}return true;}// 静态资源的请求处理 --- 将静态资源⽂件的数据读取出来放到rsp的_body中, 并设置mimevoid FileHandler(const HttpRequest req, HttpResponse *resp){std::string req_path _base_dir req._path;if (req._path.back() /){req_path index.html;}// 读取文件bool ret Util::ReadFile(req_path, resp-_body);if (ret false){return;}// 文件类型std::string mime Util::ExtMime(req_path);resp-SetHeader(Content-Type, mime);}private:// 错误处理void ErrorHandler(const HttpRequest req, HttpResponse *resp){// 返回错误页面std::string body;Util::ReadFile(ERROR_HTML, body);resp-SetContent(body);}// 将HttpResponse中的要素按照http协议格式进⾏组织、发送void WriteReponse(const PtrConnection conn, const HttpRequest req, HttpResponse resp){// 1. 先完善头部字段if (req.Close() true){resp.SetHeader(Connection, close);}else{resp.SetHeader(Connection, keep-alive);}if (resp._body.empty() false resp.HasHeader(Content-Length) false){resp.SetHeader(Content-Length, std::to_string(resp._body.size()));}if (resp._redirect_flag true){resp.SetHeader(Location, resp._redirect_url);}// 将resp中的要素按照http协议格式进⾏组织std::stringstream resp_line;resp_line req._version std::to_string(resp._status_code) Util::StatuDesc(resp._status_code) \r\n;// 组织头部字段for (auto header : resp._headers){resp_line header.first : header.second \r\n;}// 空行resp_line \r\n;resp_line resp._body;// 数据发送conn-Send(resp_line.str().c_str(), resp_line.str().size());}// 功能性请求的分类处理void Dispatcher(HttpRequest req, HttpResponse *resp, Handlers handlers){// 在对应请求⽅法的路由表(handlers)中// 查找是否含有对应资源请求的处理函数有则调⽤没有则发送404for (auto handler : handlers){const std::regex reg(handler.first);const Handler functor handler.second;req._path.pop_back();bool ret std::regex_match(req._path, req._matches, reg);if (ret false){DBG_LOG(资源路径:%s, req._path.c_str());continue;}return functor(req, resp); // 传⼊请求信息和空的rsp执⾏处理函数}DBG_LOG(404:%s, req._path.c_str());resp-_status_code 404;}void Route(HttpRequest req, HttpResponse *resp){// 对请求进⾏分辨是⼀个静态资源请求还是⼀个功能性请求:// 1.静态资源请求则进⾏静态资源的处理// 2.功能性请求则需要通过⼏个请求路由表来确定是否有处理函数// 3.既不是静态资源请求也没有设置对应的功能性请求处理函数就返回405if (IsFileHandler(req) true){FileHandler(req, resp);}if (req._method GET || req._method HEAD){return Dispatcher(req, resp, _get_route);}else if (req._method POST){return Dispatcher(req, resp, _post_route);}else if (req._method PUT){return Dispatcher(req, resp, _put_route);}else if (req._method DELETE){return Dispatcher(req, resp, _delete_route);}resp-_status_code 405;}public:// 设置上下⽂void OnConnected(const PtrConnection conn){conn-SetContext(HttpContext());DBG_LOG(NEW CONNECTION %p, conn.get());}// 缓冲区数据解析处理void OnMessage(const PtrConnection conn, Buffer *buffer){while (buffer-ReadAbleSize() 0){// 1. 获取上下⽂HttpContext *context conn-GetContext()-getHttpContext();// 通过上下⽂对缓冲区数据进⾏解析得到HttpRequest对象// 1. 如果缓冲区的数据解析出错就直接回复出错响应// 2. 如果解析正常且请求已经获取完毕才开始去进⾏处理context-RecvHttpRequest(buffer);HttpRequest req context-Request();HttpResponse resp(context-get_status_code());if (context-get_status_code() 400){// 进⾏错误响应并关闭连接// 填充⼀个错误显⽰⻚⾯数据到rsp中ErrorHandler(req, resp);// 组织响应发送给客⼾端WriteReponse(conn, req, resp);context-ReSet();buffer-MoveReadOffset(buffer-ReadAbleSize()); // 出错了就把缓冲区数据清空conn-Shutdown(); // 关闭连接return;}if (context-get_recv_status() ! RECV_HTTP_OVER){// 当前请求还没有接收完整,则退出等新数据到来再重新继续处理return;}// 3. 请求路由 业务处理Route(req, resp);// 4. 对HttpResponse进⾏组织发送WriteReponse(conn, req, resp);// 5. 重置上下⽂context-ReSet();// 6. 根据⻓短连接判断是否关闭连接或者继续处理if (resp.Close() true){conn-Shutdown(); // 短链接则直接关闭}}return;}public:HttpServer(int port, int timeout DEFALT_TIMEOUT) : _server(port){_server.EnableInactiveRelease(timeout);_server.SetConnectedCallback(std::bind(HttpServer::OnConnected, this, std::placeholders::_1));_server.SetMessageCallback(std::bind(HttpServer::OnMessage, this, std::placeholders::_1, std::placeholders::_2));}bool SetBaseDir(const std::string path WWWROOT){assert(Util::IsDirectory(path) true);_base_dir path;return false;}// 设置/添加请求请求的正则表达与处理函数的映射关系void Get(const std::string pattern, const Handler handler){_get_route.push_back(std::make_pair(std::regex(pattern), handler));}void Post(const std::string pattern, const Handler handler){_post_route.push_back(std::make_pair(std::regex(pattern), handler));}void Put(const std::string pattern, const Handler handler){_put_route.push_back(std::make_pair(std::regex(pattern), handler));}void Delete(const std::string pattern, const Handler handler){_delete_route.push_back(std::make_pair(std::regex(pattern), handler));}void SetThreadCount(int count){_server.SetThreadCount(count);}void Listen(){DBG_LOG(HttpServer Start...);_server.Start();}
}; 测试模块:
通信功能测试
(1) 使⽤Postman进⾏基本功能测试
● 设置功能性函数 ● 功能性函数作用: 将请求报头封装后回显给请求端 我们将服务器启动使用Postman模拟浏览器向该服务器发送GET请求方法。 (2) ⻓连接连续请求测试
● ⻓连接测试1创建⼀个客⼾端持续给服务器发送数据直到超过超时时间看看是否正常 不超过timeout时间内长连接正常进行通信一旦超过timeout服务器会自动释放该连接。 (3) 超时连接释放
● 超时连接释放测试1: 连接服务器告诉服务器要发送100字节正⽂数据给服务器但是实际上发送数据不⾜100字节然后看服务器处理情况。
一次不够的请求: 多发请求: ① 如果数据只发送⼀次服务器将得不到完整请求就不会进⾏业务处理客⼾端也就得不到响应最终超时关闭。 ②连着给服务器发送了多次 ⼩的请求服务器会将后边的请求当作前边请求的正⽂进⾏处理⽽后便处理的时候有可能就会因为处理错误⽽关闭连接。 ● 超时连接释放测试2: 接收请求的数据但是业务处理的时间过⻓超过了设置的超时销毁时间(服务器性能达到瓶颈)观察服务端的处理。 预期结果在⼀次业务处理中耗费太⻓时间导致其他连接被连累超时导致其他的连接有可能会超时释放。
例如: 假设有12345描述符就绪了在处理1的时候花费了30s处理完超时了导致2345描述符因为⻓时间没有刷新活跃度则存在两种可能处理结果:
① 如果接下来的2345描述符都是通信连接描述符恰好本次也都就绪了事件则并不影响等1处 理完了接下来就会进⾏处理并刷新活跃度。
② 如果接下来的2号描述符是定时器事件描述符定时器触发超时执⾏定时任务就会将345描述符给释放掉这时候⼀旦345描述符对应的连接被释放接下来在处理345事件的时候就会导致程序崩溃。 因此在任意的事件处理中都不应该直接对连接进⾏释放⽽应该将释放操作压⼊到任务池中等所有连接事件处理完了然后执⾏任务池中的任务的时候再去进⾏释放。 连接被销毁了则后续的操作也就⽆法进⾏了。因此剩下的连接并没有得到正常的响应⽽是连接被关闭。
(4) 数据中多条请求处理测试 给服务器发送的⼀条数据中包含有多个HTTP请求观察服务器的处理。 (5) PUT⼤⽂件上传测试 使⽤put请求上传⼀个⼤⽂件进⾏保存⼤⽂件数据的接收会被分在多次请求中接收然后计算源⽂件和上传后保存的⽂件的MD5值判断请求的接收处理是否存在问题。(这⾥主要观察的是上下⽂的处理过程是否正常)。
什么是MD5值 MD5是一种信息摘要算法一种被广泛使用的密码散列函数可以产生出一个128位16字节的散列值用于确保信息输出的完整一致。 ● 创建大文件
dd if/dev/zero of./hello.txt bs1M count300 ● 上传大文件 性能测试 采⽤webbench进⾏服务器性能测试Webbench是知名的⽹站压⼒测试⼯具它是由Lionbridge(点我)公司开发的。webbench测试原理是创建指定数量的进程在每个进程中不断创建套接字向服务器发送请求并通过管道最终将每个进程的结果返回给主进程进⾏数据统计。 webbench的标准测试可以向我们展⽰服务器的两项内容: 每秒钟相应请求数 和 “每秒钟传输数据量”。
性能测试的两个重点衡量标准吞吐量QPS。
● QPS: 每秒钟处理的包的数量
● 吞吐量: 单位时间内成功地传送数据的数量。 安装webbench 点击网址可以选择clone或者下载zip。 clone到我们的服务器上进入该目录make进行安装。 我们直接使用./webbench会提示该软件的使用手册。 测试环境: 服务器:2核4G带宽1M,服务器程序采⽤1主3从reactor模式。 webbench服务器:Centos7 4核4G 将程序运⾏起来根据进程ID在/proc⽬录下查看程序中的各项限制信息能够看到当前⽤⼾的进程的最⼤数量为4096因为后边需要模拟上万并发量需要创建上万个进程。 修改配置⽂件/etc/security/limits.conf在末尾添加内容nofile是修改可打开⽂件数,nproc是修改进程数。 修改配置⽂件/etc/security/limits.d/20-nproc.conf。 重启机器修改成功: 测试15000个客⼾端连接的情况下测试结果 [WGzZshecs-75008 http]$./webbench -c 5000 -t 60 http://127.0.0.1:8500/hello Webbench - Simple Web Benchmark 1.5 Copyright (c) Radim Kolar 1997-2004, GPL Open Source Software. Request: GET / HTTP/1.0 User-Agent: WebBench 1.5 Host: 127.0.0.1 Runing info: 5000 clients, running 60 sec. Speed384577 pages/min, 2589020 bytes/sec. Requests: 384577 susceed, 0 failed. 本篇到此结束感谢你的阅读。
祝你好运向阳而生~
