课程网站建设课程,手机自建网站平台,网站建设 软文发布,做网站用小型机或服务器One Thread One Loop主从Reactor模型⾼并发服务器 文章目录 One Thread One Loop主从Reactor模型⾼并发服务器一些补充HTTP服务器Reactor 模型eventfd通用类Any 目标功能模块划分#xff1a;SERVER模块Buffer模块#xff1a;编写思路#xff1a;接口设计#xff1a;具体实现…One Thread One Loop主从Reactor模型⾼并发服务器 文章目录 One Thread One Loop主从Reactor模型⾼并发服务器一些补充HTTP服务器Reactor 模型eventfd通用类Any 目标功能模块划分SERVER模块Buffer模块编写思路接口设计具体实现 日志宏模块编写思路具体实现 Socket模块编写思路接口设计具体实现 Channel模块编写思路接口设计具体实现 Connection模块编写思路接口设计具体实现 Acceptor模块编写思路接口设计具体实现 TimerQueue模块编写思路接口设计具体实现 Poller模块编写思路接口设计具体实现 EventLoop模块编写思路接口设计具体实现 TcpServer模块编写思路接口设计具体实现 协议模块Util模块具体实现 HttpRequest模块具体实现 HttpResponse模块具体实现 请求接收上下文模块具体实现 HttpServer模块具体实现 性能测试测试环境测试结果 One Thread One Loop的思想就是把所有的操作都放到⼀个线程中进行⼀个线程对应⼀个事件处理的循环。 一些补充
HTTP服务器
HTTPHyper Text Transfer Protocol超⽂本传输协议是应⽤层协议是⼀种简单的请求-响应协议客⼾端根据⾃⼰的需要向服务器发送请求服务器针对请求提供服务完毕后通信结束。
统一监听事件收到事件后分发给处理进程或线程。
Reactor 模式的优点包括
高性能通过异步处理事件可以充分利用系统资源提高性能。可扩展性可以轻松添加新的事件类型和处理程序使应用程序更加灵活和可扩展。响应性能够快速响应事件适用于需要实时性的应用程序如网络服务器。
Reactor模式一般可以分为三类 单Reactor单线程单I/O多路复⽤业务处理 即在单个线程中进行事件监控和处理。 单Reactor多线程单I/O多路复⽤线程池业务处理 多Reactor多线程多I/O多路复⽤线程池业务处理
基于单Reactor多线程的缺点改进我们选择让主Reactor线程单独进行新连接事件监控从属Reactor线程进行IO事件监控从属Reactor线程再让业务线程池进行业务处理。 注意点
执行流并非越多越好执行流太多反而增加了CPU切换调度的成本
eventfd eventfd是一种事件通知机制。创建一个描述符用于实现事件通知。 eventfd本质在内核中管理的是一个计数器。创建eventfd就会在内核中创建一个计数器结构向eventfd中写入一个数值用于表示事件通知次数。可以用read进行数据的读取读取到的数据就是通知的次数。如每次给eventfd中写入个1就表示通知1次连续三次后read读取的数字就是3读取后计数清零。
用处
在EventLoop模块中实现线程间的事件通知功能
#include sys/eventfd.h
int eventfd(unsigned int inital,int flags);
功能创建一个eventfd对象实现事件通知
参数initial: 计数初值flagsEFD_CLOEXEC 禁止进程复制EFD_NONBLOCK 启动非阻塞属性
返回值返回一个文件描述符用于操作
eventfd也是通过read/write/close等进行操作的。 注意点read和write进行IO的时候数据只能是一个8字节数据。
通用类Any
可以直接用库里自带的这里用的是自己实现的
#pragma once
#includeiostream
#includetypeinfo
#includecassert
#includeunistd.hclass Any
{
private:class holder{public:virtual ~holder() {}virtual const std::type_info type() 0; virtual holder *clone() 0;};template class Tclass placeholder : public holder{public:placeholder(const T val):_val(val) {};//获取子类对象保存的数据类型virtual const std::type_info type(){return typeid(T);}//克隆一个新的子类对象virtual holder *clone(){return new placeholder(_val);}public:T _val;};holder* _content;public:Any():_content(nullptr) {}templateclass TAny(const T val):_content(new placeholderT(val)){}Any(const Any other):_content(other._content ? other._content-clone() : nullptr){} //拷贝构造的对象为空就置为空非空调用拷贝对象的clone~Any(){delete _content;}Any swap(Any other){std::swap(_content,other._content);return *this;}templateclass TT* get() //返回子类对象保存数据的指针{//获取类型与保存类型必须一致assert(typeid(T) _content-type());return ((placeholderT*)_content)-_val; //考虑好优先级 基类指针强转为派生类才能访问派生类对象中存储的数据}templateclass TAny operator(const T val) //赋值运算符重载函数{//类似于现代写法Any(val).swap(*this); return *this; //返回引用 为了能连续赋值}Any operator(const Any other){Any(other).swap(*this);return *this;}};目标
我们要实现的服务器本⾝并不存在业务咱们要实现的应该算是⼀个⾼性能服务器基础库是⼀个基础组件。
采用多Reactor多线程模式也叫主从Reactor模型
**为了方便实现将业务线程部分去掉了直接交给从属Reactor线程进行处理即One Thread One Event Loop.**一个线程对应了一个循环IO事件监控IO操作业务处理 功能模块划分
基于以上的理解我们要实现的是⼀个带有协议⽀持的Reactor模型⾼性能服务器因此将整个项⽬的实现划分为两个⼤的模块
• SERVER模块实现Reactor模型的TCP服务器
• 协议模块对当前的Reactor模型服务器提供应⽤层协议⽀持
SERVER模块
SERVER模块就是对所有的连接以及线程进⾏管理让它们各司其职在合适的时候做合适的事最终完成⾼性能服务器组件的实现。
Buffer模块 编写思路
1.实现缓冲区得有一块内存空间。采用vector容器进行空间管理为何不用string因为string操作中遇到‘/0’就停止
2.要素
默认的空间大小当前的读取数据位置当前的写入数据位置
以类似于循环数组的方式存储数据当剩余空间不足时再选择扩容 写入数据
从当前写入位置指向的位置开始写入。看剩余空间是否足够包含读偏移之前的空闲空间
足够数据都前移到起始位置不够从当前位置扩容足够大小
写入成功后写偏移向后移动
读取数据
有数据可读时从读偏移指向位置读取。
可读数据大小 写偏移 - 读偏移
接口设计
//接口设计
class Buffer{private:std::vectorchar _buffer;uint64_t _read_index;uint64_t _write_idx;public://1.获取当前写位置地址//2.确保可写空间足够 (足够移动 不足扩容)//3.获取读偏移前空闲空间大小//4.获取写偏移后空闲空间大小//5.写位置向后移动指定长度//6.读位置向后移动指定长度//7.获取读位置地址//8.获取可读空间大小//9.清理功能 - 读写位置归0
};具体实现
class Buffer //自己实现时删除冗余部分
{
private:std::vectorchar _buffer; // 使用vector进行内存空间管理uint64_t _read_idx; // 读偏移uint64_t _writer_idx; // 写偏移public:Buffer() : _read_idx(0), _writer_idx(0), _buffer(BUFFER_DEFAULT_SIZE) {}char *Begin(){return *_buffer.begin();}// 1.获取当前写入起始地址char *WritePosition(){// buffer的空间起始地址加上写偏移量return Begin() _writer_idx;}// 2.获取当前读取起始地址char *ReadPosition(){return Begin() _read_idx;}// 3.获取空闲空间大小 前沿空间大小 后延空间大小uint64_t TailIdleSize(){// 总体空间大小减去写偏移return _buffer.size() - _writer_idx;}uint64_t HeadIdleSize(){return _read_idx;} // 4.获取可读数据大小uint64_t ReadAbleSize(){return _writer_idx - _read_idx;}// 5.将读偏移向后移动void MoveReadOffset(uint64_t len){// 向后移动大小必须小于可读数据大小if(0 len){return;}assert(len ReadAbleSize());_read_idx len;}// 6.将写偏移向后移动void MoveWriteOffset(uint64_t len){assert(len TailIdleSize()); // 写入前需要确保后延空间足够_writer_idx len;}// 7.确保可写空间足够 数据挪动 还是 扩容void EnsureWriteSpace(uint64_t len){// 1.末尾空闲空间足够 就返回if (TailIdleSize() len){return;}// 末尾空间不够 加起始空间大小够就挪动 不够就扩容if (len TailIdleSize() HeadIdleSize()){// 挪动数据uint64_t rsz ReadAbleSize(); // 获取当前数据大小std::copy(ReadPosition(), ReadPosition() rsz, Begin()); // 可读数据拷贝到了起始位置_read_idx 0; // 读偏移归零_writer_idx rsz; // 写偏移置为可读数据大小即写偏移量}else // 空间不够 需扩容{// 不移动数据直接给写偏移之后扩容足够空间即可_buffer.resize(_writer_idx len);}}// 写入数据void Write(const void *data, uint64_t len){if(0 len){return;}// 1.保证空间足够 2.拷贝数据EnsureWriteSpace(len);const char *d (const char *)data;std::copy(d, d len, WritePosition());}void WriteAndPush(const void *data, uint64_t len){Write(data, len);MoveWriteOffset(len);}void WriteString(const std::string data){Write(data.c_str(), data.size());}void WriteStringAndPush(const std::string data){WriteString(data);MoveWriteOffset(data.size());}void WriteBuffer(Buffer data){Write(data.ReadPosition(), data.ReadAbleSize());}void WriteBufferAndPush(Buffer data){WriteBuffer(data);MoveWriteOffset(data.ReadAbleSize());}// 读取数据void Read(void *buf, uint64_t len){// 要求获取数据大小小于可读数据大小assert(len ReadAbleSize());std::copy(ReadPosition(), ReadPosition() len, (char *)buf);}void ReadAndPop(void *buf, uint64_t len) // 读数据并且弹出已读数据{Read(buf, len);MoveReadOffset(len);}std::string ReadAsString(uint64_t len){assert(len ReadAbleSize());std::string str;str.resize(len);Read(str[0], len);return str;}std::string ReadAsStringAndPop(uint64_t len){assert(len ReadAbleSize());std::string str ReadAsString(len);MoveReadOffset(len);return str;}char *FindCRLF(){void *res memchr(ReadPosition(), \n, ReadAbleSize()); // 在空间中找对应字符return (char *)res;}std::string GetLine(){char *pos FindCRLF();if (pos nullptr){return ;}return ReadAsString(pos - ReadPosition() 1); //1是为了取出换行字符}std::string GetLineAndPop(){std::string str GetLine();MoveReadOffset(str.size());return str;}// 清空缓冲区void Clear() // 将偏移量归零即可{_read_idx 0;_writer_idx 0;}
};日志宏模块
编写思路
为了方便调试我编写了一个简单的打印日志的宏函数 ,首先设置了三个日志等级INF、DBG、ERR,我们可以通过更改打印的日志等级来实现日志是否需要打印所以我们额外分别设置这三个等级的日志接口其实就是在我们基本的日志宏函数传了个等级参数罢了。
具体实现
#pragma once
#includectime
#includecstdio
#define INF 0
#define DBG 1
#define ERR 2
#define LOG_LEVEL DBG
#define LOG(level,format,...) do{\if(level LOG_LEVEL) break;\time_t t time(NULL);\struct tm* ptm localtime(t);\ char tmp[32] {0};\strftime(tmp,31,%H:%M:%S,ptm);\fprintf(stdout,[%s %s:%d] format \n,tmp,__FILE__,__LINE__,##__VA_ARGS__);\
}while(0);
#define INFLOG(format,...) LOG(INF,format,##__VA_ARGS__);
#define DBGLOG(format,...) LOG(DBG,format,##__VA_ARGS__);
#define ERRLOG(format,...) LOG(ERR,format,##__VA_ARGS__);Socket模块 编写思路
常规的套接字接口如上图所示的10个功能需要进行实现。除此之外我们额外需要实现两个接口
开启地址端口重用 — 防止服务器出问题后无法立即重启设置套接字为非阻塞属性 — ET模式的要求我们需要一次性读完需要的数据即读到没有数据为止如果是阻塞属性会导致阻塞因此我们需要设置非阻塞属性 。
接口设计
class Socket{private: int _sockfd;public:Socket(); //构造 ~Scoket(); //析构//1.创建套接字//2.绑定地址信息//3.开始监听//4.向服务器发起连接//5.获取新连接//6.接收数据//7.发送数据//8.关闭套接字//9.创建一个服务端连接//10.创建一个客户端连接//11.开启地址端口重用//12.设置套接字为非阻塞属性
};具体实现
#define MAX_LISTEN 1024
class Socket
{
private:int _sockfd;
public:Socket() : _sockfd(-1) {}Socket(int fd) : _sockfd(fd) {}~Socket(){Close();}int FD(){return _sockfd;}// 创建套接字bool Create(){// int socket(int domain, int type, int protocol);_sockfd socket(AF_INET, SOCK_STREAM, 0);if (_sockfd 0){ERR_LOG(CREATE SOCKET FAILED);return false;}return true;}// 绑定地址信息bool Bind(const std::string ip, uint16_t port){// int bind(int sockfd, const struct sockaddr *addr, socklen_t addrlen);struct sockaddr_in addr;addr.sin_family AF_INET;addr.sin_port htons(port);addr.sin_addr.s_addr inet_addr(ip.c_str());socklen_t len sizeof(addr);int ret bind(_sockfd, (struct sockaddr *)addr, len);if (ret 0){ERR_LOG(BIND FALIED);return false;}return true;}// 开始监听bool Listen(int backlog MAX_LISTEN){// int listen(int sockfd, int backlog);int ret listen(_sockfd, backlog);if (ret 0){ERR_LOG(LISTEN FALIED);return false;}return true;}// 向服务器发起连接bool Connect(const std::string ip, uint16_t port){// int connect(int sockfd, const struct sockaddr *addr,socklen_t addrlen);struct sockaddr_in addr;addr.sin_family AF_INET;addr.sin_port htons(port);addr.sin_addr.s_addr inet_addr(ip.c_str());socklen_t len sizeof(addr);int ret connect(_sockfd, (struct sockaddr *)addr, len);if (ret 0){ERR_LOG(CONNECT FALIED);return false;}return true;}// 获取新连接int Accept(){// int accept(int sockfd, struct sockaddr *addr, socklen_t *addrlen);int newfd accept(_sockfd, NULL, NULL);if (newfd 0){ERR_LOG(ACCEPT FAILED);return -1;}return newfd;}// 接收数据ssize_t Recv(void *buf, size_t len, int flag 0){// ssize_t recv(int sockfd, void *buf, size_t len, int flags);ssize_t ret recv(_sockfd, buf, len, flag);if (ret 0){ // EINTR 表示当前socket阻塞等待被信号打断了if (errno EAGAIN || errno EINTR){ // EAGAIN 当前socket的接收缓冲区没数据非阻塞情况下才有该错误return 0;}return -1;}return ret; // 实际接收数据长度}// 非阻塞接收数据ssize_t NonBlockRecv(void *buf, size_t len){return Recv(buf, len, MSG_DONTWAIT); // MSG_DONTWAIT 表示当前接收为非阻塞}// 发送数据ssize_t Send(const void *buf, size_t len, int flag 0){// ssize_t send(int sockfd, const void *buf, size_t len, int flags);ssize_t ret send(_sockfd, buf, len, flag);if (ret 0){ERR_LOG(SEND FAILED);return -1;}return ret; // 实际发送的数据长度}ssize_t NonBlockSend(void *buf, size_t len){return Send(buf, len, MSG_DONTWAIT); // MSG_DONTWAIT 表示当前接收为非阻塞}// 关闭套接字bool Close(){if(_sockfd ! -1){close(_sockfd);}_sockfd -1;}// 创建一个服务器连接bool CreateServer(uint16_t port,bool flag false){//1.创建套接字 //2.绑定地址 //3.开始监听 //4,设置非阻塞 //5.启动地址重用if(Create() false) return false;if(flag) NonBlock();if(Bind(0.0.0.0,port) false) return false;if(Listen() false) return false;ReuseAddress();return true;}// 创建一个客户端连接bool CreateClient(const std::string ip,uint16_t port){//1,创建套接字//2.连接服务器if(Create() false) return false;if(Connect(ip,port) false) return false;return true;}// 设置套接字选项---开启地址端口重用void ReuseAddress(){// int setsockopt(int sockfd, int level, int optname,const void *optval, socklen_t optlen);int val 1;setsockopt(_sockfd,SOL_SOCKET,SO_REUSEADDR,(void*)val,sizeof(int));val 1;setsockopt(_sockfd,SOL_SOCKET,SO_REUSEPORT,(void*)val,sizeof(int));}// 设置套接字阻塞属性void NonBlock(){//int fcntl(int fd, int cmd, ... /* arg */ );int flag fcntl(_sockfd,F_GETFL,0);fcntl(_sockfd,F_SETFL,flag|O_NONBLOCK);}
};Channel模块 编写思路
对一个描述符进行事件管理因为我们是用epoll进行事件监控所以我们操作就是uint32_t event标志位 如EPOLLIN可读、EPOLLOUT可写、EPOLLRDHUP连接断开、EPOLLHUP挂断、EPOLLERR错误、EPOLLPRI优先数据 等 对事件触发后如何处理进行管理 需要处理的事件可读、可写、挂断、错误、任意有五种事件需要处理需要五个回调函数
接口设计
class Channel{private:uint32_t _events; //当前需要监控的事件uint32_t _revents; //当前实际触发的事件using EventCallback std::functionvoid( );EventCallback _read_callback; //可读事件回调函数EventCallback _write_callback; //可写事件回调函数EventCallback _error_callback; //错误事件回调函数EventCallback _close_callback; //连接断开回调函数EventCallback _event_callback; //任意事件回调函数public://事件管理//1.描述符是否可读//2.描述符是否可写//3.对描述符监控可读//4.对描述符监控可写//5.解除读事件监控//6.解除写事件监控//7.解除所有监控//8.移除监控将描述符信息从epoll中移除//事件触发后的处理管理//1.设置各种事件的回调//2.整合的一个事件处理函数触发了事件就调用该函数根据_revents来判断调用对应的处理函数
};
具体实现
// channel模块
// 对一个描述符进行IO事件管理 实现对描述符可读、可写等的操作
// Poller模块对描述符进行IO 事件监控就绪后回调不同的处理函数功能
class Poller;
class EventLoop;
class Channel
{
private:int _fd;uint32_t _event; // 当前需要监控的事件uint32_t _events; // 当前连续触发的事件EventLoop *_loop;using EventCallBack std::functionvoid();EventCallBack _read_callback; // 读事件触发回调EventCallBack _write_callback; // 写事件触发回调EventCallBack _error_callback; // 错误事件触发回调EventCallBack _close_callback; // 断开事件触发回调EventCallBack _event_callback; // 任意事件触发回调
public:Channel(EventLoop *loop, int fd) : _loop(loop), _fd(fd), _event(0), _events(0){};int Fd(){return _fd;}uint32_t Event(){return _event;}// 设置实际就绪实际void SetEvents(uint32_t _event){_events _event;}// 1.是否可读bool IsRead(){return _event EPOLLIN;}// 2.是否可写bool IsWrite(){return _event EPOLLOUT;}// 3.使 可读void EnableRead(){_event | EPOLLIN;UpDate();}// 4.使 可写void EnableWrite(){_event | EPOLLOUT;UpDate();}// 5.取消可读void DisableRead(){_event ~EPOLLIN;UpDate();}// 6.取消可写void DisableWrite(){_event ~EPOLLOUT;UpDate();}// 7.取消所有监控void DisableAll(){_event 0;UpDate();}// 8.设置各个回调函数void SetWriteCallBack(const EventCallBack cb){_write_callback cb;}void SetReadCallBack(const EventCallBack cb){_read_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;}// 9.事件处理函数void HandleEvent(){if (_event_callback){_event_callback();}if ((_events EPOLLIN) || (_events EPOLLRDHUP) || (_events EPOLLPRI)){ // 对方关闭连接也触发可读事件if (_read_callback){_read_callback();}}if (_events EPOLLOUT){ if (_write_callback){_write_callback();}} // 有可能释放连接的操作事件一次只处理一个else if (_events EPOLLERR){if (_error_callback){_error_callback();}}else if (_events EPOLLHUP) { if (_close_callback){_close_callback();}}}void UpDate();void ReMove();// void UpDate(){// return _poller-UpDateEvent(this);// }// //移除监控// void ReMove(){// return _poller-ReMoveEvent(this);// }
};
Connection模块 编写思路
对连接进行全方位的管理。如
套接字的管理 进行套接字的操作连接事件的管理 可读、可写、错误、挂断、任意缓冲区的管理便于socket数据的接收和发送。协议上下文的管理请求数据的处理回调函数的管理
接口设计
enum ConnStatu{DISCONNECTED, //连接关闭状态CONNECTING, // 连接建立成功 - 待处理状态CONNECTED, //连接建立完成各设置已完成 可以通信DISCONNCETING //待关闭状态
};
class Connection{private:uint64_t _conn_id; //唯一标识 连接idConnStatu _statu; //连接状态bool _enable_inactive_release; //连接是否启动非活跃销毁的判断标志EventLoop* _loop; //连接所关联的一个Eventloopint _sockfd; //连接关联的描述符Socket _sock; //套接字操作管理Channel _channel; //连接事件管理Buffer _in_buffer; //输入缓冲区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 _close_callback;AnyEventCallback _event_callback;//组件内的连接关闭回调ClosedCallback _server_closed_callback;public://构造//析构//1.发送数据//2.提供给组件使用者的关闭接口//3.启动非活跃连接销毁//4.取消非活跃连接销毁//5.协议切换 --- 重置上下文和阶段性处理函数 具体实现
enum ConnStatu{DISCONNECTED, //连接关闭状态CONNECTING, // 连接建立成功 - 待处理状态CONNECTED, //连接建立完成各设置已完成 可以通信DISCONNCETING //待关闭状态
};
class Connection;
using PtrConnection std::shared_ptrConnection;class Connection: public std::enable_shared_from_thisConnection{
private:uint64_t _con_id; //连接的唯一ID便于连接管理和查找//uint64_t _timer_id; //定时器ID必须唯一 为了简化操作直接使用_con_id作为定时器id.int _sockfd; //连接关联的文件描述符bool _enable_inactive_release; //连接是否启动非活跃销毁的判断标志EventLoop* _loop; //连接所关联的一个EventloopSocket _socket; //套接字操作管理Channel _channel; //连接事件管理Buffer _in_buffer; //输入缓冲区 --- 存放从socket中读取到的数据Buffer _out_buffer; //输出缓冲区 --- 存放发送给对端的数据Any _contex; //请求的接收上下文ConnStatu _statu; //连接的状态//这四个回调函数是由组件使用者来设置的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 _close_callback;AnyEventCallback _event_callback;//组件内的连接关闭回调ClosedCallback _server_closed_callback;
private://五个channel的事件回调函数void HandlerRead(){ //描述符可读事件触发后调用的函数char buf[65536];ssize_t ret _socket.NonBlockRecv(buf,65535);if(ret0){ return ShutDownInLoop();//return;}else if(ret 0){return; //这里的等于0是未读取到数据 而非连接断开 }//将数据放入输入缓冲区_in_buffer.WriteAndPush(buf,ret);//2.调用message_callback 进行业务处理if(_in_buffer.ReadAbleSize()0){return _message_callback(shared_from_this(),_in_buffer);}return ReleaseInLoop(); //实际的关闭操作}void HandlerWrite(){ //描述符触发写事件//out_buff中保存的数据就是要发送的数据ssize_t ret _socket.NonBlockSend(_out_buffer.ReadPosition(),_out_buffer.ReadAbleSize());if(ret0){//发送错误该关闭连接了if(_in_buffer.ReadAbleSize() 0){_message_callback(shared_from_this(),_in_buffer);return ReleaseInLoop(); //这时候就是实际的关闭释放操作了}}_out_buffer.MoveReadOffset(ret); //记得将读偏移向后移动if(_out_buffer.ReadAbleSize() 0){_channel.DisableWrite(); //没有数据待发送就关闭写事件监控if(_statu DISCONNCETING){return ReleaseInLoop();}}return;}void HandlerClose(){ //一旦连接挂断套接字就什么都干不了因此有数据待处理就处理 完成关闭连接if(_in_buffer.ReadAbleSize()0){_message_callback(shared_from_this(),_in_buffer);}return ReleaseInLoop();}void HandlerError(){HandlerClose();}void HandlerEvent(){ // 1.刷新连接活跃度 延迟定时销毁任务 2.调用组件使用者的任意事件回调if(_enable_inactive_release true){_loop-TimerRefresh(_con_id);}if(_event_callback){ _event_callback(shared_from_this());}}void EstablishedInLoop(){ //连接获取之后给channel设置事件回调 启动读监控//1.修改连接状态 2.启动读事件监控 3.调用回调函数assert(_statu CONNECTING); //状态必定是半连接状态_statu CONNECTED;_channel.EnableRead();if(_connected_callback){_connected_callback(shared_from_this());}}void SendInLoop(const char* data, size_t len){ //并不是实际的发送接口 只是把数据放到发送缓冲区 启动可写事件监控if(_statu DISCONNECTED) return;_out_buffer.WriteAndPush(data,len);if(_channel.IsWrite() false){_channel.EnableWrite();}}void ReleaseInLoop(){ //这个接口才是实际的释放接口// 1.修改连接状态 置为DISCONNECTED 2.移除连接的事件监控 3. 关闭描述符 4.如果当前定时器队列中还有定时销毁任务则取消任务 5.调用关闭回调函数_statu DISCONNECTED;_channel.ReMove();_socket.Close();if(_loop-HasTimer(_con_id)){ //.......................把这个删掉了 额外添加了个特化的CancleInactiveRelease();}if(_close_callback){_close_callback(shared_from_this());}if(_server_closed_callback){ //移除服务器内部管理的连接信息_server_closed_callback(shared_from_this());}}void ShutDownInLoop(){ //这个关闭操作并非实际的连接释放操作 需判断还有没有数据待处理 _statu DISCONNCETING;if(_in_buffer.ReadAbleSize()0){if(_message_callback){_message_callback(shared_from_this(),_in_buffer);}}if(_out_buffer.ReadAbleSize()0){ //要么写入数据时出错关闭要么没数据发送直接关闭if(_channel.IsWrite() false){_channel.EnableWrite();}}if(_out_buffer.ReadAbleSize() 0){ReleaseInLoop();}}void EnableInactiveReleaseInLoop(int sec){ //启动超时释放规则//1.将判断标准 _enable_inactive_release 置为true_enable_inactive_release true;//2.添加定时销毁任务 如果已经存在 刷新延迟即可if(_loop-HasTimer(_con_id)){return _loop-TimerRefresh(_con_id);}//3.如果不存在定时销毁任务则新增_loop-TimerAdd(_con_id,sec,std::bind(Connection::ReleaseTimeInLoop,this));}void CancleInactiveReleaseInLoop(){_enable_inactive_release false;if(_loop-HasTimer(_con_id))_loop-TimerCancle(_con_id);}void UpgradeInLoop(const Any context,const ConnectedCallback conn, const MessageCallback msg, const ClosedCallback closed,const AnyEventCallback event){ //切换协议 重置上下文数据和阶段处理函数SetContext(context);_connected_callback conn;_message_callback msg;_close_callback closed;_event_callback event;}
public:Connection(EventLoop* loop,uint64_t con_id,int sockfd):_con_id(con_id), _sockfd(sockfd) , _enable_inactive_release(false), _loop(loop),_statu(CONNECTING),_socket(sockfd),_channel(loop,_sockfd){_channel.SetCloseCallBack(std::bind(Connection::HandlerClose,this));_channel.SetEventCallBack(std::bind(Connection::HandlerEvent,this));_channel.SetReadCallBack(std::bind(Connection::HandlerRead,this));_channel.SetWriteCallBack(std::bind(Connection::HandlerWrite,this));_channel.SetErrorCallBack(std::bind(Connection::HandlerError,this));}~Connection(){DBG_LOG(RELEASE CONNECTION: %p,this);}int FD(){return _sockfd;}int ID(){return _con_id;}bool Connected(){return (_statu CONNECTED);}ConnStatu Statu(){return _statu ;} //返回状态void SetContext(const Anycontext){ //设置上下文_contex context;}Any* GetContext(){ //获取上下文return _contex;}void SetConnectedCallback(const ConnectedCallback cb){_connected_callback cb;}void SetMessageCallback(const MessageCallback cb){_message_callback cb;}void SetClosedCallback(const ClosedCallback cb){_close_callback cb;}void SetAnyEventCallback(const ConnectedCallback cb){_event_callback cb;}void SetServerClosedCallback(const ClosedCallback cb){_server_closed_callback cb;}void Established(){ //连接就绪后进行channel回调设置 启动读监控 调用_connected_callback_loop-RunInLoop(std::bind(Connection::EstablishedInLoop,this));}void Send(const char* data, size_t len){ //发送数据到发送缓冲区启动写事件监控 _loop-RunInLoop(std::bind(Connection::SendInLoop,this,data,len));}void Shutdown(){ //关闭接口 --- 并不实际关闭 需判断是否有数待处理_loop-RunInLoop(std::bind(Connection::ShutDownInLoop,this));}void EnableInactiveRelease(int sec){ //启动非活跃销毁并定义多长时间 添加 定时任务_loop-RunInLoop(std::bind(Connection::EnableInactiveReleaseInLoop,this,sec));}void CancleInactiveRelease(){_loop-RunInLoop(std::bind(Connection::CancleInactiveReleaseInLoop,this));}void ReleaseTimeInLoop(){_loop-RunInLoop(std::bind(Connection::ReleaseInLoop,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; //用于对监听套接字进行事件管理using AcceptCallback std::functonvoid(int);AcceptCallback _accept_callback;private://监听套接字的读事件回调public://构造//析构//设置回调函数
};具体实现
class Acceptor{private:Socket _socket; // 用于创建监听套接字EventLoop* _loop; //用于对监听套接字进行事件监控Channel _channel; //用于对监听套接字进行事件管理using AcceptCallback std::functionvoid(int);AcceptCallback _accept_callback;private://监听套接字的读事件回调处理函数 --- 获取新连接调用_accept_callback函数进行新连接处理void HandRead(){int newfd _socket.Accept();if(newfd0){return;}if(_accept_callback) _accept_callback(newfd);}int CreateServer(int port){bool ret _socket.CreateServer(port);assert(ret true);return _socket.FD();}public://不能将启动读事件监控放到构造函数中必须在设置回调函数后再去启动Acceptor(EventLoop* loop,int port):_socket(CreateServer(port)),_loop(loop),_channel(_loop,_socket.FD()){_channel.SetReadCallBack(std::bind(Acceptor::HandRead,this));}void SetAcceptCallback(const AcceptCallback cb){_accept_callback cb;}void Listen(){_channel.EnableRead();}
};TimerQueue模块 编写思路
如何实现定时任务的功能
建小根堆。采用时间轮方案。
我们这里采用的是时间轮的方案。通过多级时间轮来进行定时任务的实现。
如何延迟已经设定的定时非活跃连接销毁任务非活跃连接接收新数据后刷新定时销毁任务
解决方案类的析构函数智能指针shared_ptr
封装一个定时任务对象定时任务的执行放到析构函数中使用shared_ptr管理定时任务对象计数器为0时才释放。刷新定时任务只需要在后面继续添加该定时任务对象即可。
shared_ptr使用注意点 如果多个shared_ptr都是对原始对象进行构造计数都只会为1。因此要存储weak_ptr进行协助。 #includeiostream
#includememory
class TimerTask {};
int main() {TimerTask* t1 new TimerTask;{std::shared_ptrTimerTask pi1(t1);std::shared_ptrTimerTask pi2(t1); }return 0;
}如上面的代码所示我们使用pi1和pi2管理原始对象t1此时pi1和pi2的计数都为1其中一个释放后会导致另一个对象管理的目标已经被释放了。而我们也不能存储shared_ptr因为这会导致计数加1所以要使用weak_ptr进行协助。
接口设计 class TimerTask{private: uint64_t id; //唯一标识uint32_t timeout; //超时时间using TaskFunc std::functionvoid(); //定时任务public:TimerTask(); //构造~TimerTask(); //析构};class TimerWheel{ //时间轮private:using WeakTask std::weak_ptrTimerTask;using PtrTask std::shared_ptrTimerTask;std::vectorstd::vectorPtrTask _whell;int _tick; //秒针int _capacity; //表盘最大数量std::unordered_mapuint64_t,weak_ptrTimerTask _timers;public://构造//析构//添加定时任务//刷新定时任务};具体实现
// 封装定时任务对象 通过析构函数和shared_ptr实现延迟销毁class TimerTask
{private:uint64_t _id; // 标识该任务对象iduint32_t _timeout; // 超时时间TaskFunc _task_cb; // 要执行的定时任务ReleaseFunc _release; // 用于删除timewheel中保存的定时任务对象信息bool _canceled; // false 表示未被取消 true表示取消public:TimerTask(uint64_t id, uint32_t delay, const TaskFunc cb): _id(id), _timeout(delay), _task_cb(cb), _canceled(false){}~TimerTask(){// 析构的时候再执行定时任务if (_canceled false){_task_cb();_release();}}void TimerCancle(){_canceled true;}void SetRelease(const ReleaseFunc re) // 设置{_release re;return;}uint32_t DelayTime(){return _timeout; // 获取延迟时间}
};class EventLoop;
// 时间轮
class TimeWheel
{private:using WeakTask std::weak_ptrTimerTask;using PtrTask std::shared_ptrTimerTask;int _tick; // 秒针指到哪执行到哪里int _capacity; // 表盘的最大数量即最大的延迟时间// 用二维数组存储时间段及该时间的定时任务std::vectorstd::vectorPtrTask _wheel;// 建立定时任务id和weakptr的映射关系 涉及share_ptr的缺陷std::unordered_mapuint64_t, WeakTask _timers;EventLoop *_loop;int _timerfd; // 定时器描述 -- 可读事件回调就是读取计数器 执行定时任务std::unique_ptrChannel _timer_channel;private:void RemoveTimer(uint64_t id){auto it _timers.find(id);if (it ! _timers.end()) // 找到该定时任务{_timers.erase(it);}return;}static int CreateTimerfd(){int timerfd timerfd_create(CLOCK_MONOTONIC, 0);if (timerfd 0){ERR_LOG(timerfd_create FAILED);return -1;}// int timerfd_settime(int fd, int flags,const struct itimerspec *new_value,struct itimerspec *old_value);struct itimerspec itime;itime.it_value.tv_sec 1;itime.it_value.tv_nsec 0;itime.it_interval.tv_sec 1;itime.it_interval.tv_nsec 0;timerfd_settime(timerfd, 0, itime, NULL);return timerfd;}void ReadTimerfd(){uint64_t times;int ret read(_timerfd, times, 8);if (ret 0){ERR_LOG(READ TIMEFD FAILED);abort();}return;}void OnTime(){ReadTimerfd();RunTimerTask();}void TimerAddInLoop(uint64_t id, uint32_t delay, const TaskFunc cb) // 添加定时任务{PtrTask pt(new TimerTask(id, delay, cb));pt-SetRelease(std::bind(TimeWheel::RemoveTimer, this, id)); // 为什么要_timers[id] WeakTask(pt);int pos (_tick delay) % _capacity; // 循环队列一样_wheel[pos].push_back(pt); // 插入延迟任务}void TimerCancleInLoop(uint64_t id){auto it _timers.find(id);if (it _timers.end()){return; // 没该定时任务返回}// 找到后调用PtrTask pt it-second.lock(); //获取智能指针的时候需要判断获取的智能指针是否是有管理对象的if (pt) pt-TimerCancle();return;}void TimerRefreshInLoop(uint64_t id){ // 刷新/延迟定时任务// 通过保存的定时器对象中的weak_ptr构造一个share_ptr添加到轮子auto it _timers.find(id);if (it _timers.end()){return; // 没该定时任务返回}PtrTask pt it-second.lock(); // lock获取weak_ptr管理的对象对应的shared_ptruint32_t delay pt-DelayTime();int pos (_tick delay) % _capacity; // 循环队列一样_wheel[pos].push_back(pt); // 插入延迟任务}public:TimeWheel(EventLoop *loop): _capacity(60), _tick(0), _wheel(_capacity), _loop(loop), _timerfd(CreateTimerfd()), _timer_channel(new Channel(_loop, _timerfd)){_timer_channel-SetReadCallBack(std::bind(TimeWheel::OnTime, this));_timer_channel-EnableRead(); // 启动读事件监控}// 定时器的操作有可能在多线程中执行得保证线程安全// 不想加锁就得把操作都放到一个线程中执行void TimerAdd(uint64_t id, uint32_t delay, const TaskFunc cb);void TimerRefresh(uint64_t id);void TimerCancle(uint64_t id);bool HashTimer(uint64_t id){auto it _timers.find(id);if (it _timers.end()){return false; // 没该定时任务返回}return true;}void RunTimerTask(){ // 执行定时任务应该每秒执行一次相当于秒针走一步_tick (_tick 1) % _capacity;_wheel[_tick].clear(); // 清空该位置的定时任务自动调用定时任务对象的析构}
};Poller模块 编写思路
通过epoll实现对描述符的IO事件监控。
添加描述符的事件监控修改描述符的事件监控移除描述符的事件监控
1和2可整合存在则修改不存在则添加
如何封装
用于一个epoll的操作句柄即文件描述符拥有一个struct epoll_event 结构数组监控保存所有的活跃事件使用hash表管理描述符与对应的事件管理Channel对象
接口设计
class Poller{private:int _epfd;struct epoll_event _evs[MAX_EPOLLEVENTS];std::uinordered_mapint,Channel*private://1.判断要更新的事件的描述符是否存在//2.对epoll直接操作public://1.添加或更新描述符所监控的事件//2.移除描述符的监控//3.开始监控获取就绪的Channel
};具体实现
#define MAX_EPOLLEVENTS 1024
// poller模块
class Poller
{
private:int _epfd;struct epoll_event _evs[MAX_EPOLLEVENTS];std::unordered_mapint, Channel * _channels;private:// 对epoll的直接操作void UpDate(Channel *channel, int op){// int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);int fd channel-Fd();struct epoll_event ep;ep.data.fd fd;ep.events channel-Event();int ret epoll_ctl(_epfd, op, fd, ep);if (ret 0){ERR_LOG(EPOLL CTL ERR);}return;}// 判断一个Channel是否添加了监控bool HasChannel(Channel *channel){auto it _channels.find(channel-Fd());if (it _channels.end()){return false;}return true;}public:Poller(){// int epoll_create(int size); the size argument is ignored, but must be greater than zero;int ret epoll_create(1);if (ret 0){ERR_LOG(EPOLL CREATE ERR);abort(); // 创建失败 终止程序}_epfd ret;}// 1.添加或修改监控事件void UpDateEvent(Channel *channel){bool ret HasChannel(channel);if (ret false){ // 不存在 添加_channels.insert(std::make_pair(channel-Fd(), channel));return UpDate(channel, EPOLL_CTL_ADD);}// 存在更新return UpDate(channel, EPOLL_CTL_MOD);}// 2.移除监控void ReMoveEvent(Channel *channel){auto it _channels.find(channel-Fd());if (it ! _channels.end()){_channels.erase(it);}UpDate(channel, EPOLL_CTL_DEL);}// 3.开始监听 返回活跃连接void Poll(std::vectorChannel * *active){// int epoll_wait(int epfd, struct epoll_event *events,int maxevents, int timeout);int nfds epoll_wait(_epfd, _evs, MAX_EPOLLEVENTS, -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 it _channels.find(_evs[i].data.fd);assert(it ! _channels.end());it-second-SetEvents(_evs[i].events);active-push_back(it-second);}}
};
EventLoop模块 编写思路
进行事件监控以及事件处理的模块此模块是与线程一一对应的。
如何保证一个连接的所有操作都在eventloop对应的线程中
给eventloop模块中添加一个任务队列。对连接的所有操作都进行一次封装对连接的操作当作任务添加到任务队列中。等所有就绪事件处理完然后从task任务队列中一一取出进行操作执行
因此我们只需要给任务队列一把锁即可。
接口设计
class EventLoop{private:std::thread::id _thread_id //线程idint _event_fd; //eventfd唤醒IO事件监控有可能导致的阻塞Poller _poller; //进行所有描述符的事件监控using Functor std::functionvoid(); std::vectorFunctor _tasks; //任务池std::mutex _mutex; //实现任务池操作的线程安全private://执行所有的任务public:EventLoop();//1. 判断将要执行的任务是否处于当前线程中如果是则执行不是则压入队列。//2.将操作压入任务池//3.判断当前线程是否是eventLoop对应的线程//4.添加/修改描述符的事件监控//5.移除描述符的监控//6.启动
};具体实现
class EventLoop
{
private:using Functor std::functionvoid();std::thread::id _thread_id; // 线程IDint _event_fd; // eventfd唤醒IO事件监控可能导致的阻塞Poller _poller; // 进行所有描述符的事件监控std::unique_ptrChannel _event_channel;std::vectorFunctor _tasks; // 任务池std::mutex _mutex; // 实现任务池操作的线程安全TimeWheel _timer_wheel;public:// 执行任务池中的所有任务void RunAllTask(){std::vectorFunctor functor;{std::unique_lockstd::mutex _lock(_mutex);_tasks.swap(functor);}for (auto f : functor){f();}return;}static int CreateEventFd(){int efd eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);if (efd 0){ERR_LOG(CREATE EVENTFD FAILED!!);abort();}return efd;}void ReadEventfd(){uint64_t res 0;int ret read(_event_fd, res, sizeof(res));if (ret 0){ // EINTER 表示被信号打断 EAGAIN 表示无数据可读if (errno EINTR){return;}ERR_LOG(READ EVENTFD FAILED);abort();}return;}void WakeUpEventFd(){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()), _poller(), _event_channel(new Channel(this, _event_fd)), _timer_wheel(this){_event_channel-SetReadCallBack(std::bind(EventLoop::ReadEventfd, this));// 启动eventfd的读事件监控_event_channel-EnableRead();}void RunInLoop(const Functor cb){ // 当前任务在当前线程即执行不是则压入队列if (IsInLoop()){return cb();}return QueueInLoop(cb);}void QueueInLoop(const Functor cb){ // 将操作压入任务池{std::unique_lockstd::mutex _lock(_mutex);_tasks.push_back(cb);}// 唤醒有可能以为没事情就绪导致的epoll阻塞// 其实就是给eventfd写一个数据eventfd就会触发可读事件WakeUpEventFd();}bool IsInLoop(){ // 判断当前线程是否是EventLoop对应的线程return _thread_id std::this_thread::get_id();}void AssertInLoop(){assert(_thread_id std::this_thread::get_id());}void UpdateEvent(Channel *channel){ // 添加/修改描述符的事件监控_poller.UpDateEvent(channel);}void RemoveEvent(Channel *channel){ // 移除监控_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 TimerCancle(uint64_t id){return _timer_wheel.TimerCancle(id);}bool HasTimer(uint64_t id){return _timer_wheel.HashTimer(id);}void Start(){ // 事件监控 --- 就绪事件处理 ---- 执行任务// 1.事件监控while(1){std::vectorChannel * actives;_poller.Poll(actives);// 2.事件处理for (auto channel : actives){channel-HandleEvent();}// 3.执行任务RunAllTask(); }}
};TcpServer模块 编写思路
对所有模块的整合通过TcpServer模块实例化对象可以非常简单的完成一个服务器的搭建。
接口设计
class TcpServer{
private:int _port;uint64_t _conn_id; //自动增长的连接IdAcceptor _acceptor; //监听套接字的管理对象EventLoop _baseloop; //这是主线程的EventLoop对象负责监听事件的处理LoopThreadPool _pool; //这是从属EventLoop线程池int _timeout; //这是非活跃连接的统计事件 -- 多长时间无通信就是非活跃连接bool _enable_inactive_release; //释放需要释放非活跃连接的标志std::unordered_mapuint64_t,PtrConnection _conns; //保存管理所有连接对应的shared_ptr对象using Functor std::functionvoid(); //用户设置的定时任务//这四个回调函数是由组件使用者来设置的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 _close_callback;AnyEventCallback _event_callback;
public://构造//析构//设置从属线程数量//设置回调//定时任务添加//启动服务器具体实现 class TcpServer{private:int _port;uint64_t _conn_id; //自动增长的连接IdAcceptor _acceptor; //监听套接字的管理对象EventLoop _baseloop; //这是主线程的EventLoop对象负责监听事件的处理LoopThreadPool _pool; //这是从属EventLoop线程池int _timeout; //这是非活跃连接的统计事件 -- 多长时间无通信就是非活跃连接bool _enable_inactive_release; //释放需要释放非活跃连接的标志std::unordered_mapuint64_t,PtrConnection _conns; //保存管理所有连接对应的shared_ptr对象using Functor std::functionvoid(); //用户设置的定时任务//这四个回调函数是由组件使用者来设置的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 _close_callback;AnyEventCallback _event_callback;private: void NewConnection(int fd){ //给新连接构造一个Connection管理_conn_id;PtrConnection conn(new Connection(_pool.NextLoop(),_conn_id,fd));conn-SetMessageCallback(_message_callback);conn-SetClosedCallback(_close_callback);conn-SetConnectedCallback(_connected_callback);conn-SetAnyEventCallback(_event_callback);conn-SetServerClosedCallback(std::bind(TcpServer::RemoveConnection,this,std::placeholders::_1));if(_enable_inactive_release) conn-EnableInactiveRelease(_timeout); //启动超时销毁conn-Established(); //就绪初始化_conns.insert(std::make_pair(_conn_id,conn));}void RemoveConnectionInLoop(const PtrConnection conn){int id conn-ID();auto it _conns.find(id);if(it ! _conns.end()){_conns.erase(it);}}void RunAfterInLoop(const Functor task,int delay){_conn_id;_baseloop.TimerAdd(_conn_id,delay,task);}public:TcpServer(int port):_port(port),_conn_id(0),_enable_inactive_release(false),_acceptor(_baseloop,_port),_pool(_baseloop){_acceptor.SetAcceptCallback(std::bind(TcpServer::NewConnection,this,std::placeholders::_1));_acceptor.Listen(); //将监听套接字挂到baseloop上开始监控}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){_close_callback cb;}void SetAnyEventCallback(const ConnectedCallback 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 RemoveConnection(const PtrConnection conn){ //从管理Connection的_conns中移除连接信息_baseloop.RunInLoop(std::bind(TcpServer::RemoveConnectionInLoop,this,conn));}void Start(){_pool.Create(); //创建线程池中的从属线程return _baseloop.Start();}};协议模块
Util模块 具体实现
class Util{public://字符串分割函数 将src字符串按照sep字符进行分割 得到各个字符串放到arry中最终返回子串的数量static size_t Split(const std::string src,const std::string sep,std::vectorstd::string* arry){size_t offset 0;while(offset src.size()){size_t pos src.find(sep,offset); //在src字符串偏移量offset处 开始向后查找seo字符/子串返回查找的位置if(pos std::string::npos){ //未找到特定的字符//将剩余的部分当作一个子串 放入arry中arry-push_back(src.substr(offset));return arry-size(); }if(pos offset) {offset pos sep.size();continue; //当前子串是空的} arry-push_back(src.substr(offset,pos - offset));offset pos sep.size();}return arry-size(); }//读取文件的所有内容将读取的内容放入一个Buffer中static bool ReadFile(const std::string filename,std::string* buf){std::ifstream ifs(filename,std::ios::binary);if(ifs.is_open() false){ERR_LOG(open failed!!,filename.c_str());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){ERR_LOG(READ %s FILE FAILED!,filename.c_str());ifs.close();return false;}ifs.close();return true;}//向文件写入数据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){ERR_LOG(open failed!!,filename.c_str());return false;}ofs.write(buf.c_str(),buf.size());if(ofs.good() false){ERR_LOG(WRITE %s FILE FAILED!,filename.c_str());ofs.close();return false;}ofs.close();return true;}//URL编码 避免URL中资源路径与查询字符串中的特殊字符与Http请求中特殊字符产生歧义//编码格式将特殊字符的ascii值转换为两个16进制字符 前缀% c - c%2B%2B//不编码的特殊字符 RFC3986文档规定 . - _ ~ 字母数字属于绝对不编码字符//RFC3986文档规定 编码格式 %HH//W3C文档中规定 查询字符串中的空格需要被编码成 解码则是转空格 static std::string UrlEncode(const std::string url,bool convert_space_to_plus){std::string res; for(auto c : url){if(c . || c - || c_ || c ~ ||isalnum(c)){res c;continue;}if(c convert_space_to_plus){res ;continue;}//剩下的字符都需要编码成%HH格式char tmp[4] {0};snprintf(tmp,4,%%%02X,c);res tmp;}return res; }static int HEXTOI(char c){ //16进制转整型if(c 0 c 9){return c-0;}else if(c a c z){return c - a 10;}else if(cA cZ){return c - A 10;}return -1;}//URL解码static std::string UrlDecode(const std::string url,bool convert_plus_to_space){//遇到了%则将紧随其后的2个字符转换成数字第一个数字左移4位然后加上第二个数字 %2b 24 11;std::string res;for(int i 0;iurl.size();i){if(url[i] convert_plus_to_space){res ;}if(url[i] % (i2) url.size()){char v1 HEXTOI(url[i1]);char v2 HEXTOI(url[i2]);char v (v14) v2;res v;i2;continue;}res url[i];}return res;}//响应状态码的描述信息获取static std::string StatueDesc(int statu){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},};auto it _statu_msg.find(statu);if(it ! _statu_msg.end()){return it-second;}else UNKNOW STATU;} //根据文件后缀名获取文件mimestatic std::string FileNameExt(const std::string filename){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},};size_t pos filename.find_last_of(.);if(pos std::string::npos){return applaction/octet-stream;}std::string ext filename.substr(pos); //获取扩展名auto it _mime_msg.find(ext);if(it _mime_msg.end()){return applaction/octet-stream;}return it-second;}//判断是否是目录static bool IsDirectory(const std::string filename){struct stat st;int ret stat(filename.c_str(),st);if(ret0){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(ret0){return false;}return S_ISREG(st.st_mode);}//http请求的资源路径是否有效// /index.html --- / 相对根目录//客户端只能请求相对跟目录的资源其他地方的资源都不理会 遇到..对层数减一static bool ValiedPath(const std::string path){ //计算目录深度std::vectorstd::string subdir;Split(path,/,subdir);int level 0;for(auto dir :subdir){if(dir ..){level--;if(level 0){return false;} continue;}level;}return true;}
};
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 _headrs; //头部字段std::unordered_mapstd::string,std::string _params; //查询字符串public: void ReSet(){_method.clear();_path.clear();_version.clear();_body.clear();std::smatch match;_matches.swap(match);_headrs.clear();_params.clear();}//插入头部字段 void SetHeader(const std::string key ,const std::string val){_headrs.insert(std::make_pair(key,val));}//判断是否存在指定头部字段bool HasHeader(const std::string key) const {auto it _headrs.find(key);if (it _headrs.end()) {return false;}return true;}//获取指定头部字段的值std::string GetHeader(const std::string key) const {auto it _headrs.find(key);if (it _headrs.end()) {return ;}return it-second;}//插入查询字符串void SetParam(std::string key,std::string val){_params.insert(std::make_pair(key,val));}//判断是否存在查询字符串bool HasParam(const std::string key){auto it _params.find(key);if(it _params.end()){return false;}return true;}//获取指定查询字符串的值std::string GetParam(const std::string key){auto it _params.find(key);if(it _params.end()){return ;}return it-second;}//获取正文长度size_t ContentLength(){//Content-Lengthbool ret HasHeader(Content-Length);if(ret false){return 0;}std::string clen GetHeader(Content-Length);return std::stol(clen);}//判断是否是短链接bool Close() const {// 没有Connection字段或者有Connection但是值是close则都是短链接否则就是长连接if (HasHeader(Connection) true GetHeader(Connection) keep-alive) {return false;}return true;}
};
HttpResponse模块 具体实现
class HttpResponse{public:int _statu;bool _redirect_flag;std::string _body;std::string _redirect_url;std::unordered_mapstd::string,std::string _headers;public:HttpResponse():_redirect_flag(false),_statu(200) {}HttpResponse(int statu):_redirect_flag(false),_statu(statu) {}void Rest(){_statu 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(std::string body,const std::string type){_body body;SetHeader(Content-Type,type);}void SerRedirect(std::string url,int statu 302){_statu statu;_redirect_flag true;_redirect_url url;}//判断是否是短链接bool Close(){//没有Connection字段 或者字段是close 都是短连接 if(HasHeader(Connection) true GetHeader(Connection) keep-alive){return true;}return false;}
};请求接收上下文模块 具体实现
enum HttpRecvStatu{RECV_HTTP_ERROR,RECV_HTTP_LINE,RECV_HTTP_HEAD,RECV_HTTP_BODY,RECV_HTTP_OVER
};
#define MAX_LINE 8192
class HttpContext{private:int _resp_statu; //响应状态码HttpRecvStatu _recv_statu;HttpRequest _request;private:bool ParseHttpLine(const std::string line){std::smatch matches;//请求方法匹配 GET HEAD POST PUT DELETE 。。。。std::regex e((GET|HEAD|POST|PUT|DELETE) ([^?]*)(?:\\?(.*))? (HTTP/1\\.[01])(?:\n|\r|\r\n)?) ;bool ret std::regex_match(line,matches,e);if(ret false){_recv_statu RECV_HTTP_ERROR;_resp_statu 400; //BAD REQUESTreturn false;}_request._method matches[1]; //请求方法的获取_request._path Util::UrlDecode(matches[2],false); //资源路径的获取 需要解码操作 不需要转空格_request._version matches[4]; //协议版本的获取std::vectorstd::string query_string_arry;std::string query_string matches[3]; //查询字符串的处理Util::Split(query_string,,query_string_arry); // keyvalkeyval 先以分割for(auto str : query_string_arry){size_t pos str.find();if(pos std::string::npos){_recv_statu RECV_HTTP_ERROR;_resp_statu 400; //BAD REQUESTreturn false;}std::string key Util::UrlDecode(str.substr(0,pos),true); //以等会分割并且解码 需要转空格std::string val Util::UrlDecode(str.substr(pos1),true);_request.SetParam(key,val);}}bool RecvHttpLine(Buffer* buf){if(_recv_statu ! RECV_HTTP_LINE) return false;//获取一行数据 可能 缓冲区数据不足一行 获取的一行数据超大std::string line buf-GetLineAndPop();if(line.size() 0){//缓冲区数据不足一行,则需要判断缓冲区数据可读长度如果很长都不足一行 有问题if(buf-ReadAbleSize() MAX_LINE){_recv_statu RECV_HTTP_ERROR;_resp_statu 414; //URL TOO LONGreturn false;}//等新数据到来return true;}bool ret ParseHttpLine(line);if(ret false){return false;}buf-MoveReadOffset(line.size());_recv_statu RECV_HTTP_HEAD; //首行处理完毕 进入头部获取阶段return true;}bool RecvHttpHead(Buffer* buf){ //一行行取出数据 遇到空行为止 key: /r/n.../r/nif(_recv_statu! RECV_HTTP_HEAD){return false;}while(1){//获取一行数据 可能 缓冲区数据不足一行 获取的一行数据超大std::string line buf-GetLineAndPop();if(line.size() 0){//缓冲区数据不足一行,则需要判断缓冲区数据可读长度如果很长都不足一行 有问题if(buf-ReadAbleSize() MAX_LINE){_recv_statu RECV_HTTP_ERROR;_resp_statu 414; //URL TOO LONGreturn false;}//等新数据到来return true;}if(line \n || line \r\n){break;}bool ret ParseHttpHead(line);if(ret false){return false;}} _recv_statu RECV_HTTP_BODY; //头部获取结束 进入正文获取阶段return true; }bool ParseHttpHead(const std::string line){size_t pos line.find(: );if(pos std::string::npos){_recv_statu RECV_HTTP_ERROR;_resp_statu 400; //URL TOO LONGreturn false;}std::string key line.substr(0,pos);std::string val line.substr(pos2);_request.SetHeader(key,val);return true;}bool RecvHttpBody(Buffer* buf){if(_recv_statu ! RECV_HTTP_BODY) return false;//1。获取正文长度size_t content_length _request.ContentLength();if(content_length 0){//无正文接收完毕_recv_statu RECV_HTTP_OVER;return true;}//2.当前已经接收了多少正文 取决于_request.body size_t real_len content_length - _request._body.size(); //实际还需接收的正文长度//3.接收正文放到body中但是也要考虑当前缓冲区中的数据是否是全部的正文。//3.1 缓冲区中数据包含了当前请求的所有正文则取出所需的数据if(buf-ReadAbleSize() real_len){_request._body.append(buf-ReadPosition(),real_len);buf-MoveReadOffset(real_len);_recv_statu RECV_HTTP_OVER;return true;}//3.2 缓冲区中数据无法满足当前正文的需要数据不足取出数据然后等待新数据到来_request._body.append(buf-ReadPosition(),buf-ReadAbleSize());buf-MoveReadOffset(buf-ReadAbleSize());return true;}public:HttpContext():_resp_statu(200),_recv_statu(RECV_HTTP_LINE){};int RespStatu(){return _resp_statu;}void ReSet(){_resp_statu 200;_recv_statu RECV_HTTP_LINE;_request.ReSet();} HttpRecvStatu RecvStatu(){return _recv_statu;}HttpRequest Request(){return _request;}void RecvHttpRequest(Buffer* buf){//不同状态做不同的事情这里不要break,switch(_recv_statu){case RECV_HTTP_LINE: RecvHttpLine(buf);case RECV_HTTP_HEAD: RecvHttpHead(buf);case RECV_HTTP_BODY: RecvHttpBody(buf);}return;}
};HttpServer模块 具体实现
#define DEFALT_TIMEOUT 30
class HttpServer {private:using Handler std::functionvoid(const HttpRequest , HttpResponse *);using Handlers std::vectorstd::pairstd::regex, Handler;Handlers _get_route;Handlers _post_route;Handlers _put_route;Handlers _delete_route;std::string _basedir; //静态资源根目录TcpServer _server;private:void ErrorHandler(const HttpRequest req, HttpResponse *rsp) {//1. 组织一个错误展示页面std::string body;body html;body head;body meta http-equivContent-Type contenttext/html;charsetutf-8;body /head;body body;body h1;body std::to_string(rsp-_statu);body ;body Util::StatueDesc(rsp-_statu);body /h1;body /body;body /html;//2. 将页面数据当作响应正文放入rsp中rsp-SetContent(body, text/html);}//将HttpResponse中的要素按照http协议格式进行组织发送void WriteReponse(const PtrConnection conn, const HttpRequest req, HttpResponse rsp) {//1. 先完善头部字段if (req.Close() true) {rsp.SetHeader(Connection, close);}else {rsp.SetHeader(Connection, keep-alive);}if (rsp._body.empty() false rsp.HasHeader(Content-Length) false) {rsp.SetHeader(Content-Length, std::to_string(rsp._body.size()));}if (rsp._body.empty() false rsp.HasHeader(Content-Type) false) {rsp.SetHeader(Content-Type, application/octet-stream);}if (rsp._redirect_flag true) {rsp.SetHeader(Location, rsp._redirect_url);}//2. 将rsp中的要素按照http协议格式进行组织std::stringstream rsp_str;rsp_str req._version std::to_string(rsp._statu) Util::StatueDesc(rsp._statu) \r\n;for (auto head : rsp._headers) {rsp_str head.first : head.second \r\n;}rsp_str \r\n;rsp_str rsp._body;//3. 发送数据conn-Send(rsp_str.str().c_str(), rsp_str.str().size());}bool IsFileHandler(const HttpRequest req) {// 1. 必须设置了静态资源根目录if (_basedir.empty()) {return false;}// 2. 请求方法必须是GET / HEAD请求方法if (req._method ! GET req._method ! HEAD) {return false;}// 3. 请求的资源路径必须是一个合法路径if (Util::ValiedPath(req._path) false) {return false;}// 4. 请求的资源必须存在,且是一个普通文件// 有一种请求比较特殊 -- 目录/, /image/ 这种情况给后边默认追加一个 index.html// index.html /image/a.png// 不要忘了前缀的相对根目录,也就是将请求路径转换为实际存在的路径 /image/a.png - ./wwwroot/image/a.pngstd::string req_path _basedir 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 *rsp) {std::string req_path _basedir req._path;if (req._path.back() /) {req_path index.html;}bool ret Util::ReadFile(req_path, rsp-_body);if (ret false) {return;}std::string mime Util::FileNameExt(req_path);rsp-SetHeader(Content-Type, mime);return;}//功能性请求的分类处理void Dispatcher(HttpRequest req, HttpResponse *rsp, Handlers handlers) {//在对应请求方法的路由表中查找是否含有对应资源请求的处理函数有则调用没有则发挥404//思想路由表存储的时键值对 -- 正则表达式 处理函数//使用正则表达式对请求的资源路径进行正则匹配匹配成功就使用对应函数进行处理// /numbers/(\d) /numbers/12345for (auto handler : handlers) {const std::regex re handler.first;const Handler functor handler.second;bool ret std::regex_match(req._path, req._matches, re);if (ret false) {continue;}return functor(req, rsp);//传入请求信息和空的rsp执行处理函数}rsp-_statu 404;}void Route(HttpRequest req, HttpResponse *rsp) {//1. 对请求进行分辨是一个静态资源请求还是一个功能性请求// 静态资源请求则进行静态资源的处理// 功能性请求则需要通过几个请求路由表来确定是否有处理函数// 既不是静态资源请求也没有设置对应的功能性请求处理函数就返回405if (IsFileHandler(req) true) {//是一个静态资源请求, 则进行静态资源请求的处理return FileHandler(req, rsp);}if (req._method GET || req._method HEAD) {return Dispatcher(req, rsp, _get_route);}else if (req._method POST) {return Dispatcher(req, rsp, _post_route);}else if (req._method PUT) {return Dispatcher(req, rsp, _put_route);}else if (req._method DELETE) {return Dispatcher(req, rsp, _delete_route);}rsp-_statu 405;// Method Not Allowedreturn ;}//设置上下文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();//2. 通过上下文对缓冲区数据进行解析得到HttpRequest对象// 1. 如果缓冲区的数据解析出错就直接回复出错响应// 2. 如果解析正常且请求已经获取完毕才开始去进行处理context-RecvHttpRequest(buffer);HttpRequest req context-Request();HttpResponse rsp(context-RespStatu());if (context-RespStatu() 400) {//进行错误响应关闭连接ErrorHandler(req, rsp);//填充一个错误显示页面数据到rsp中WriteReponse(conn, req, rsp);//组织响应发送给客户端context-ReSet();buffer-MoveReadOffset(buffer-ReadAbleSize());//出错了就把缓冲区数据清空conn-Shutdown();//关闭连接return;}if (context-RecvStatu() ! RECV_HTTP_OVER) {//当前请求还没有接收完整,则退出等新数据到来再重新继续处理return;}//3. 请求路由 业务处理Route(req, rsp);//4. 对HttpResponse进行组织发送WriteReponse(conn, req, rsp);//5. 重置上下文context-ReSet();//6. 根据长短连接判断是否关闭连接或者继续处理if (rsp.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));}void SetBaseDir(const std::string path) {assert(Util::IsDirectory(path) true);_basedir path;}/*设置/添加请求请求的正则表达与处理函数的映射关系*/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() {_server.Start();}
};性能测试
采⽤webbench进⾏服务器性能测试。
webbench测试原理是创建指定数量的进程在进程中创建客户端向服务器发送请求收到响应后关闭连接开始下一个连接的建立。
性能测试的两个重点衡量标准
并发量可以同时处理多少客户端的请求而不会出现连接失败QPS每秒钟处理的包的数量 测试环境
服务器环境2核2G带宽3M的云服务器服务器程序采用1主3从reactor模式
webbench客户端环境同一云服务器
使用webbench分别以500500010000并发量向服务器发送请求进行了一分钟测试。
测试结果
测试1500个客⼾端连接的情况下测试结果
[solahcss-ecs-e2f8 WebBench-master]$ ./webbench -c 500 -t 60 http://127.0.0.1:8088/
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.1Runing info: 500 clients, running 60 sec.Speed209176 pages/min, 1284062 bytes/sec.
Requests: 209176 susceed, 0 failed.测试25000个客⼾端连接的情况下测试结果
[solahcss-ecs-e2f8 WebBench-master]$ ./webbench -c 5000 -t 60 http://127.0.0.1:8088/
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.1Runing info: 5000 clients, running 60 sec.Speed200718 pages/min, 1232855 bytes/sec.
Requests: 200718 susceed, 0 failed.
测试310000个客⼾端连接的情况下测试结果 [solahcss-ecs-e2f8 WebBench-master]$ ./webbench -c 10000 -t 60 http://127.0.0.1:8088/
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.1Runing info: 10000 clients, running 60 sec.Speed197783 pages/min, 1215718 bytes/sec.
Requests: 197783 susceed, 0 failed.
以上测试中使⽤浏览器访问服务器均能流畅获取请求的⻚⾯。但是根据测试结果能够看出虽然并发量⼀直在提⾼但是总的请求服务器的数量并没有增加反⽽有所降低侧⾯反馈了处理所耗时间更多了基本上可以根据19w/min左右的请求量计算出10000并发量时服务器的极限了.
⽬前受限于设备环境配置尚未进⾏更多并发量的测试.
