台州企业自助建站,网站建设活动计划,怎么在网上开店卖东西,开发菏泽网站建设基于LT模式修改#xff0c;并加入前面的应用层计算器#xff0c;实现稍完整的服务器功能 1.修改tcp_socket.hpp#xff0c;新增非阻塞读和非阻塞写接口 2.对于accept返回的new_sock加上EPOLLET这样的选项
注意#xff1a;此代码暂时未考虑listen_sock ET的情况#xff0c…基于LT模式修改并加入前面的应用层计算器实现稍完整的服务器功能 1.修改tcp_socket.hpp新增非阻塞读和非阻塞写接口 2.对于accept返回的new_sock加上EPOLLET这样的选项
注意此代码暂时未考虑listen_sock ET的情况如果将listen_sock设为ET则需要非阻塞轮询的方式accept否则会导致同一时刻大量的客户端同时连接的情况只能accept一次的问题
目录
整体结构流程运行示例全改进Reactor的理论 1. 整体结构 TpcServer服务器类维护_listen套接字用来获取连接和监听读写事件map用套接字做键值rev数组作为epoll_wait的输出参数
每一个连接都是一个session结构包含读写缓冲区ip和端口方便调试读写错误的回调处理函数指针回指服务器 nocopy类用来给某些类提供无法拷贝的功能 socket类提供套接字的创建监听等功能 协议类和计算类为前面章节的内容用来对收到的数据处理返回结果 Epoll类提供epoll多路转接功能 comm类单独拎出来的设置非阻塞功能因为很多地方都要用到
2. 流程
服务端
TpcServer.hpp 继承enable_shared_from_this类可以解决智能指针不能用this构造的问题使用智能指针对象需要用shared_from_this()功能来获取 定义两个函数模板构造时传入报文处理回调 Init函数初始化套接字设置非阻塞AddConnection函数添加listen套接字到关注事件中绑定事件分配函数Accepter
AddConnection函数参数传入要设置的套接字事件三个函数ip和端口用来调试 Connection 构造时传入sock初始化成员变量 作为连接管理类需要管理每个连接的发送和接收缓冲区所以提供存入缓冲区数据的和返回缓冲区内容的功能再提供初始化自己成员函数的功能 TpcServer.hpp 继续说明AddConnection函数这个函数的作用为每个连接初始化session添加关注事件和管理后面每个新链接都要用这个函数
构造一个Connection的临时对象 设置成员TpcServer和回调函数ip和port 添加对象到map结构里添加listen的事件listen关注读
Accpeter连接管理器函数参数是事件就绪的会话
不一定只有一个连接到来所以需要循环读取。用accept获取就绪连接设置非阻塞后调用AddConnection函数加入会话管理作为连接会话三个回调函数分别是读写错误
当错误码是EWOULDBLOCK的时候说明已经获取完退出循环EINTR表示系统调用被信号中断所以继续读取其他情况退出
Recver数据读取函数用来提供读取数据添加到Connection缓冲区的功能
首先判断了连接的生命周期如果消亡就退出。通过lock获取一个shared指针对象。 因为是ET模式所以一次性需要读完所有数据用recv函数返回值n大于0表示读取到数据添加到接收缓冲区中等于0对方客户度退出调用错误处理函数小于0和上面一样判断是否读完不是就走错误处理
最后将读取到的数据交给处理函数所有报文情况都由它处理
Sender函数获取连接的发送缓冲区发送一次性将数据都发送返回值大于0发送成功将发送了的内容删除判断如果发送缓冲区为空就退出。0表示没发送任何内容也退出其他情况判断是否走错误处理
epoll/select/poll因为写事件经常都是就绪的发送缓冲区基本会有空间如果设置了写关心每次都会就绪经常返回浪费cpu资源。所以对于读需要设置常关心写按需求设置 当发送完后检查缓冲区不为空没发送完就对写事件开启关心发送完将事件关闭
EnableEvent函数设置套接字的读和写根据传入的参数判断有没有读和写通过三木运算符有就加入event最后修改套接字的事件 Excepter函数错误处理函数遇到错误就是关闭这个链接。如果连接在读和写时发生错误用这个函数。取消这个套接字的所有关心关闭文件map中移除
IsSafeConnection函数检查链接是否合法遍历map是否存在
主逻辑 Loop函数服务器的运行循环传入超时时间不断调用事件分配函数和打印连接函数
PrintConnection函数打印出map中所有的fd用来调试
Dispatcher函数timeout等待时间是上一个函数传入。不断wait监听revs数组添加了的套接字n会返回就绪的个数取到套接字和事件将异常转为读写统一处理。如果是读事件就绪并且连接合法就调用读取函数写事件调用写函数 TpcServer.cc 全局的计算类对象DefaultOnMessage函数是默认的报文处理函数对报文的完整性判断计算返回结果并发送 调用计算类的函数判断返回的字符串是否为空为空说明报文不完整或有错误。如果处理完成将结果加入到发送缓冲区用tcpserver对象发送 main函数创建svr对象传入报文处理函数启动服务器
客户端
客户端链接服务器生成5个随机报文发送接收结果打印 是前面章节的网络计算器 网络计算器
3. 运行示例 4. 全
TcpServer.hpp
#pragma ocne
#include iostream
#include memory
#include functional
#include unordered_map
#include Comm.hpp
#include log.hpp
#include Epoll.hpp
#include Socket.hppclass Connection;
class TpcServer;
using func_t std::functionvoid(std::weak_ptrConnection); // 用户缓冲区处理函数模板
using except_func_t std::functionvoid(std::weak_ptrConnection);
static const uint16_t port 8000;
static const int g_buff_size 128;
// 设置et
uint32_t EVENT_IN (EPOLLIN | EPOLLET);
uint32_t EVENt_OUT (EPOLLOUT | EPOLLET);class Connection
{
public:Connection(int sock){_sock sock;}~Connection(){}void AppendInbuff(const std::string message){_inbuff message;}void AppendOutbuff(const std::string message){_outbuff message;}int Fd(){return _sock;}std::string Inbuffer() // for debug{return _inbuff;}std::string Outbuffer() // for debug{return _outbuff;}void SetHandler(func_t recv_cb, func_t send_cb, except_func_t except_cb){_recv_cb recv_cb;_send_cb send_cb;_except_cb except_cb;}void SetWeakPtr(std::weak_ptrTpcServer tcp_setver_ptr){_tcp_server_ptr tcp_setver_ptr;}private:int _sock;std::string _inbuff; // string不能二进制, 需要vectorstd::string _outbuff;
public:func_t _recv_cb;func_t _send_cb;except_func_t _except_cb;std::weak_ptrTpcServer _tcp_server_ptr; // 回指向服务器std::string _ip;uint16_t _port;
};class TpcServer :public std::enable_shared_from_thisTpcServer, public nocopy
{static const int num 64;public:TpcServer(func_t OnMessage): _listensocket_ptr(new Sock()), _epoll_ptr(new Epoll()), _OnMessage(OnMessage), _quit(true){}void AddConnection(int sock, uint32_t event, func_t recv_cb, func_t send_cb, \except_func_t except_cb, const std::string ip 0.0.0.0, uint16_t port 0){// 1. 给sock创建connection对象, 将lstensock添加到connection中// 同时,listeinsock和connection放入_connectionsstd::shared_ptrConnection new_con(new Connection(sock));new_con-SetWeakPtr(shared_from_this()); // 返回当前对象的shared_ptrnew_con-SetHandler(recv_cb, send_cb, except_cb);new_con-_ip ip;new_con-_port port;// 2. 添加到map_connections.insert(std::make_pair(sock, new_con));// 3. 添加对应事件_epoll_ptr-EpollUpdate(EPOLL_CTL_ADD, sock, event);}void Init(){_listensocket_ptr-Socket();int opt 1;setsockopt(_listensocket_ptr-Fd(), SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, opt, sizeof(opt));// et模式需要非阻塞SetNonBlock(_listensocket_ptr-Fd());lg.logmessage(info, listensock create success:%d, _listensocket_ptr-Fd());_listensocket_ptr-Bind(port);_listensocket_ptr-Listen();// 关联connectionAddConnection(_listensocket_ptr-Fd(), EVENT_IN,std::bind(TpcServer::Accepter, this, std::placeholders::_1), nullptr, nullptr);}void Accepter(std::weak_ptrConnection con){// 获取强引用对象, 检查是否销毁auto connection con.lock();// 获取新链接while (true){struct sockaddr_in peer;socklen_t len sizeof(peer);// ::调用原生函数int sock ::accept(connection-Fd(), (struct sockaddr *)peer, len);if (sock 0){char ipbuf[128];inet_ntop(AF_INET, peer.sin_addr.s_addr, ipbuf, sizeof(ipbuf));uint16_t port ntohs(peer.sin_port);lg.logmessage(info, get a new clinet[%s:%d]:%d, ipbuf, port, sock);// 设置非阻塞SetNonBlock(sock);// 添加连接事件AddConnection(sock, EVENT_IN,\std::bind(TpcServer::Recver, this, std::placeholders::_1),\std::bind(TpcServer::Sender, this, std::placeholders::_1),\std::bind(TpcServer::Excepter, this, std::placeholders::_1),\ipbuf, port);}else {if (errno EWOULDBLOCK){break;}else if (errno EINTR) // 信号中断{continue;}else{break;}}}}void Recver(std::weak_ptrConnection con){if (con.expired())return;auto connec con.lock();int sock connec-Fd();while (true){char buff[g_buff_size];memset(buff, 0, sizeof(buff));ssize_t n recv(sock, buff, sizeof(buff) - 1, 0); // 非阻塞读取if (n 0){connec-AppendInbuff(buff);}else if (n 0) // 错误处理{lg.logmessage(info, sockfd:%d, client[%s:%d] quit, sock, connec-_ip.c_str(), connec-_port);connec-_except_cb(connec);return;}else{if (errno EWOULDBLOCK) // 读完{break;}else if (errno EINTR){continue;}else{lg.logmessage(warning, sockfd:%d, client[%s:%d] recv error, sock, connec-_ip.c_str(), connec-_port);connec-_except_cb(connec);return;}}}// 数据有了, 不一定安全 1.检测 2.如果有完整报文,处理_OnMessage(connec);}void Sender(std::weak_ptrConnection con){if (con.expired())return;auto connection con.lock();auto outbuff connection-Outbuffer();while (true){ssize_t n send(connection-Fd(), outbuff.c_str(), outbuff.size(), 0);if (n 0){outbuff.erase(0, n);if (outbuff.empty()){break;}}else if (n 0){return; // 没有发}else{if (errno EWOULDBLOCK){break;}else if (errno EINTR){continue;}else{lg.logmessage(info, sockfd:%d, client[%s:%d] recv error, connection-Fd(), connection-_ip.c_str(), connection-_port);connection-_except_cb(connection);return;}}}// 没发完, 开启对写事件关心if (!outbuff.empty()){EnableEvent(connection-Fd(), true, true);}else{EnableEvent(connection-Fd(), true, false);}}void EnableEvent(int sock, bool readable, bool writeable){uint32_t evnet 0;evnet | ((readable ? EPOLLIN : 0) | (writeable ? EPOLLOUT : 0) | EPOLLET);_epoll_ptr-EpollUpdate(EPOLL_CTL_MOD, sock, evnet);}void Excepter(std::weak_ptrConnection con){if (con.expired())return;auto connection con.lock();int fd connection-Fd();lg.logmessage(warning, Excepter handler:%d, client[%s:%d] excepter error, connection-Fd(), connection-_ip.c_str(), connection-_port);// 1. 移除关心_epoll_ptr-EpollUpdate(EPOLL_CTL_DEL, fd, 0);// 2. 关闭文件lg.logmessage(debug, close %d..., fd);close(fd);// 3. unordered_map中移除lg.logmessage(debug, remove connection %d, fd);_connections.erase(fd);}bool IsSafeConnection(int sock){auto it _connections.find(sock);if (it _connections.end()){return false;}else{return true;} }void Dispatcher(int timeout){int n _epoll_ptr-EpollWait(_revs, num, timeout);for (int i 0; i n; i){int sock _revs[i].data.fd;uint32_t event _revs[i].events;// 统一将异常转换为读写问题if (event EPOLLERR){event | (EPOLLIN | EPOLLOUT);}if (event EPOLLHUP){event | (EPOLLIN | EPOLLOUT);}if ((event EPOLLIN) IsSafeConnection(sock)){if (_connections[sock]-_recv_cb){_connections[sock]-_recv_cb(_connections[sock]);}}if ((event EPOLLOUT) IsSafeConnection(sock)){if (_connections[sock]-_send_cb){_connections[sock]-_send_cb(_connections[sock]);}}}}void Loop(){_quit false;while (!_quit){Dispatcher(3000);PrintConnection();}_quit true;}void PrintConnection(){std::cout _connection list: ;for (auto con: _connections){std::cout con.second-Fd() ,;}std::cout std::endl;}private:std::shared_ptrSock _listensocket_ptr; // 监听socket, 可以移到外部std::shared_ptrEpoll _epoll_ptr; // 内核std::unordered_mapint, std::shared_ptrConnection _connections;struct epoll_event _revs[num];func_t _OnMessage;bool _quit;
};TcpServer.cc
#include memory
#include TpcServer.hpp
#include Calculator.hppCalculator calculator;
void DefaultOnMessage(std::weak_ptrConnection con)
{if(con.expired()) return;auto connection_ptr con.lock();std::cout connection_ptr-Inbuffer() std::endl;std::string response_str calculator.Handler(connection_ptr-Inbuffer()); // 业务逻辑简单,如果复杂,需要拿到结果单独线程处理if (response_str.empty()){return;}lg.logmessage(debug, %s, response_str.c_str());connection_ptr-AppendOutbuff(response_str);//connection_ptr-_send_cb(connection_ptr); auto tcpserver connection_ptr-_tcp_server_ptr.lock(); tcpserver-Sender(connection_ptr);
}int main()
{std::shared_ptrTpcServer svr(new TpcServer(DefaultOnMessage));svr-Init();svr-Loop();return 0;
}Clinet.cc
#include time.h
#include unistd.h
#include assert.h
#include Socket.hpp
#include Protocol.hppint main()
{srand(time(NULL));std::cout 准备连接 std::endl;uint16_t serverport 8000;string serverip 106.54.46.147;struct sockaddr_in server;bzero(server, sizeof(server));server.sin_family AF_INET;server.sin_addr.s_addr inet_addr(serverip.c_str());server.sin_port htons(serverport);const string opers -*/%^;Sock socket;socket.Socket();bool r socket.Connect(serverip, serverport);if (!r)return 1;std::cout 连接成功, 开始发送数据 std::endl;int cnt 1;while (cnt 5){std::cout 第 cnt 次测试.... std::endl;string package;int x rand() % 100;int y rand() % 100 1;char op opers[rand() % opers.size()];Request req(x, y, op);req.DebugPrint();req.Serialize(package);package Encode(package);std::cout package std::endl;write(socket._sockfd, package.c_str(), package.size());char buff[1024];int n read(socket._sockfd, buff, sizeof(buff));string inbuff_stream;if (n 0){buff[n] 0;inbuff_stream buff;std::cout inbuff_stream std::endl;string content;bool r Decode(inbuff_stream, content);assert(r);Response resp;r resp.Deserialize(content);assert(r);resp.DebugPrint();}std::cout std::endl;sleep(1);cnt;}socket.Close();return 0;}
Comm.hpp
#pragma once
#include fcntl.h
#include unistd.h
#include Socket.hppvoid SetNonBlock(int sock)
{int f1 fcntl(sock, F_GETFL);if (f1 0){exit(NONBLOCKERR);}fcntl(sock, F_SETFL, f1 | O_NONBLOCK);
}Epoll.hpp
#pragma once
#include sys/epoll.h
#include nocopy.hpp
#include log.hppclass Epoll : public nocopy
{static const int size 128;Log log;public:Epoll(){_epfd epoll_create(size);if (_epfd -1){log.logmessage(ERROR, epoll create error:%s, strerror(errno));}else{log.logmessage(info, epoll create success:%d, _epfd);}}int EpollWait(struct epoll_event revents[], int num, int timeout){int n epoll_wait(_epfd, revents, num, timeout);return n;}int EpollUpdate(int oper, int sock, uint32_t event){int n 0;if (oper EPOLL_CTL_DEL){n epoll_ctl(_epfd, oper, sock, nullptr);if (n ! 0){log.logmessage(ERROR, epoll_ctl delete error);}}else{struct epoll_event ev;ev.events event;ev.data.fd sock;n epoll_ctl(_epfd, oper, sock, ev);if (n ! 0){log.logmessage(ERROR, epoll_ctl add error);}}}~Epoll(){if (_epfd 0){close(_epfd);}}private:int _epfd;int _timeout{3000};
};nocopy.hpp
#pragma onceclass nocopy
{
public:nocopy(){}nocopy(const nocopy ) delete;nocopy operator(const nocopy) delete;
};Socket.hpp
#pragma once
#include sys/types.h
#include sys/socket.h
#include netinet/in.h
#include arpa/inet.h
#include unistd.h
#include stdlib.h
#include fcntl.h
#include log.hppenum
{SOCKERR 1,BINDERR,LISERR,NONBLOCKERR
};Log lg;
const int backlog 5;
class Sock
{
public:Sock(){}void Socket(){_sockfd socket(AF_INET, SOCK_STREAM, 0);if (_sockfd 0){lg.logmessage(fatal, socket error);exit(SOCKERR);}int opt 1;setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, opt, sizeof(opt)); // 防止偶发性的服务器无法进行立即重启(tcp协议的时候再说)}void Bind(uint16_t port){struct sockaddr_in local;memset(local, 0, sizeof(local));local.sin_family AF_INET;local.sin_addr.s_addr INADDR_ANY;local.sin_port htons(port);int bret bind(_sockfd, (const struct sockaddr*)local, sizeof(local));if (bret 0){lg.logmessage(fatal, bind error);exit(BINDERR);}}void Listen(){int lret listen(_sockfd, backlog);if (lret 0){lg.logmessage(fatal, listen error);exit(LISERR);}}int Accept(string* clientip, uint16_t* clientport){sockaddr_in peer;socklen_t len sizeof(peer);int newfd accept(_sockfd, (sockaddr*)peer, len);if (newfd 0){lg.logmessage(warning, accept error);return -1;}char ipstr[64];inet_ntop(AF_INET, peer.sin_addr, ipstr, sizeof(ipstr));*clientip ipstr;*clientport ntohs(peer.sin_port);return newfd;}bool Connect(const string ip, const uint16_t port){sockaddr_in peer;memset(peer, 0, sizeof(peer));peer.sin_family AF_INET;inet_pton(AF_INET, ip.c_str(), peer.sin_addr);peer.sin_port htons(port);int cret connect(_sockfd, (const struct sockaddr*)peer, sizeof(peer));if (cret -1){lg.logmessage(warning, connect error);return false;}return true;}void Close(){close(_sockfd);}int Fd(){return _sockfd;}~Sock(){}
public:int _sockfd;
};
Protocol.hpp
#pragma once
#include string
#include jsoncpp/json/json.h//#define MYSELF 1
//分隔符
const std::string black_sep ;
const std::string protocol_sep \n;//解决报文外部格式//len\n正文\nstd::string Encode(std::string message){std::string package std::to_string(message.size());package protocol_sep;package message;package protocol_sep;return package;}//len\na b\nbool Decode(std::string message, std::string* content){std::size_t pos message.find(protocol_sep);if (pos std::string::npos){return false;}std::string len_str message.substr(0, pos);std::size_t len std::stoi(len_str);std::size_t total_len len_str.size() len 2;//检查长度if (message.size() total_len){return false;}*content message.substr(pos 1, len);//earse 移除报文message.erase(0, total_len);return true;}class Request
{
public:Request(){}Request(int a, int b, char oper){_num1 a;_num2 b;_op oper;}//a bbool Serialize(std::string* out){
#ifdef MYSELF//构建报文有效载荷std::string str;str std::to_string(_num1);str black_sep;str _op;str black_sep;str std::to_string(_num2);*out str;return true;
#elseJson::Value root;root[x] _num1;root[y] _num2;root[op] _op;Json::FastWriter w;*out w.write(root);return true;
#endif}//a bbool Deserialize(std::string in){
#ifdef MYSELF//astd::size_t left in.find(black_sep);if (left std::string::npos){return false; }std::string part_a in.substr(0, left);// bstd::size_t right in.rfind(black_sep);if (right std::string::npos){return false; }std::string part_b in.substr(right 1);//if (left 2 ! right){return false;}_op in[left1];_num1 std::stoi(part_a);_num2 std::stoi(part_b);return true;
#else Json::Value root;Json::Reader r;r.parse(in, root);_num1 root[x].asInt();_num2 root[y].asInt();_op root[op].asInt();return true;
#endif}void DebugPrint(){std::cout 新请求构建完成: _num1 _op _num2 std::endl;}public:int _num1;int _num2;char _op;
};class Response
{
public:Response(){}Response(int res, int cod){_result res;_code cod;}//1000 0bool Serialize(std::string* out){
#ifdef MYSELFstring str std::to_string(_result);str black_sep;str std::to_string(_code);*out str;return true;
#else Json::Value root;root[res] _result;root[code] _code;Json::FastWriter w;*out w.write(root);return true;
#endif}//1000 0bool Deserialize(std::string in){
#ifdef MYSELFstd::size_t pos in.find(black_sep);if (pos std::string::npos){return false;}std::string left in.substr(0, pos);std::string right in.substr(pos 1);_result std::stoi(left);_code std::stoi(right);return true;
#else Json::Value root;Json::Reader r;r.parse(in, root);_result root[res].asInt();_code root[code].asInt();return true;
#endif}void DebugPrint(){std::cout 结果响应完成,result: _result ,code: _code std::endl;}public:int _result;int _code; //0可信,否则表明对应的错误
};#define MySelf 1
Calcluator.hpp
#pragma once
#include Protocol.hppenum
{DIVZERO 1,MODZERO,OTHER_OPER
};class Calculator
{
public:Calculator(){}Response CalculatorHelp(const Request req){Response res(0, 0);switch (req._op){case :res._result req._num1 req._num2;break;case -:res._result req._num1 - req._num2;break;case *:res._result req._num1 * req._num2;break;case /:if (req._num2 0){res._code DIVZERO;}else{res._result req._num1 / req._num2;}break;case %:if (req._num2 0){res._code MODZERO;}else{res._result req._num1 % req._num2;break;}default:res._code OTHER_OPER;break;}return res;}std::string Handler(std::string package){std::string content;bool r Decode(package, content);if (!r){return ;}Request req;r req.Deserialize(content);if (!r){return ;}req.DebugPrint();content ;Response res CalculatorHelp(req);res.DebugPrint();res.Serialize(content);content Encode(content); // len\n正文\nreturn content;}~Calculator(){}
};
Log.hpp
#pragma once
#include stdarg.h
#include iostream
#include stdio.h
#include cstring
#include time.h
#include cerrno
#include sys/types.h
#include sys/stat.h
#include fcntl.h
#include unistd.husing namespace std;#define info 0
#define debug 1
#define warning 2
#define ERROR 3
#define fatal 4#define screen 1
#define onefile 2
#define classfile 3#define PATH log.txtclass Log
{
public:Log(int style screen){printstyle style;dir log/;}void enable(int method){printstyle method;}const char *leveltostring(int level){switch (level){case 0:return info;break;case 1:return debug;break;case 2:return warning;break;case 3:return error;break;case 4:return fatal;break;default:return none;break;}}void printlog(int level, const string logtxt){switch (printstyle){case screen:cout logtxt;break;case onefile:printonefile(PATH, logtxt);break;case classfile:printclassfile(level, logtxt);break;}}void logmessage(int level, const char *format, ...){time_t t time(0);tm *ctime localtime(t);char leftbuff[1024];sprintf(leftbuff, [%s]%d-%d-%d %d:%d:%d:, leveltostring(level), ctime-tm_year 1900,ctime-tm_mon 1, ctime-tm_mday, ctime-tm_hour, ctime-tm_min, ctime-tm_sec);char rightbuff[1024];va_list s;va_start(s, format);vsprintf(rightbuff, format, s);va_end(s);char logtext[2048];sprintf(logtext, %s %s\n, leftbuff, rightbuff);//printf(logtext);printlog(level, logtext);}void printonefile(const string logname, const string logtxt){int fd open(logname.c_str(), O_WRONLY | O_CREAT | O_APPEND, 0666); if (fd 0){return;}write(fd, logtxt.c_str(), logtxt.size());close(fd);}void printclassfile(int level, const string logtxt){//log.txt.infostring filename dir PATH;filename .;filename leveltostring(level);printonefile(filename, logtxt);}~Log(){};private:int printstyle;string dir; //分类日志,放入目录中
};// int sum(int n, ...)
// {
// int sum 0;
// va_list s;
// va_start(s, n);// while (n)
// {
// sum sum va_arg(s, int);
// n--;
// }// return sum;
// }CMakeLists.txt
cmake_minimum_required(VERSION 2.8.12.2)
project(TpcServer)
set(CMAKE_CXX_FLAGS ${CMAKE_CXX_FLAGS} -stdc11)
set(CMAKE_CXX_STANDARD_REQUIRED ON) # 确保必须使用指定的标准add_executable(TpcServer TpcServer.cc)
target_link_libraries(TpcServer jsoncpp)add_executable(Client Client.cc)
target_link_libraries(Client jsoncpp)
5. 改进
listen分离
将TpcServer里的listen套接字提出来单独成一个类将Accpeter获取连接的函数放到这个类通过AddCpnnection函数加入关注这个类就做listen的连接管理。这样做的好处是可以将listen连接也作为一个正常连接统一处理
后续的设计可以加入多线程主线程维护vector里面是获得的连接fd线程之间通信争取这个fd会话每一个线程也是一个Ractor也可以负载均衡式的分配连接。这样就可以加入多线程同时每个连接的报文内容处理如果比较复杂可以交由线程池来处理只需要拿到结果 这种一个连接一个reacotr叫one thread one loop
连接管理
对于一些不活跃的连接需要处理。使用一个定时器类里面保存一个最小堆存每个连接的超时时间可以回指connection和tpcserver做更多开发。在主循环每次事件分配后可以做一些其他事情就是查询超时连接如果有可以走错误处理关闭这个链接。同时事件分配的等待时间可以设置为堆顶的超时时间
6. Reactor的理论
它是一个半同步半异步的模型类似于打地鼠监测哪个链接有事件就处理哪个。同步体现在事件的就绪是需要等异步体现在回调函数如果不想自己做可以交由线程处理比如报文的处理可以由线程来直接取得结果。这种叫反应堆
