絡(luò)編程（5）—— Reactor實現(xiàn)(v1))
文章目錄回顧Reactor_v1實現(xiàn)Socket類實現(xiàn)頭文件源文件InetAddr類的實現(xiàn)頭文件源文件Acceptor類的實現(xiàn)頭文件源文件SocketIO類的實現(xiàn)頭文件源文件TcpConnection類的實現(xiàn)頭文件源文件回顧先回顧一下我們整個的實現(xiàn)框架所以第一步應(yīng)該是實現(xiàn)的是——封裝系統(tǒng)調(diào)用另外為了方便打印調(diào)試信息可以把日志系統(tǒng)用單例模式先實現(xiàn)出來。關(guān)于用單例模式實現(xiàn)日志系統(tǒng)我們在之前的文章中已經(jīng)實現(xiàn)過了這里就不過多贅述了。詳情可以跳轉(zhuǎn)到該文章中查看Reactor網(wǎng)絡(luò)模型本質(zhì)上就是使用epoll的并發(fā)網(wǎng)絡(luò)模型通過連接器連接到服務(wù)器網(wǎng)絡(luò)然后注冊三個事件(對應(yīng)epoll的三個接口函數(shù))通過事件驅(qū)動(事件循環(huán)和事件分發(fā))來響應(yīng)客戶端的請求。Reactor_v1實現(xiàn)在寫代碼之前要整理一下思緒框架圖中已經(jīng)很明確的寫出了有哪些類我們在第一個版本的內(nèi)容中實際上就是對系統(tǒng)調(diào)用的封裝下面就畫一個類圖整理一下具體實現(xiàn)的內(nèi)容。Acceptor類中可以將地址復(fù)用、端口復(fù)用、bind、listen這些寫法固定的流程封裝在一個ready函數(shù)中。關(guān)于其它類行該可以比較容易的理清思路。需要著重考慮的是IO操作因為使用send/recv或write/read一次處理的字符數(shù)量是由上限的。所以我們應(yīng)該設(shè)置一個合適大小的緩沖用來接收數(shù)據(jù)流。Socket類實現(xiàn)頭文件/** * 封裝文件描述符 */#ifndef_SOCKET_H#define_SOCKET_H#includesys/types.h#includesys/socket.hclassSocket{public:Socket();/** * param fd * 禁止隱式轉(zhuǎn)換防止出現(xiàn)問題 */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類的實現(xiàn)頭文件/** * 地址相關(guān)操作 */#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源文件/** * 地址相關(guān)操作 */#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類的實現(xiàn)頭文件/** * 連接器的實現(xiàn) */#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源文件/** * 連接器的實現(xiàn) */#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類的實現(xiàn)頭文件/** * 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;// 剩余沒讀的字節(jié)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類的實現(xiàn)頭文件/** * 連接管理 */#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的全部函數(shù)完成了對系統(tǒng)調(diào)用的封裝。下面再寫一個測試文件#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;}下面再開一個終端自測一下基本上還是符合我們預(yù)期的也就是說明我們對于底層的封裝沒有了太大問題。我們的Reactor第一代基本完成了當(dāng)然可以把之前封裝的log4cpp的代碼替代perror()。下面的工作就是完成核心層事件驅(qū)動的代碼就是對epoll的封裝另外TcpConnection是用來管理連接的所以當(dāng)我們使用服務(wù)器采用事件驅(qū)動的形式時TcpConnection中的數(shù)據(jù)成員和成員函數(shù)也要隨之變動。 網(wǎng)絡(luò)編程（5）—— Reactor實現(xiàn)(v1)
文章目錄回顧Reactor_v1實現(xiàn)Socket類實現(xiàn)頭文件源文件InetAddr類的實現(xiàn)頭文件源文件Acceptor類的實現(xiàn)頭文件源文件SocketIO類的實現(xiàn)頭文件源文件TcpConnection類的實現(xiàn)頭文件源文件回顧先回顧一下我們整個的實現(xiàn)框架所以第一步應(yīng)該是實現(xiàn)的是——封裝系統(tǒng)調(diào)用另外為了方便打印調(diào)試信息可以把日志系統(tǒng)用單例模式先實現(xiàn)出來。關(guān)于用單例模式實現(xiàn)日志系統(tǒng)我們在之前的文章中已經(jīng)實現(xiàn)過了這里就不過多贅述了。詳情可以跳轉(zhuǎn)到該文章中查看Reactor網(wǎng)絡(luò)模型本質(zhì)上就是使用epoll的并發(fā)網(wǎng)絡(luò)模型通過連接器連接到服務(wù)器網(wǎng)絡(luò)然后注冊三個事件(對應(yīng)epoll的三個接口函數(shù))通過事件驅(qū)動(事件循環(huán)和事件分發(fā))來響應(yīng)客戶端的請求。Reactor_v1實現(xiàn)在寫代碼之前要整理一下思緒框架圖中已經(jīng)很明確的寫出了有哪些類我們在第一個版本的內(nèi)容中實際上就是對系統(tǒng)調(diào)用的封裝下面就畫一個類圖整理一下具體實現(xiàn)的內(nèi)容。Acceptor類中可以將地址復(fù)用、端口復(fù)用、bind、listen這些寫法固定的流程封裝在一個ready函數(shù)中。關(guān)于其它類行該可以比較容易的理清思路。需要著重考慮的是IO操作因為使用send/recv或write/read一次處理的字符數(shù)量是由上限的。所以我們應(yīng)該設(shè)置一個合適大小的緩沖用來接收數(shù)據(jù)流。Socket類實現(xiàn)頭文件/** * 封裝文件描述符 */#ifndef_SOCKET_H#define_SOCKET_H#includesys/types.h#includesys/socket.hclassSocket{public:Socket();/** * param fd * 禁止隱式轉(zhuǎn)換防止出現(xiàn)問題 */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類的實現(xiàn)頭文件/** * 地址相關(guān)操作 */#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源文件/** * 地址相關(guān)操作 */#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類的實現(xiàn)頭文件/** * 連接器的實現(xiàn) */#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源文件/** * 連接器的實現(xiàn) */#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類的實現(xiàn)頭文件/** * 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;// 剩余沒讀的字節(jié)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類的實現(xiàn)頭文件/** * 連接管理 */#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的全部函數(shù)完成了對系統(tǒng)調(diào)用的封裝。下面再寫一個測試文件#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;}下面再開一個終端自測一下基本上還是符合我們預(yù)期的也就是說明我們對于底層的封裝沒有了太大問題。我們的Reactor第一代基本完成了當(dāng)然可以把之前封裝的log4cpp的代碼替代perror()。下面的工作就是完成核心層事件驅(qū)動的代碼就是對epoll的封裝另外TcpConnection是用來管理連接的所以當(dāng)我們使用服務(wù)器采用事件驅(qū)動的形式時TcpConnection中的數(shù)據(jù)成員和成員函數(shù)也要隨之變動。