3.4Reactor服务器实现

1.connect封装

​ 每一个连接都要有一个文件描述符和输入输出缓冲区，还有读、写、异常处理的回调方法；

​ 还包括指向服务器的回指指针；

class connection;
class tcpserver;using func_t = std::function<void(std::shared_ptr<connection>)>;
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_ptr<tcpserver> 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.hpp"class connection;
class tcpserver;
uint32_t event_in = (EPOLLIN | EPOLLET);
uint32_t event_out = (EPOLLOUT | EPOLLET);
using func_t = std::function<void(std::shared_ptr<connection>)>;
static const uint16_t defaultport = 8080;class connection
{
public:connection(int sockfd, std::shared_ptr<tcpserver> 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_ptr<tcpserver> 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_ptr<connection> c_ptr = std::make_shared<connection>(sockfd, std::shared_ptr<tcpserver>(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_ptr<connection> 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_ptr<epoll> ep_ptr_;                                    // 创建epoll对象std::shared_ptr<Sock> listensock_ptr_;                             // 构建了监听套接字对象；std::unordered_map<int, std::shared_ptr<connection>> 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);
}