网站建设及规划,龙岗区建设工程交易服务网,电子产品外贸交易平台,网站公司上海3.4Reactor服务器实现
1.connect封装
每一个连接都要有一个文件描述符和输入输出缓冲区#xff0c;还有读、写、异常处理的回调方法#xff1b;
还包括指向服务器的回指指针#xff1b;
class connection;
class tcpserver;using func_t std::functionvoid(s…3.4Reactor服务器实现
1.connect封装
每一个连接都要有一个文件描述符和输入输出缓冲区还有读、写、异常处理的回调方法
还包括指向服务器的回指指针
class connection;
class tcpserver;using func_t std::functionvoid(std::shared_ptrconnection);
class connection
{
public:connection() {}~connection() {}private:int sockfd_; // 文件描述符std::string inbuffer_; // 输入缓冲区string缺陷就是不可以处理二进制流std::string outbuffer_; // 输出缓冲区func_t rcb_; // 设置回调函数将IO方法交给上层func_t wcb_; // 写回调func_t ecb_; // 异常回调std::shared_ptrtcpserver tsvr_; // 添加一个回指指针
};2.tcpserver封装
设置epoll对象(用来将连接的属性设置进内核)监听套接字文件描述符和连接的映射并用智能指针管理
设置nocopy对象不允许服务器拷贝
通过使用EPOLLET事件实现ET工作模式并且非阻塞文件描述符
使用addconnection函数实现管理所有的文件描述符
epoll_event结构体数组来获取就绪的事件
#include iostream
#include string
#include memory
#include functional
#include unordered_map
#include fcntl.h
#include unistd.h
#include Socket.hpp
#include Log.hpp
#include nocopy.hpp
#include epoll.hpp
#include Socket.hpp
#include comm.hppclass connection;
class tcpserver;
uint32_t event_in (EPOLLIN | EPOLLET);
uint32_t event_out (EPOLLOUT | EPOLLET);
using func_t std::functionvoid(std::shared_ptrconnection);
static const uint16_t defaultport 8080;class connection
{
public:connection(int sockfd, std::shared_ptrtcpserver tsvr) : sockfd_(sockfd), tsvr_(tsvr) {}~connection() {}public:void sethandler(func_t rcb, func_t wcb, func_t ecb){rcb_ rcb;wcb_ wcb;ecb_ ecb;}int Fd(){return sockfd_;}private:int sockfd_; // 文件描述符std::string inbuffer_; // 输入缓冲区string缺陷就是不可以处理二进制流std::string outbuffer_; // 输出缓冲区
public:func_t rcb_ nullptr; // 设置回调函数将IO方法交给上层func_t wcb_ nullptr; // 写回调func_t ecb_ nullptr; // 异常回调std::shared_ptrtcpserver tsvr_; // 添加一个回指指针
};class tcpserver : public nocopy
{static const int num 64;public:tcpserver(uint16_t port defaultport): ep_ptr_(new epoll()), listensock_ptr_(new Sock()), port_(port), quit_(true){}~tcpserver(){listensock_ptr_-Close();}public:void init(){listensock_ptr_-Socket();setnonblock(listensock_ptr_-Fd());listensock_ptr_-Bind(port_);listensock_ptr_-Listen();lg(Info, create listen socket success, listensockfd: %d, listensock_ptr_-Fd());addconnection(listensock_ptr_-Fd(), event_in, std::bind(tcpserver::Accept, this, std::placeholders::_1), nullptr, nullptr);}void addconnection(int sockfd, uint32_t event, func_t rcb, func_t wcb, func_t ecb){// 1.建立sockfd对应connection对象std::shared_ptrconnection c_ptr std::make_sharedconnection(sockfd, std::shared_ptrtcpserver(this));c_ptr-sethandler(rcb, wcb, ecb);// 2.将connection添加到映射表中connections_[sockfd] c_ptr;// 3.将fd与事件添加到内核中ep_ptr_-Update(EPOLL_CTL_ADD, sockfd, event);lg(Debug, add a new connection, sockfd: %d, sockfd);}void Accept(std::shared_ptrconnection connection_ptr){while (true){sockaddr_in client;socklen_t len sizeof(client);int sockfd accept(connection_ptr-Fd(), (sockaddr *)client, len);if (sockfd 0){uint16_t clientport ntohs(client.sin_port);char ipstr[64];inet_ntop(AF_INET, client.sin_addr, ipstr, sizeof(ipstr));std::string clientip ipstr;lg(Debug, get a new client, ip: %s, port: %d, sockfd: %d, clientip.c_str(), clientport, sockfd);setnonblock(sockfd);addconnection(sockfd, event_in, nullptr, nullptr, nullptr);}else{if (errno EWOULDBLOCK){break;}else if (errno EINTR){continue;}else{break;}}}}bool IsConnectionSafe(int fd){auto iter connections_.find(fd);if (iter connections_.end()){return false;}return true;}void dispatcher(int timeout){int n ep_ptr_-Wait(r_events_, num, timeout);for (int i 0; i n; i){uint32_t event r_events_[i].events;int sockfd r_events_[i].data.fd;// 统一把异常问题转化成读写问题if (event EPOLLERR){event | (EPOLLIN | EPOLLOUT);}if (event EPOLLHUP){event | (EPOLLIN | EPOLLOUT);}// 处理读写if (IsConnectionSafe(sockfd) (event event_in)){if (connections_[sockfd]-rcb_){connections_[sockfd]-rcb_(connections_[sockfd]);}}if (IsConnectionSafe(sockfd) (event event_out)){if (connections_[sockfd]-wcb_){connections_[sockfd]-wcb_(connections_[sockfd]);}}}}void printconnection(){std::cout connections fds: std::endl;for (auto e : connections_){std::cout e.second-Fd() ;}std::cout std::endl;}void loop(){quit_ false;while (!quit_){dispatcher(3000);printconnection();}quit_ true;}private:std::shared_ptrepoll ep_ptr_; // 创建epoll对象std::shared_ptrSock listensock_ptr_; // 构建了监听套接字对象std::unordered_mapint, std::shared_ptrconnection connections_; // 构建了文件描述符和连接对象的映射epoll_event r_events_[num];uint16_t port_;bool quit_;
};
void setnonblock(int fd)
{int fl fcntl(fd, F_GETFL);if (fl 0){exit(NON_BLOCK);}fcntl(fd, F_SETFL, fl | O_NONBLOCK);
}
