)
文章目录回顾Reactor_v1实现Socket类实现头文件源文件InetAddr类的实现头文件源文件Acceptor类的实现头文件源文件SocketIO类的实现头文件源文件TcpConnection类的实现头文件源文件回顾先回顾一下我们整个的实现框架所以第一步应该是实现的是——封装系统调用另外为了方便打印调试信息可以把日志系统用单例模式先实现出来。关于用单例模式实现日志系统我们在之前的文章中已经实现过了这里就不过多赘述了。详情可以跳转到该文章中查看Reactor网络模型本质上就是使用epoll的并发网络模型通过连接器连接到服务器网络然后注册三个事件(对应epoll的三个接口函数)通过事件驱动(事件循环和事件分发)来响应客户端的请求。Reactor_v1实现在写代码之前要整理一下思绪框架图中已经很明确的写出了有哪些类我们在第一个版本的内容中实际上就是对系统调用的封装下面就画一个类图整理一下具体实现的内容。Acceptor类中可以将地址复用、端口复用、bind、listen这些写法固定的流程封装在一个ready函数中。关于其它类行该可以比较容易的理清思路。需要着重考虑的是IO操作因为使用send/recv或write/read一次处理的字符数量是由上限的。所以我们应该设置一个合适大小的缓冲用来接收数据流。Socket类实现头文件/** * 封装文件描述符 */#ifndef_SOCKET_H#define_SOCKET_H#includesys/types.h#includesys/socket.hclassSocket{public:Socket();/** * param fd * 禁止隐式转换防止出现问题 */explicitSocket(intfd);~Socket();intgetFd();private:int_fd;};#endif//_SOCKET_H源文件/** * 封装文件描述符 */#includeSocket.h#includestdio.h#includestdlib.h/** * Socket implementation */Socket::Socket(){_fd::socket(AF_INET,SOCK_STREAM,0);if(_fd-1){perror(socket error);exit(EXIT_FAILURE);}}/** * param fd */Socket::Socket(intfd):_fd(fd){}Socket::~Socket(){}/** * return int */intSocket::getFd(){return_fd;}InetAddr类的实现头文件/** * 地址相关操作 */#ifndef_INETADDRESS_H#define_INETADDRESS_H#includestring#includenetinet/in.h#includenetinet/ip.h#includesys/socket.h#includearpa/inet.husingstd::string;classInetAddress{public:/** * param ip * param port */InetAddress(conststringip,unsignedshortport);/** * param addr */InetAddress(conststructsockaddr_inaddr);~InetAddress();stringgetIp();unsignedshortgetPort();conststructsockaddr_in*getInetAddressPtr();private:structsockaddr_in_addr;};#endif//_INETADDRESS_H源文件/** * 地址相关操作 */#includeInetAddress.h/** * InetAddress implementation *//** * param ip * param port */InetAddress::InetAddress(conststringip,unsignedshortport){_addr.sin_familyAF_INET;_addr.sin_porthtons(port);_addr.sin_addr.s_addr::inet_addr(ip.c_str());}/** * param addr */InetAddress::InetAddress(conststructsockaddr_inaddr):_addr(addr){}InetAddress::~InetAddress(){}/** * return string */stringInetAddress::getIp(){string temp::inet_ntoa(_addr.sin_addr);returntemp;}/** * return unsigned short */unsignedshortInetAddress::getPort(){return::ntohs(_addr.sin_port);}/** * return const struct sockaddr_in * */conststructsockaddr_in*InetAddress::getInetAddressPtr(){return_addr;}Acceptor类的实现头文件/** * 连接器的实现 */#ifndef_ACCEPTOR_H#define_ACCEPTOR_H#includeInetAddress.h#includeSocket.husingstd::string;classAcceptor{public:/** * param ip * param port */Acceptor(conststringip,unsignedshortport);~Acceptor();voidready();intaccept();private:Socket _sock;InetAddress _addr;voidsetReuseAddr();voidsetReusePort();voidbind();voidlisten();};#endif//_ACCEPTOR_H源文件/** * 连接器的实现 */#includeAcceptor.h/** * Acceptor implementation *//** * param ip * param port */Acceptor::Acceptor(conststringip,unsignedshortport):_sock(),_addr(ip,port){}Acceptor::~Acceptor(){}/** * return void */voidAcceptor::ready(){setReuseAddr();setReusePort();bind();listen();return;}/** * return int */intAcceptor::accept(){intclientfd::accept(_sock.getFd(),NULL,NULL);if(clientfd-1){perror(accept error);return-1;}returnclientfd;}/** * return void */voidAcceptor::setReuseAddr(){intopt1;if(::setsockopt(_sock.getFd(),SOL_SOCKET,SO_REUSEADDR,opt,sizeof(opt))-1){perror(reuse sockaddr error);exit(EXIT_FAILURE);}return;}/** * return void */voidAcceptor::setReusePort(){intopt1;if(::setsockopt(_sock.getFd(),SOL_SOCKET,SO_REUSEPORT,opt,sizeof(opt))-1){perror(reuse port error);exit(EXIT_FAILURE);}return;}/** * return void */voidAcceptor::bind(){if(::bind(_sock.getFd(),(structsockaddr*)_addr.getInetAddressPtr(),sizeof(structsockaddr))-1){perror(bind error);exit(EXIT_FAILURE);}return;}/** * return void */voidAcceptor::listen(){if(::listen(_sock.getFd(),100)-1){perror(listen error);exit(EXIT_FAILURE);}return;}SocketIO类的实现头文件/** * IO操作 */#ifndef_SOCKETIO_H#define_SOCKETIO_H#includeunistd.h#includeerrno.h#includestdlib.h#includestdio.h#includesys/socket.h#includesys/types.hclassSocketIO{public:/** * param fd */explicitSocketIO(intfd);~SocketIO();/** * param buf * param len */intreadn(char*buf,intlen);/** * param buf * param len */intreadLine(char*buf,intlen);/** * param buf * param len */intwriten(constchar*buf,intlen);private:int_fd;};#endif//_SOCKETIO_H源文件/** * IO操作 */#includeSocketIO.h/** * SocketIO implementation *//** * param fd */SocketIO::SocketIO(intfd):_fd(fd){}SocketIO::~SocketIO(){if(_fd!-1){close(_fd);_fd-1;}}/** * param buf * param len * return int */intSocketIO::readn(char*buf,intlen){intremainlen;// 剩余没读的字节intret0;char*pstrbuf;while(remain0){ret::read(_fd,pstr,sizeof(remain));if(ret-1errnoEINTR){::perror(read error);return-1;}elseif(ret0){break;}else{pstrret;remain-ret;}}returnlen-remain;}/** * param buf * param len * return int */intSocketIO::readLine(char*buf,intlen){intremainlen-1;intret0,total0;char*pstrbuf;while(remain0){ret::recv(_fd,pstr,remain,MSG_PEEK);if(ret-1errnoEINTR){::perror(readLine error);return-1;}elseif(ret0){break;}else{for(inti0;iret;i){if(pstr[i]\n){intszi1;readn(pstr,sz);pstrsz;*pstr\0;returnsztotal;}}readn(pstr,ret);totalret;pstrret;remain-ret;}}*pstr\0;returntotal;}/** * param buf * param len * return int */intSocketIO::writen(constchar*buf,intlen){intremainlen;constchar*pstrbuf;intret0;while(remain0){retwrite(_fd,pstr,remain);if(ret-1errnoEINTR){::perror(writen error);return-1;}elseif(ret0){break;}else{pstrret;remain-ret;}}returnlen-remain;}TcpConnection类的实现头文件/** * 连接管理 */#ifndef_TCPCONNECTION_H#define_TCPCONNECTION_H#includestring#includeSocketIO.husingstd::string;classTcpConnection{public:/** * param fd */explicitTcpConnection(intfd);~TcpConnection();stringreceive();/** * param msg */voidsend(conststringmsg);private:SocketIO _sockIO;};#endif//_TCPCONNECTION_H源文件/** * 连接管理 */#includeTcpConnection.h/** * TcpConnection implementation *//** * param fd */TcpConnection::TcpConnection(intfd):_sockIO(fd){}TcpConnection::~TcpConnection(){}/** * return string */stringTcpConnection::receive(){charbuffer[1024]{0};_sockIO.readLine(buffer,sizeof(buffer));returnstring(buffer);}/** * param msg * return void */voidTcpConnection::send(conststringmsg){_sockIO.writen(msg.c_str(),msg.size());return;}以上就是Reactor的全部函数完成了对系统调用的封装。下面再写一个测试文件#includeAcceptor.h#includeTcpConnection.h#includeiostreamusingstd::cout;usingstd::endl;voidtest0(){Acceptoracceptor(0.0.0.0,8080);acceptor.ready();TcpConnectioncon(acceptor.accept());while(1){/* cout recv msg from client: con.receive() endl; *//* con.send(hello,world\n); */string msgcon.receive();coutrecv msg from client: msgendl;con.send(msg:msg);}}intmain(){test0();return0;}下面再开一个终端自测一下基本上还是符合我们预期的也就是说明我们对于底层的封装没有了太大问题。我们的Reactor第一代基本完成了当然可以把之前封装的log4cpp的代码替代perror()。下面的工作就是完成核心层事件驱动的代码就是对epoll的封装另外TcpConnection是用来管理连接的所以当我们使用服务器采用事件驱动的形式时TcpConnection中的数据成员和成员函数也要随之变动。
文章目录回顾Reactor_v1实现Socket类实现头文件源文件InetAddr类的实现头文件源文件Acceptor类的实现头文件源文件SocketIO类的实现头文件源文件TcpConnection类的实现头文件源文件回顾先回顾一下我们整个的实现框架所以第一步应该是实现的是——封装系统调用另外为了方便打印调试信息可以把日志系统用单例模式先实现出来。关于用单例模式实现日志系统我们在之前的文章中已经实现过了这里就不过多赘述了。详情可以跳转到该文章中查看Reactor网络模型本质上就是使用epoll的并发网络模型通过连接器连接到服务器网络然后注册三个事件(对应epoll的三个接口函数)通过事件驱动(事件循环和事件分发)来响应客户端的请求。Reactor_v1实现在写代码之前要整理一下思绪框架图中已经很明确的写出了有哪些类我们在第一个版本的内容中实际上就是对系统调用的封装下面就画一个类图整理一下具体实现的内容。Acceptor类中可以将地址复用、端口复用、bind、listen这些写法固定的流程封装在一个ready函数中。关于其它类行该可以比较容易的理清思路。需要着重考虑的是IO操作因为使用send/recv或write/read一次处理的字符数量是由上限的。所以我们应该设置一个合适大小的缓冲用来接收数据流。Socket类实现头文件/** * 封装文件描述符 */#ifndef_SOCKET_H#define_SOCKET_H#includesys/types.h#includesys/socket.hclassSocket{public:Socket();/** * param fd * 禁止隐式转换防止出现问题 */explicitSocket(intfd);~Socket();intgetFd();private:int_fd;};#endif//_SOCKET_H源文件/** * 封装文件描述符 */#includeSocket.h#includestdio.h#includestdlib.h/** * Socket implementation */Socket::Socket(){_fd::socket(AF_INET,SOCK_STREAM,0);if(_fd-1){perror(socket error);exit(EXIT_FAILURE);}}/** * param fd */Socket::Socket(intfd):_fd(fd){}Socket::~Socket(){}/** * return int */intSocket::getFd(){return_fd;}InetAddr类的实现头文件/** * 地址相关操作 */#ifndef_INETADDRESS_H#define_INETADDRESS_H#includestring#includenetinet/in.h#includenetinet/ip.h#includesys/socket.h#includearpa/inet.husingstd::string;classInetAddress{public:/** * param ip * param port */InetAddress(conststringip,unsignedshortport);/** * param addr */InetAddress(conststructsockaddr_inaddr);~InetAddress();stringgetIp();unsignedshortgetPort();conststructsockaddr_in*getInetAddressPtr();private:structsockaddr_in_addr;};#endif//_INETADDRESS_H源文件/** * 地址相关操作 */#includeInetAddress.h/** * InetAddress implementation *//** * param ip * param port */InetAddress::InetAddress(conststringip,unsignedshortport){_addr.sin_familyAF_INET;_addr.sin_porthtons(port);_addr.sin_addr.s_addr::inet_addr(ip.c_str());}/** * param addr */InetAddress::InetAddress(conststructsockaddr_inaddr):_addr(addr){}InetAddress::~InetAddress(){}/** * return string */stringInetAddress::getIp(){string temp::inet_ntoa(_addr.sin_addr);returntemp;}/** * return unsigned short */unsignedshortInetAddress::getPort(){return::ntohs(_addr.sin_port);}/** * return const struct sockaddr_in * */conststructsockaddr_in*InetAddress::getInetAddressPtr(){return_addr;}Acceptor类的实现头文件/** * 连接器的实现 */#ifndef_ACCEPTOR_H#define_ACCEPTOR_H#includeInetAddress.h#includeSocket.husingstd::string;classAcceptor{public:/** * param ip * param port */Acceptor(conststringip,unsignedshortport);~Acceptor();voidready();intaccept();private:Socket _sock;InetAddress _addr;voidsetReuseAddr();voidsetReusePort();voidbind();voidlisten();};#endif//_ACCEPTOR_H源文件/** * 连接器的实现 */#includeAcceptor.h/** * Acceptor implementation *//** * param ip * param port */Acceptor::Acceptor(conststringip,unsignedshortport):_sock(),_addr(ip,port){}Acceptor::~Acceptor(){}/** * return void */voidAcceptor::ready(){setReuseAddr();setReusePort();bind();listen();return;}/** * return int */intAcceptor::accept(){intclientfd::accept(_sock.getFd(),NULL,NULL);if(clientfd-1){perror(accept error);return-1;}returnclientfd;}/** * return void */voidAcceptor::setReuseAddr(){intopt1;if(::setsockopt(_sock.getFd(),SOL_SOCKET,SO_REUSEADDR,opt,sizeof(opt))-1){perror(reuse sockaddr error);exit(EXIT_FAILURE);}return;}/** * return void */voidAcceptor::setReusePort(){intopt1;if(::setsockopt(_sock.getFd(),SOL_SOCKET,SO_REUSEPORT,opt,sizeof(opt))-1){perror(reuse port error);exit(EXIT_FAILURE);}return;}/** * return void */voidAcceptor::bind(){if(::bind(_sock.getFd(),(structsockaddr*)_addr.getInetAddressPtr(),sizeof(structsockaddr))-1){perror(bind error);exit(EXIT_FAILURE);}return;}/** * return void */voidAcceptor::listen(){if(::listen(_sock.getFd(),100)-1){perror(listen error);exit(EXIT_FAILURE);}return;}SocketIO类的实现头文件/** * IO操作 */#ifndef_SOCKETIO_H#define_SOCKETIO_H#includeunistd.h#includeerrno.h#includestdlib.h#includestdio.h#includesys/socket.h#includesys/types.hclassSocketIO{public:/** * param fd */explicitSocketIO(intfd);~SocketIO();/** * param buf * param len */intreadn(char*buf,intlen);/** * param buf * param len */intreadLine(char*buf,intlen);/** * param buf * param len */intwriten(constchar*buf,intlen);private:int_fd;};#endif//_SOCKETIO_H源文件/** * IO操作 */#includeSocketIO.h/** * SocketIO implementation *//** * param fd */SocketIO::SocketIO(intfd):_fd(fd){}SocketIO::~SocketIO(){if(_fd!-1){close(_fd);_fd-1;}}/** * param buf * param len * return int */intSocketIO::readn(char*buf,intlen){intremainlen;// 剩余没读的字节intret0;char*pstrbuf;while(remain0){ret::read(_fd,pstr,sizeof(remain));if(ret-1errnoEINTR){::perror(read error);return-1;}elseif(ret0){break;}else{pstrret;remain-ret;}}returnlen-remain;}/** * param buf * param len * return int */intSocketIO::readLine(char*buf,intlen){intremainlen-1;intret0,total0;char*pstrbuf;while(remain0){ret::recv(_fd,pstr,remain,MSG_PEEK);if(ret-1errnoEINTR){::perror(readLine error);return-1;}elseif(ret0){break;}else{for(inti0;iret;i){if(pstr[i]\n){intszi1;readn(pstr,sz);pstrsz;*pstr\0;returnsztotal;}}readn(pstr,ret);totalret;pstrret;remain-ret;}}*pstr\0;returntotal;}/** * param buf * param len * return int */intSocketIO::writen(constchar*buf,intlen){intremainlen;constchar*pstrbuf;intret0;while(remain0){retwrite(_fd,pstr,remain);if(ret-1errnoEINTR){::perror(writen error);return-1;}elseif(ret0){break;}else{pstrret;remain-ret;}}returnlen-remain;}TcpConnection类的实现头文件/** * 连接管理 */#ifndef_TCPCONNECTION_H#define_TCPCONNECTION_H#includestring#includeSocketIO.husingstd::string;classTcpConnection{public:/** * param fd */explicitTcpConnection(intfd);~TcpConnection();stringreceive();/** * param msg */voidsend(conststringmsg);private:SocketIO _sockIO;};#endif//_TCPCONNECTION_H源文件/** * 连接管理 */#includeTcpConnection.h/** * TcpConnection implementation *//** * param fd */TcpConnection::TcpConnection(intfd):_sockIO(fd){}TcpConnection::~TcpConnection(){}/** * return string */stringTcpConnection::receive(){charbuffer[1024]{0};_sockIO.readLine(buffer,sizeof(buffer));returnstring(buffer);}/** * param msg * return void */voidTcpConnection::send(conststringmsg){_sockIO.writen(msg.c_str(),msg.size());return;}以上就是Reactor的全部函数完成了对系统调用的封装。下面再写一个测试文件#includeAcceptor.h#includeTcpConnection.h#includeiostreamusingstd::cout;usingstd::endl;voidtest0(){Acceptoracceptor(0.0.0.0,8080);acceptor.ready();TcpConnectioncon(acceptor.accept());while(1){/* cout recv msg from client: con.receive() endl; *//* con.send(hello,world\n); */string msgcon.receive();coutrecv msg from client: msgendl;con.send(msg:msg);}}intmain(){test0();return0;}下面再开一个终端自测一下基本上还是符合我们预期的也就是说明我们对于底层的封装没有了太大问题。我们的Reactor第一代基本完成了当然可以把之前封装的log4cpp的代码替代perror()。下面的工作就是完成核心层事件驱动的代码就是对epoll的封装另外TcpConnection是用来管理连接的所以当我们使用服务器采用事件驱动的形式时TcpConnection中的数据成员和成员函数也要随之变动。
