网站首页推荐,网络服务提供者发现用户利用其网络服务对未成年,可以做本地生活服务的有哪些网站,小程序找不到怎么办项目完整在#xff1a; 文章目录 一、Connection模块#xff1a;这是一个对于通信连接进行整体管理的一个模块#xff0c;对一个连接的操作都是通过这个模块来进行#xff01;二、提供的功能三、实现思想#xff08;一#xff09;功能#xff08;二#xff09;意义 文章目录 一、Connection模块这是一个对于通信连接进行整体管理的一个模块对一个连接的操作都是通过这个模块来进行二、提供的功能三、实现思想一功能二意义三功能设计 四、框架五、代码 一、Connection模块这是一个对于通信连接进行整体管理的一个模块对一个连接的操作都是通过这个模块来进行
二、提供的功能
Connection模块一个连接有任何的事件怎么处理都是有这个模块来进行处理的因为组件的设计也不知道使用者要如何处理事件因此只能是提供一些事件回调函数由使用者设置。
三、实现思想
一功能
发送数据 —— 给用户提供的发生数据的接口并不是真的发送接口而只是把数据发送到发送缓冲区然后启动写事件监控关闭连接 —— 给用户提供的关闭链接接口在实际释放连接之前看看输入输出缓冲区是否有数据待处理启动非活跃连接超时销毁功能取消非活跃连接超时销毁功能协议切换 —— 一个链接接受数据后如何进行业务处理取决于上下文以及数据的业务处理回调函数
二意义
这个模块本身不是一个单独的功能模块是一个对连接做管理的模块。
三功能设计
套接字的管理能够进行套接字的操作连接事件的管理可读可写错误挂断任意缓冲区管理把socket读取的数据放进缓冲区要有输入缓冲区和输出缓冲区管理协议上下文的管理记录请求数据的处理过程回调函数的管理 因为连接收到数据之后该如何处理需要由用户决定必须要有业务处理函数 一个连接建立成功后应该如何处理由用户决定因此必须有连接建立成功的回调函数 一个连接关闭前该如何处理有用户决定因此必须有关闭连接回调函数 任何事件的产生有没有某些处理由用户决定因此必须任意事件的回调函数
场景对连接进行操作的时候对于连接以及被释放导致内存访问错误最终程序崩溃 解决方案使用智能指针share_ptr 对Connect 对象进行管理这样可以保证任意一个地方对Connect对象进行操作的时候 保存了一分share_ptr,因此就算其他地方进行了释放也只是对share_ptr的计数器-1而不会导致Connection的实际释放
四、框架
DISCONECTED -- 连接关闭状态 CONNECTING -- 连接建立成功-待处理状态
//CONNECTED -- 连接建立完成各种设置已完成可以通信的状态 DISCONNECTING -- 待关闭状态
type enum { // 连接关闭// 连接建立成功 —— 待处理状态// 连接设立完成可以通信// 待关闭状态DISCONECTED,CONNECTING,CONNECTED,DISCONECTING} ConnStatu;
using PreConnection std::shared_ptrConnection;
class Connection {private:uint64_t _conn_id; //连接的唯一ID便于连接的管理和查找bool _enable_inactive_release; // 连接是否启动非活跃销毁的判断标志默认为falseint _sockfd; // 连接关联的文件描述符ConnStatu _statu; // Socket _socket; // 套接字操作管理Channel _channel; // 连接二点事件管理Buffer _in_buffer; // 输入缓冲区 —— 存放从socket中读到的数据buffer _out_buffer; // 输出缓冲区 —— 发送给对端的是数据等到描述符事件可写再发Any _context; // 请求的接受处理上下文/*这四个回调函数是让服务器模块来设置的其实服务器模块的处理回调也是组件使用者设置的*//*换句话说这几个回调都是组件使用者使用的*/using ConnectCallback std::functionvoid(const PreConnection);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;/*组件内的连接关闭回调--组件内设置的因为服务器组件内会把所有的连接管理起来一旦某个连接要关闭*//*就应该从管理的地方移除掉自己的信息*/ClosedCallback _server_closed_callback;private:// /*五个channel的事件回调函数*///描述符可读事件触发后调用的函数接收socket数据放到接收缓冲区中然后调用_message_callbackvoid HandleRead() {}void HandleRead() {}void HandleClose() {}void HandleError() {}//描述符触发任意事件: 1. 刷新连接的活跃度--延迟定时销毁任务 2. 调用组件使用者的任意事件回调void HandleEvent() { }//连接获取之后所处的状态下要进行各种设置启动读监控,调用回调函数void EstablishedInLoop() { }//这个接口才是实际的释放接口void ReleaseInLoop() {}//这个接口并不是实际的发送接口而只是把数据放到了发送缓冲区启动了可写事件监控void SendInLoop(Buffer buf) {}//这个关闭操作并非实际的连接释放操作需要判断还有没有数据待处理待发送void ShutdownInLoop() {}//启动非活跃连接超时释放规则void EnableInactiveReleaseInLoop(int sec) {}void CancelInactiveReleaseInLoop() {}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) : _sockfd(sockfd),_enable_inactive_release(false), _loop(loop), _statu(CONNECTING), _socket(_sockfd), _channel(loop, _sockfd) {_channel.SetCloseCallback(std::bind(Connection::HandleClose, this));_channel.SetEventCallback(std::bind(Connection::HandleEvent, this));_channel.SetReadCallback(std::bind(Connection::HandleRead, this));_channel.SetWriteCallback(std::bind(Connection::HandleWrite, this));_channel.SetErrorCallback(std::bind(Connection::HandleError, this));}~Connection() { DBG_LOG(RELEASE CONNECTION:%p, this); }//获取管理的文件描述符int Fd() {return _sockfd; }// 获取连接IDint Id() {return _conn_id; }// 是否处于CONNECTED状态bool Connected() { return (_statu CONNECTED); }//设置上下文--连接建立完成时进行调用void SetContext(const Any context) { _context context; }//获取上下文返回的是指针Any *GetContext() { return _context; }void SetConnectedCallback(const ConnectedCallbackcb) { _connected_callback cb; }void SetMessageCallback(const MessageCallbackcb) { _message_callback cb; }void SetClosedCallback(const ClosedCallbackcb) { _closed_callback cb; }void SetAnyEventCallback(const AnyEventCallbackcb) { _event_callback cb; }void SetSrvClosedCallback(const ClosedCallbackcb) { _server_closed_callback cb; }//连接建立就绪后进行channel回调设置启动读监控调用_connected_callbackvoid Established() {}_loop-RunInLoop(std::bind(Connection::EstablishedInLoop, this));// 发送数据将数据放到发送缓冲区启动写事件监控void Send(const char *data, size_t len) {}//提供给组件使用者的关闭接口--并不实际关闭需要判断有没有数据待处理void Shutdown() {}void Release() {}//启动非活跃销毁并定义多长时间无通信就是非活跃添加定时任务void EnableInactiveRelease(int sec) { }//取消非活跃销毁void CancelInactiveRelease() {}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));}};五、代码
class Connection;
//DISCONECTED -- 连接关闭状态 CONNECTING -- 连接建立成功-待处理状态
//CONNECTED -- 连接建立完成各种设置已完成可以通信的状态 DISCONNECTING -- 待关闭状态
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便于连接的管理和查找//uint64_t _timer_id; //定时器ID必须是唯一的这块为了简化操作使用conn_id作为定时器IDint _sockfd; // 连接关联的文件描述符bool _enable_inactive_release; // 连接是否启动非活跃销毁的判断标志默认为falseEventLoop *_loop; // 连接所关联的一个EventLoopConnStatu _statu; // 连接状态Socket _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;/*组件内的连接关闭回调--组件内设置的因为服务器组件内会把所有的连接管理起来一旦某个连接要关闭*//*就应该从管理的地方移除掉自己的信息*/ClosedCallback _server_closed_callback;private:/*五个channel的事件回调函数*///描述符可读事件触发后调用的函数接收socket数据放到接收缓冲区中然后调用_message_callbackvoid HandleRead() {//1. 接收socket的数据放到缓冲区char buf[65536];ssize_t ret _socket.NonBlockRecv(buf, 65535);if (ret 0) {//出错了,不能直接关闭连接return ShutdownInLoop();}//这里的等于0表示的是没有读取到数据而并不是连接断开了连接断开返回的是-1//将数据放入输入缓冲区,写入之后顺便将写偏移向后移动_in_buffer.WriteAndPush(buf, ret);//2. 调用message_callback进行业务处理if (_in_buffer.ReadAbleSize() 0) {//shared_from_this--从当前对象自身获取自身的shared_ptr管理对象return _message_callback(shared_from_this(), _in_buffer);}}//描述符可写事件触发后调用的函数将发送缓冲区中的数据进行发送void HandleWrite() {//_out_buffer中保存的数据就是要发送的数据ssize_t ret _socket.NonBlockSend(_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 (_statu DISCONNECTING) {return Release();}}return;}//描述符触发挂断事件void HandleClose() {/*一旦连接挂断了套接字就什么都干不了了因此有数据待处理就处理一下完毕关闭连接*/if (_in_buffer.ReadAbleSize() 0) {_message_callback(shared_from_this(), _in_buffer);}return Release();}//描述符触发出错事件void HandleError() {return HandleClose();}//描述符触发任意事件: 1. 刷新连接的活跃度--延迟定时销毁任务 2. 调用组件使用者的任意事件回调void HandleEvent() {if (_enable_inactive_release true) { _loop-TimerRefresh(_conn_id); }if (_event_callback) { _event_callback(shared_from_this()); }}//连接获取之后所处的状态下要进行各种设置启动读监控,调用回调函数void EstablishedInLoop() {// 1. 修改连接状态 2. 启动读事件监控 3. 调用回调函数assert(_statu CONNECTING);//当前的状态必须一定是上层的半连接状态_statu CONNECTED;//当前函数执行完毕则连接进入已完成连接状态// 一旦启动读事件监控就有可能会立即触发读事件如果这时候启动了非活跃连接销毁_channel.EnableRead();if (_connected_callback) _connected_callback(shared_from_this());}//这个接口才是实际的释放接口void ReleaseInLoop() {//1. 修改连接状态将其置为DISCONNECTED_statu DISCONNECTED;//2. 移除连接的事件监控_channel.Remove();//3. 关闭描述符_socket.Close();//4. 如果当前定时器队列中还有定时销毁任务则取消任务if (_loop-HasTimer(_conn_id)) CancelInactiveReleaseInLoop();//5. 调用关闭回调函数避免先移除服务器管理的连接信息导致Connection被释放再去处理会出错因此先调用用户的回调函数if (_closed_callback) _closed_callback(shared_from_this());//移除服务器内部管理的连接信息if (_server_closed_callback) _server_closed_callback(shared_from_this());}//这个接口并不是实际的发送接口而只是把数据放到了发送缓冲区启动了可写事件监控void SendInLoop(Buffer buf) {if (_statu DISCONNECTED) return ;_out_buffer.WriteBufferAndPush(buf);if (_channel.WriteAble() false) {_channel.EnableWrite();}}//这个关闭操作并非实际的连接释放操作需要判断还有没有数据待处理待发送void ShutdownInLoop() {_statu 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 EnableInactiveReleaseInLoop(int sec) {//1. 将判断标志 _enable_inactive_release 置为true_enable_inactive_release true;//2. 如果当前定时销毁任务已经存在那就刷新延迟一下即可if (_loop-HasTimer(_conn_id)) {return _loop-TimerRefresh(_conn_id);}//3. 如果不存在定时销毁任务则新增_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 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), _statu(CONNECTING), _socket(_sockfd),_channel(loop, _sockfd) {_channel.SetCloseCallback(std::bind(Connection::HandleClose, this));_channel.SetEventCallback(std::bind(Connection::HandleEvent, this));_channel.SetReadCallback(std::bind(Connection::HandleRead, this));_channel.SetWriteCallback(std::bind(Connection::HandleWrite, this));_channel.SetErrorCallback(std::bind(Connection::HandleError, this));}~Connection() { DBG_LOG(RELEASE CONNECTION:%p, this); }//获取管理的文件描述符int Fd() { return _sockfd; }//获取连接IDint Id() { return _conn_id; }//是否处于CONNECTED状态bool Connected() { return (_statu CONNECTED); }//设置上下文--连接建立完成时进行调用void SetContext(const Any context) { _context context; }//获取上下文返回的是指针Any *GetContext() { return _context; }void SetConnectedCallback(const ConnectedCallbackcb) { _connected_callback cb; }void SetMessageCallback(const MessageCallbackcb) { _message_callback cb; }void SetClosedCallback(const ClosedCallbackcb) { _closed_callback cb; }void SetAnyEventCallback(const AnyEventCallbackcb) { _event_callback cb; }void SetSrvClosedCallback(const ClosedCallbackcb) { _server_closed_callback cb; }//连接建立就绪后进行channel回调设置启动读监控调用_connected_callbackvoid Established() {_loop-RunInLoop(std::bind(Connection::EstablishedInLoop, this));}//发送数据将数据放到发送缓冲区启动写事件监控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)));}//提供给组件使用者的关闭接口--并不实际关闭需要判断有没有数据待处理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));}//切换协议---重置上下文以及阶段性回调处理函数 -- 而是这个接口必须在EventLoop线程中立即执行//防备新的事件触发后处理的时候切换任务还没有被执行--会导致数据使用原协议处理了。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));}
};
