首页 > 代码库 > 0806------Linux网络编程----------Echo 网络库 学习笔记
0806------Linux网络编程----------Echo 网络库 学习笔记
1.Echo网络库的编写
1.1 Echo网络库1.0
1.1.1 Echo网络库 1.0 框架分析
a）class InetAddress: 主要用来定义一个struct sockaddr_in 结构(用自定义端口号初始化）,并提供获取这个结构体成员如IP、Port等的接口；
b）class Socket : 主要用来把一个普通的 sockfd 变为 listenfd（这里用一个sockfd初始化对象），提供bind 、listen、accept 等接口。
c）class TcpConnection：包含一个TCP连接 和 一个Rio缓冲区，该类对外提供读写的接口并且通过调用 Rio 的函数封装了自己的IO接口；
d）class PollPoller：封装Poll模型的主要成员和接口，这里增加了一个 map<fd, TcpConnectionPtr>类型的成员变量，用来表示 fd 到 TCP 连接的映射。
1.1.2 源码和测试案例
a）Rio 类
#ifndef __RIO_H__#define __RIO_H__#include "NonCopyable.h"#include <stdio.h>#define RIO_BUFFER 8192/* * 将read write 操作封装到一个带有缓冲区的 IO 系统中 * */class Rio : NonCopyable{ public: explicit Rio(int fd); ssize_t readn(char *usrbuf, size_t n); ssize_t writen(const char *usrbuf, size_t n); ssize_t readline(char *usrbuf, size_t maxline); private: ssize_t read(char *usrbuf, size_t n); //这个函数供上述函数调用 int fd_; ssize_t left_; //缓冲区剩余可读取的字节数 char *bufptr_; //指向可读取字节缓冲区 char buffer_[RIO_BUFFER];};#endif /*__RIO_H__*/#include "Rio.h"#include <unistd.h>#include <string.h>#include <stdio.h>#include <errno.h>#define ERR_EXIT(m) do { perror(m); exit(EXIT_FAILURE); }while(0)Rio::Rio(int fd) :fd_(fd), left_(0), bufptr_(buffer_){ memset(buffer_, 0, sizeof buffer_);}ssize_t Rio::readn(char *usrbuf, size_t n){ char *bufptr = usrbuf; ssize_t nleft = n; ssize_t nread; while(nleft > 0){ nread = read(bufptr, nleft); if(nread == -1){ return -1; } else if(nread == 0) break; nleft -= nread; bufptr += nread; } return (n - nleft);}ssize_t Rio::writen(const char *usrbuf, size_t n){ const char *bufptr = usrbuf; size_t nleft = n; int nwrite; while(nleft > 0){ nwrite = write(fd_, bufptr, nleft); if(nwrite <= 0){ if(errno == EINTR) nwrite = 0; else return -1; } nleft -= nwrite; bufptr += nwrite; } return n;}/* * readline 每次都读取一个字符 然后判断是否是\n * 注意读取一行时 如fgets 最后一个字符包含\n */ssize_t Rio::readline(char *usrbuf, size_t maxline){ char *bufptr = usrbuf; ssize_t nleft = maxline - 1; ssize_t nread; char c; while(nleft > 0){ nread = read(&c, 1); if(nread == -1) return -1; else if(nread == 0) break; *bufptr++ = c; nleft --; if(c == ‘\n‘) break; } *bufptr = ‘\0‘; return (maxline - 1 - nleft);}/* * 缓冲区中一直有数据可读 并且每次都是直接从buffer_中读取 * 不再系统调用 只有当缓冲区中没有数据可读时，才调用系统的read */ssize_t Rio::read(char *usrbuf, size_t n){ ssize_t nread; while(left_ <= 0){ //缓冲区中没有有效数据 nread = ::read(fd_, buffer_, sizeof buffer_); if(nread == -1){ if(errno ==EINTR) continue; return -1; } else if(nread == 0) return 0; left_ = nread; bufptr_ = buffer_; } int cnt = n; // cnt 存储返回值，即用户实际获取的字节数 if(left_ < cnt) cnt = left_; ::memcpy(usrbuf, bufptr_, cnt); left_ -= cnt; bufptr_ += cnt; return cnt;}#include "Rio.h"#include <iostream>#include <sys/types.h>#include <sys/stat.h>#include <fcntl.h>#include <stdlib.h>#include <unistd.h>#include <string.h>#define ERR_EXIT(m) do { perror(m); exit(EXIT_FAILURE); }while(0)using namespace std;int main(int argc, const char *argv[]){ int fd = open("test.txt", O_RDONLY); if(fd == -1) ERR_EXIT("open"); Rio rio(fd); char buf[50] = {0}; rio.readline(buf, 50); cout << buf ; memset(buf, 0, sizeof buf); rio.readn(buf, sizeof buf); cout << buf ; close(fd); return 0;}b）InetAddress 类
#ifndef __INET_ADDRESS_H__#define __INET_ADDRESS_H__#include "Copyable.h"#include <netinet/in.h> //strcuct sock_addr_in 结构在这里定义#include <string.h> // memset 函数/* * 对 struct sockaddr_in 的定义 * */typedef struct sockaddr SA;class InetAddress : private Copyable{ public: explicit InetAddress(uint16_t port); //explicit 禁止参数隐式转换 InetAddress(const struct sockaddr_in &addr); const struct sockaddr_in *getSockAddrInet() const{ return &addr_; } void setSockAddrInet(const struct sockaddr_in &addr){ addr_= addr; } uint16_t portNetEndian() const{ //返回sockaddr_in的端口 return addr_.sin_port; } uint32_t ipNetEndian() const{ //返回IP return addr_.sin_addr.s_addr; } private: struct sockaddr_in addr_;};inline InetAddress::InetAddress(uint16_t port){ ::memset(&addr_, 0, sizeof addr_); // ::表示全局作用域 避免重名 addr_.sin_family = AF_INET; addr_.sin_port = ::htons(port); addr_.sin_addr.s_addr = ::htonl(INADDR_ANY);}inline InetAddress::InetAddress(const struct sockaddr_in &addr) :addr_(addr){}#endif /*__INET_ADDRESS_H__*/c）Socket类
#ifndef __SOCKET_H__#define __SOCKET_H__#include "NonCopyable.h"/* * 将一个fd 变成listenfd， 并开始处理用户的请求 * */class InetAddress;class Socket : NonCopyable{ public: explicit Socket(int sockfd); ~Socket(); //关闭sockfd int fd() const{ return sockfd_; } void bindAddress(const InetAddress &addr); void listen(); int accept(); void shutdownWrite(); void setReusePort(); private: const int sockfd_;};#endif /*__SOCKET_H__*/#include "Socket.h"#include "InetAddress.h"#include <sys/types.h>#include <sys/socket.h>#include <netinet/in.h>#include <stdlib.h>#include <unistd.h>#include <stdio.h>#define ERR_EXIT(m) do { perror(m); exit(EXIT_FAILURE); }while(0)Socket::Socket(int sockfd) :sockfd_(sockfd){ }Socket::~Socket(){ ::close(sockfd_);}void Socket::bindAddress(const InetAddress &addr){ if(::bind(sockfd_, (struct sockaddr *)&addr, sizeof (struct sockaddr_in)) == -1) ERR_EXIT("bind");}void Socket::listen(){ if(::listen(sockfd_, SOMAXCONN) == -1) ERR_EXIT("listen");}int Socket::accept(){ int peerfd = ::accept(sockfd_, NULL, NULL); if(peerfd == -1) ERR_EXIT("accept"); return peerfd;}void Socket::shutdownWrite(){ ::shutdown(sockfd_, SHUT_WR);}void Socket::setReusePort(){ int on = 1; if(setsockopt(sockfd_, SOL_SOCKET, SO_REUSEADDR, &on, static_cast<socklen_t>(sizeof on)) == -1) ERR_EXIT("setsockopt");}d）TcpConnection 类
#ifndef __TCP_CONNECTION_H__#define __TCP_CONNECTION_H__#include "NonCopyable.h"#include "InetAddress.h"#include "Socket.h"#include "Rio.h"#include <string>#include <functional>#include <memory>/* * 表示一个 TCP 连接，提供收发信息，关闭连接等操作 */class TcpConnection;typedef std::shared_ptr<TcpConnection> TcpConnectionPtr; //智能指针类型class TcpConnection{ public: typedef std::function<void (const TcpConnectionPtr &)> TcpConnectionCallback; //回调函数类型,注意这里含有一个智能指针参数，因此回调的时候注意传参数 explicit TcpConnection(int sockfd); ~TcpConnection(); int fd() const{ return socket_.fd(); } // 通过调用 Rio 中的函数，封装一套自己的 IO 接口 ssize_t readn(char *usrbuf, size_t n); ssize_t readline(char *usrbuf, size_t maxline); ssize_t writen(const char *usrbuf, size_t n); void send(const std::string &s); // 向TCP连接中的对端发送数据 std::string receive();//接收数据 void shutdown(); //关闭本连接的写端 private: Socket socket_; //一个Socket对象就表示一条已经建立的TCP连接 Rio buffer_; //该TCP连接的读写缓冲区 即用本fd初始化};#endif /*__TCP_CONNECTION_H__*/#include "TcpConnection.h"#include <stdlib.h>using namespace std;#define ERR_EXIT(m) do { perror(m); exit(EXIT_FAILURE); }while(0)TcpConnection::TcpConnection(int sockfd) :socket_(sockfd), buffer_(sockfd){}TcpConnection::~TcpConnection(){ shutdown();}ssize_t TcpConnection::readn(char *usrbuf, size_t n){ ssize_t nread = buffer_.readn(usrbuf, n); if(nread == -1) ERR_EXIT("read"); return nread;}ssize_t TcpConnection::readline(char *usrbuf, size_t maxline){ ssize_t nread = buffer_.readline(usrbuf, maxline); if(nread == -1) ERR_EXIT("readline"); return nread;}ssize_t TcpConnection::writen(const char *usrbuf, size_t n){ ssize_t nwrite = buffer_.writen(usrbuf, n); if(nwrite <= 0) ERR_EXIT("writen"); return nwrite;}void TcpConnection::send(const string &s){ writen(s.c_str(), s.size()); //调用自己的IO 函数}string TcpConnection::receive(){ char buf[1024]; readline(buf, sizeof buf); return string(buf);}void TcpConnection::shutdown(){ socket_.shutdownWrite();}#include "TcpConnection.h"#include "InetAddress.h"#include "Socket.h"#include <iostream>#include <stdlib.h>#define ERR_EXIT(m) do { perror(m); exit(EXIT_FAILURE); }while(0)/* * 测试 TcpConnection 类 * */using namespace std;int main(int argc, const char *argv[]){ int fd = socket(AF_INET, SOCK_STREAM, 0); if(fd == -1) ERR_EXIT("socket"); InetAddress inet(9999); Socket sock(fd); sock.setReusePort(); sock.bindAddress(inet); sock.listen(); int peerfd; peerfd = sock.accept(); TcpConnection con(peerfd); char buf[1024]; while(1){ con.readline(buf, 1024); cout << "recv data: " << buf << endl; con.writen(buf, strlen(buf)); } return 0;}e）PollPoller类
#ifndef __POLL_POLLELR_H__#define __POLL_POLLELR_H__#include "NonCopyable.h"#include "TcpConnection.h"#include <poll.h>#include <map>#include <functional>class PollPoller : private NonCopyable{ public: typedef TcpConnection::TcpConnectionCallback PollerCallback; //为什么要用TcpConnection的回调函数类型 explicit PollPoller(int listenfd); void poll(); void handleAccept(); void handleData(); void handleConnectionEvent(int peerfd); void handleMessageEvent(int peerfd); void handleCloseEvent(int i); void setConnectionCallback(const PollerCallback &cb){ onConnectionCallback_ = cb; } void setMessageCallback(const PollerCallback &cb){ onMessageCallback_ = cb; } void setCloseCallback(const PollerCallback &cb){ onCloseCallback_ = cb; } private: struct pollfd event_[2048]; int listenfd_; int maxi_; int nready_; std::map<int, TcpConnectionPtr> lists_; // 一个fd 对应 一个指向Tcp连接的智能指针(采用引用计数) PollerCallback onConnectionCallback_; PollerCallback onMessageCallback_; PollerCallback onCloseCallback_; typedef std::map<int, TcpConnectionPtr>::iterator TcpIterator;};#endif /*__POLL_POLLELR_H__*/#include "PollPoller.h"#include <utility>#include <assert.h>#include <iostream>#define ERR_EXIT(m) do { perror(m); exit(EXIT_FAILURE); }while(0)PollPoller::PollPoller(int listenfd) :listenfd_(listenfd){ int i; for(i = 0; i < 2048; i++) event_[i].fd = -1; event_[0].fd = listenfd_; event_[0].events = POLLIN; maxi_ = 0; nready_ = 0;}void PollPoller::poll(){ int ret; do{ ret = ::poll(event_, maxi_ + 1, 10000); }while(ret == -1 && errno == EINTR); if(ret == -1) ERR_EXIT("poll"); nready_ = ret;}void PollPoller::handleAccept(){ if(event_[0].revents & POLLIN){ int peerfd; if((peerfd = ::accept(listenfd_, NULL, NULL)) == -1) ERR_EXIT("accpet"); handleConnectionEvent(peerfd); }}void PollPoller::handleData(){ // 依次处理客户 int i; for(i = 1; i <= maxi_; i++){ //注意 这里从1开始 int peerfd = event_[i].fd; if(peerfd == -1) continue; if(event_[i].revents & POLLIN){ char buf[1024]; int nread = ::recv(peerfd, buf, sizeof buf, MSG_PEEK); //预读取 if(nread == -1) ERR_EXIT("revc"); else if(nread == 0) handleCloseEvent(i); else handleMessageEvent(peerfd); } }}void PollPoller::handleConnectionEvent(int peerfd){ // 1.将新连接的fd 添加到 event 数组中 int i; for(i = 1; i < 2048; i++){ if(event_[i].fd == -1){ event_[i].fd = peerfd; event_[i].events = POLLIN; if(i > maxi_) maxi_ = i; break; } } if(i == 2048){ std::cout << " too many clients" << std::endl; exit(EXIT_FAILURE); } // 2.将该新的TCP连接添加到 map 中 TcpConnectionPtr conn(new TcpConnection(peerfd)); std::pair<TcpIterator, bool> ret = lists_.insert(make_pair(peerfd, conn)); assert(ret.second == true); //这里确认是否成功插入 // 3.调用处理连接时要用的回调函数 onConnectionCallback_(conn); // 注意回调函数要传入一个TcpConnectionPtr 对象表示当前 TCP 连接}void PollPoller::handleMessageEvent(int peerfd){ TcpIterator it = lists_.find(peerfd); assert(it != lists_.end()); onMessageCallback_(it->second);}void PollPoller::handleCloseEvent(int i){ // 1.从event_中清除 assert(i >= 0 && i < 2048); int peerfd = event_[i].fd; assert(peerfd != -1); event_[i].fd = -1; //2.调用Close时的回调函数 并从map 中清除 TcpIterator it = lists_.find(peerfd); assert(it != lists_.end()); onCloseCallback_(it->second); lists_.erase(it);}#include "PollPoller.h"#include "Socket.h"#include "InetAddress.h"#include <iostream>using namespace std;#define ERR_EXIT(m) do { perror(m); exit(EXIT_FAILURE); }while(0)// 自定义的回调函数void onConnect(const TcpConnectionPtr &conn){ conn->send("Hello!\r\n");}void onMessage(const TcpConnectionPtr &conn){ string recvData(conn->receive()); cout << "recv data:" << recvData; conn->send(recvData);}void onClose(const TcpConnectionPtr &conn){ cout << "Close conn" << endl; conn->shutdown();}int main(int argc, const char *argv[]){ int fd = socket(AF_INET, SOCK_STREAM, 0); if(fd == -1) ERR_EXIT("socket"); Socket sock(fd); InetAddress inet(9999); sock.bindAddress(inet); sock.listen(); PollPoller poller(sock.fd()); poller.setConnectionCallback(onConnect); poller.setMessageCallback(onMessage); poller.setCloseCallback(onClose); while(1){ poller.poll(); poller.handleAccept(); poller.handleData(); } return 0;}1.2 Echo网络库 2.0
1.2.1 Echo 网络库2.0改进：
a）InetAddress 结构中新增获取本地sockaddr_in 结构和对端sockaddr_in 结构的成员函数 以及 获取点分十进制的IP和本地字节序端口号的函数；
b）TcpConnection 新增回调函数的设置和执行接口，其中在 shared_ptr 的 enable_shared_from_this 类中定义了成员函数 shared_from_this() ，返回 shared_ptr<TcpConnection> 类型的变量；此外还新增了两个InetAddress类的成员变量即一个TCP连接两端的sockaddr_in 结构体，并加入了构造函数的参数中。
c）PollPoller 中函数执行回调函数的方式发生改变，这里需要向TcpConnection对象中注册回调函数，然后由TcpConnectionPtr 智能指针对象调用TcpConnection中注册的回调函数。
1.2.2 改进的程序
a）InetAddress类
#ifndef __INET_ADDRESS_H__#define __INET_ADDRESS_H__#include "Copyable.h"#include <netinet/in.h> //strcuct sock_addr_in 结构在这里定义#include <string.h> // memset 函数#include <string>/* * 对 struct sockaddr_in 的定义 * */typedef struct sockaddr SA;class InetAddress : private Copyable{ public: explicit InetAddress(uint16_t port); //explicit 禁止参数隐式转换 InetAddress(const struct sockaddr_in &addr); const struct sockaddr_in *getSockAddrInet() const{ return &addr_; } void setSockAddrInet(const struct sockaddr_in &addr){ addr_= addr; } //返回网络字节序的 IP 和 port uint16_t portNetEndian() const{ return addr_.sin_port; } uint32_t ipNetEndian() const{ return addr_.sin_addr.s_addr; } // 2.0 add // 返回主机字节序的IP 和 Port std::string toIP() const; uint16_t toPort() const; static InetAddress getLocalAddress(int sockfd); static InetAddress getPeerAddress(int sockfd); private: struct sockaddr_in addr_;};#endif /*__INET_ADDRESS_H__*/#include "InetAddress.h"#include <sys/socket.h>#include <arpa/inet.h>#include <stdlib.h>#include <stdio.h>#define ERR_EXIT(m) do { perror(m); exit(EXIT_FAILURE); }while(0)InetAddress::InetAddress(uint16_t port){ ::memset(&addr_, 0, sizeof addr_); // ::表示全局作用域 避免重名 addr_.sin_family = AF_INET; addr_.sin_port = ::htons(port); addr_.sin_addr.s_addr = ::htonl(INADDR_ANY);}InetAddress::InetAddress(const struct sockaddr_in &addr) :addr_(addr){}std::string InetAddress::toIP() const{ return std::string(inet_ntoa(addr_.sin_addr));}uint16_t InetAddress::toPort() const{ return ntohs(addr_.sin_port);}InetAddress InetAddress::getLocalAddress(int sockfd){ struct sockaddr_in addr; socklen_t len = sizeof addr; if(::getsockname(sockfd, (SA *)&addr, &len) == -1) ERR_EXIT("getsockname"); return InetAddress(addr);}InetAddress InetAddress::getPeerAddress(int sockfd){ struct sockaddr_in addr; socklen_t len = sizeof addr; if(::getpeername(sockfd, (SA *)&addr, &len) == -1) ERR_EXIT("getpeername"); return InetAddress(addr);}b）TcpConnection 类
#ifndef __TCP_CONNECTION_H__#define __TCP_CONNECTION_H__#include "NonCopyable.h"#include "InetAddress.h"#include "Socket.h"#include "Rio.h"#include <string>#include <functional>#include <memory>/* * 表示一个 TCP 连接，提供收发信息，关闭连接等操作 */class TcpConnection;typedef std::shared_ptr<TcpConnection> TcpConnectionPtr; //智能指针类型class TcpConnection : private NonCopyable, public std::enable_shared_from_this<TcpConnection>{ public: typedef std::function<void (const TcpConnectionPtr &)> TcpConnectionCallback; //回调函数类型,注意这里含有一个智能指针参数，因此回调的时候注意传参数 TcpConnection(int sockfd, const InetAddress localAddr, const InetAddress peerAddr); ~TcpConnection(); int fd() const{ return socket_.fd(); } // 通过调用 Rio 中的函数，封装一套自己的 IO 接口 ssize_t readn(char *usrbuf, size_t n); ssize_t readline(char *usrbuf, size_t maxline); ssize_t writen(const char *usrbuf, size_t n); void send(const std::string &s); // 向TCP连接中的对端发送数据 std::string receive();//接收数据 void shutdown(); //关闭本连接的写端 //2.0 add //获取本TCP 连接两端的sockaddr 结构 const InetAddress &getLocalAddr() const{ return localAddr_; } const InetAddress &getPeerAddress() const{ return peerAddr_; } //注册回调函数 void setConnectionCallback(const TcpConnectionCallback &cb){ onConnectionCallback_= cb; } void setMessageCallback(const TcpConnectionCallback &cb){ onMessageCallback_ = cb; } void setCloseCallback(const TcpConnectionCallback &cb){ onCloseCallback_ = cb; } //执行回调函数 void handleConnection(){ if(onConnectionCallback_) onConnectionCallback_(shared_from_this()); //shared_from_this 返回TcpConnectionPtr } void handleMessage(){ if(onMessageCallback_) onMessageCallback_(shared_from_this()); } void handleClose(){ if(onCloseCallback_) onCloseCallback_(shared_from_this()); else shutdown(); } private: Socket socket_; //一个Socket对象就表示一条已经建立的TCP连接 Rio buffer_; //该TCP连接的读写缓冲区 即用本fd初始化 //2.0 add const InetAddress localAddr_; const InetAddress peerAddr_; TcpConnectionCallback onConnectionCallback_; TcpConnectionCallback onMessageCallback_; TcpConnectionCallback onCloseCallback_;};#endif /*__TCP_CONNECTION_H__*/#include "TcpConnection.h"#include <stdlib.h>using namespace std;#define ERR_EXIT(m) do { perror(m); exit(EXIT_FAILURE); }while(0)TcpConnection::TcpConnection(int sockfd, const InetAddress localAddr, const InetAddress peerAddr) :socket_(sockfd), buffer_(sockfd), localAddr_(localAddr), peerAddr_(peerAddr){}TcpConnection::~TcpConnection(){ shutdown();}ssize_t TcpConnection::readn(char *usrbuf, size_t n){ ssize_t nread = buffer_.readn(usrbuf, n); if(nread == -1) ERR_EXIT("read"); return nread;}ssize_t TcpConnection::readline(char *usrbuf, size_t maxline){ ssize_t nread = buffer_.readline(usrbuf, maxline); if(nread == -1) ERR_EXIT("readline"); return nread;}ssize_t TcpConnection::writen(const char *usrbuf, size_t n){ ssize_t nwrite = buffer_.writen(usrbuf, n); if(nwrite <= 0) ERR_EXIT("writen"); return nwrite;}void TcpConnection::send(const string &s){ writen(s.c_str(), s.size()); //调用自己的IO 函数}string TcpConnection::receive(){ char buf[1024]; readline(buf, sizeof buf); return string(buf);}void TcpConnection::shutdown(){ socket_.shutdownWrite();}c）PollPoller类
#ifndef __POLL_POLLELR_H__#define __POLL_POLLELR_H__#include "NonCopyable.h"#include "TcpConnection.h"#include <poll.h>#include <map>#include <functional>class PollPoller : private NonCopyable{ public: typedef TcpConnection::TcpConnectionCallback PollerCallback; explicit PollPoller(int listenfd); void poll(); void handleAccept(); void handleData(); void handleConnectionEvent(int peerfd); void handleMessageEvent(int peerfd); void handleCloseEvent(int i); void setConnectionCallback(const PollerCallback &cb){ onConnectionCallback_ = cb; } void setMessageCallback(const PollerCallback &cb){ onMessageCallback_ = cb; } void setCloseCallback(const PollerCallback &cb){ onCloseCallback_ = cb; } private: struct pollfd event_[2048]; int listenfd_; int maxi_; int nready_; std::map<int, TcpConnectionPtr> lists_; // 一个fd 对应 一个指向Tcp连接的智能指针(采用引用计数) PollerCallback onConnectionCallback_; PollerCallback onMessageCallback_; PollerCallback onCloseCallback_; typedef std::map<int, TcpConnectionPtr>::iterator TcpIterator;};#endif /*__POLL_POLLELR_H__*/#include "PollPoller.h"#include <utility>#include <assert.h>#include <iostream>#define ERR_EXIT(m) do { perror(m); exit(EXIT_FAILURE); }while(0)PollPoller::PollPoller(int listenfd) :listenfd_(listenfd){ int i; for(i = 0; i < 2048; i++) event_[i].fd = -1; event_[0].fd = listenfd_; event_[0].events = POLLIN; maxi_ = 0; nready_ = 0;}void PollPoller::poll(){ int ret; do{ ret = ::poll(event_, maxi_ + 1, 10000); }while(ret == -1 && errno == EINTR); if(ret == -1) ERR_EXIT("poll"); nready_ = ret;}void PollPoller::handleAccept(){ if(event_[0].revents & POLLIN){ int peerfd; if((peerfd = ::accept(listenfd_, NULL, NULL)) == -1) ERR_EXIT("accpet"); handleConnectionEvent(peerfd); }}void PollPoller::handleData(){ // 依次处理客户 int i; for(i = 1; i <= maxi_; i++){ //注意 这里从1开始 int peerfd = event_[i].fd; if(peerfd == -1) continue; if(event_[i].revents & POLLIN){ char buf[1024]; int nread = ::recv(peerfd, buf, sizeof buf, MSG_PEEK); //预读取 if(nread == -1) ERR_EXIT("revc"); else if(nread == 0) handleCloseEvent(i); else handleMessageEvent(peerfd); } }}void PollPoller::handleConnectionEvent(int peerfd){ // 1.将新连接的fd 添加到 event 数组中 int i; for(i = 1; i < 2048; i++){ if(event_[i].fd == -1){ event_[i].fd = peerfd; event_[i].events = POLLIN; if(i > maxi_) maxi_ = i; break; } } if(i == 2048){ std::cout << " too many clients" << std::endl; exit(EXIT_FAILURE); } // 2.将该新的TCP连接添加到 map 中 TcpConnectionPtr conn(new TcpConnection(peerfd, InetAddress::getLocalAddress(peerfd), InetAddress::getPeerAddress(peerfd)) ); std::pair<TcpIterator, bool> ret = lists_.insert(make_pair(peerfd, conn)); assert(ret.second == true); //这里确认是否成功插入 //2.0 add 注册回调函数到Tcpconnection类中 conn->setConnectionCallback(onConnectionCallback_); conn->setMessageCallback(onMessageCallback_); conn->setCloseCallback(onCloseCallback_); // 3.调用处理连接时要用的回调函数 //onConnectionCallback_(conn); // 注意回调函数要传入一个TcpConnectionPtr 对象表示当前 TCP 连接 //2.0 回调函数的方式发生改变 conn->handleConnection();}void PollPoller::handleMessageEvent(int peerfd){ TcpIterator it = lists_.find(peerfd); assert(it != lists_.end()); //onMessageCallback_(it->second); //2.0 add it->second->handleMessage();}void PollPoller::handleCloseEvent(int i){ // 1.从event_中清除 assert(i >= 0 && i < 2048); int peerfd = event_[i].fd; assert(peerfd != -1); event_[i].fd = -1; //2.调用Close时的回调函数 并从map 中清除 TcpIterator it = lists_.find(peerfd); assert(it != lists_.end()); // onCloseCallback_(it->second); //2.0 add it->second->handleClose(); lists_.erase(it);}1.3 Echo服务器模型 3.0
1.3.1 Echo 服务器模型3.0改进
a）根据现有的类封装一个TcpServer类，包含设置回调函数，开启服务器等接口，到这里一个服务器模型已经相对完整。
1.3.2 新增 TcpServer 类
#ifndef __TCP_SERVER_H#define __TCP_SERVER_H#include "InetAddress.h"#include "Socket.h"#include "TcpConnection.h"#include "PollPoller.h"#include <memory>class TcpServer : private NonCopyable{ public: typedef TcpConnection::TcpConnectionCallback TcpServerCallback; explicit TcpServer(const InetAddress &inet); // 设置回调函数 void setConnection(const TcpServerCallback &cb){ onConnect_ = cb; } void setMessage(const TcpServerCallback &cb){ onMessage_ = cb; } void setClose(const TcpServerCallback &cb){ onClose_ =cb; } // 开启服务器 void start(); private: std::unique_ptr<Socket> socket_; std::unique_ptr<PollPoller> poller_; TcpServerCallback onConnect_; TcpServerCallback onMessage_; TcpServerCallback onClose_;};#endif /*__TCP_SERVER_H*/#include "TcpServer.h"#include <signal.h>#define ERR_EXIT(m) do { perror(m); exit(EXIT_FAILURE); }while(0)//4 addclass IgnoreSigpipe{ public: IgnoreSigpipe(){ if(::signal(SIGPIPE, SIG_IGN) == SIG_ERR) ERR_EXIT("signal"); }};IgnoreSigpipe initObj; //全局对象，系统初始化时必然处理SIGPIPETcpServer::TcpServer(const InetAddress &addr){ int sockfd = ::socket(AF_INET, SOCK_STREAM, 0); if(sockfd == -1) ERR_EXIT("socket"); socket_.reset(new Socket(sockfd)); //reset是unique_ptr的一个成员函数 socket_->setReusePort(); socket_->bindAddress(addr); socket_->listen();}void TcpServer::start(){ poller_.reset(new PollPoller(socket_->fd())); poller_->setConnectionCallback(onConnect_); poller_->setMessageCallback(onMessage_); poller_->setCloseCallback(onClose_); while(1){ poller_->poll(); poller_->handleAccept(); poller_->handleData(); }}1.4 Echo 网络库 4.0
1.4.1 Echo 网络库 4.0 改进
a）增加了线程池机制 和 异常的处理；
b）线程池中将 PtrVector<Thread> threads_;改为std::vector<std::unique_ptr<Thread> > threads_;
c）在TcpServer类中增加了sigpipe信号的处理；
1.4.2 新增类
a）MutexLock Condition Thread ThreadPool
class Thread : NonCopyable{ public: typedef std::function<void ()> Callback_Func; Thread(Callback_Func func); ~Thread(); void start(); void join(); private: pthread_t tid_; bool is_started_; Callback_Func callback_;};#endif#include "Thread.h"#include <iostream>class RunFunction{ public: typedef std::function<void ()> Callback_Func; RunFunction(Callback_Func func) :func_(func) { } void run(){ func_(); } private: std::function<void ()> func_;};Thread::Thread(Callback_Func func) :tid_(0), is_started_(false), callback_(func){}Thread::~Thread(){// if(is_started_){ pthread_detach(tid_); }}void *thread_func(void *arg){ RunFunction *pr = static_cast<RunFunction *>(arg); pr->run(); delete pr; return NULL;}void Thread::start(){ is_started_ = true; RunFunction *pr = new RunFunction(callback_); if(pthread_create(&tid_, NULL, thread_func, pr)){ delete pr; }}void Thread::join(){ is_started_ = false; pthread_join(tid_, NULL);}#ifndef __THREADPOOL_H__#define __THREADPOOL_H__#include "Condition.h"#include "MutexLock.h"#include "Thread.h"#include <functional>#include <queue>#include <memory>class Thread;class ThreadPool{ public: typedef std::function<void ()> Task; //任务类型 ThreadPool(size_t poolSize, size_t queueSize); ~ThreadPool(); void start(); void stop(); void add_task(const Task &tsk); Task get_task(); private: void runInThread(); //线程池内的回调函数 size_t poolSize_; size_t queueSize_; //PtrVector<Thread> threads_; //线程数组 // 4 add std::vector<std::unique_ptr<Thread> > threads_; std::queue<Task> que_; //任务队列 MutexLock mutex_; Condition empty_; Condition full_; bool isStarted_; //线程池是否开启};#endif /*__THREADPOOL_H__*/#include "ThreadPool.h"#include "Thread.h"ThreadPool::ThreadPool(size_t poolSize, size_t queueSize) :poolSize_(poolSize), queueSize_(queueSize), empty_(mutex_), full_(mutex_), isStarted_(false){}ThreadPool::~ThreadPool(){ if(isStarted_){ stop(); }}void ThreadPool::start(){ if(isStarted_) return; isStarted_ = true; size_t i; for(i = 0; i < poolSize_; i++){ threads_.push_back(std::unique_ptr<Thread>(new Thread(std::bind(&ThreadPool::runInThread, this)))); } for(i = 0; i < poolSize_; i++){ threads_[i]->start(); //数组中存放的是指针 Thread * }}void ThreadPool::stop(){ if(! isStarted_) return; { MutexLockGuard lock(mutex_); isStarted_ = false; full_.notifyAll(); } size_t i; for(i = 0; i < poolSize_; i++){ threads_[i]->join(); } while(!que_.empty()){ que_.pop(); } threads_.clear();}void ThreadPool::runInThread(){ while(isStarted_){ Task tsk(get_task()); if(tsk) tsk(); //执行真正的任务 }}void ThreadPool::add_task(const Task &tsk){ MutexLockGuard lock(mutex_); while(que_.size() == queueSize_){ // 任务队列已满 empty_.wait(); } que_.push(tsk); full_.notify();}ThreadPool::Task ThreadPool::get_task(){ MutexLockGuard lock(mutex_); while(que_.empty() && isStarted_){ full_.wait(); } //若是被stop函数唤醒 此时队列已被清空 Task tsk; if(!que_.empty()){ tsk = que_.front(); que_.pop(); empty_.notify(); } return tsk;}b）Exception 类
#ifndef __EXCEPTION_H__#define __EXCEPTION_H__#include <exception>#include <string>class Exception : public std::exception{ public: Exception(const char *); Exception(const std::string &); virtual ~Exception() throw();//表示这个函数不抛出异常 virtual const char * what() const throw(); const char* stackTrace()throw(); private: void fillStackTrace();// std::string message_; //异常的名字 std::string stack_; //栈痕迹};#endif#include "Exception.h"#include <execinfo.h>#include <stdlib.h>Exception::Exception(const char *s) :message_(s){ fillStackTrace();}Exception::Exception(const std::string &s) :message_(s){ fillStackTrace();}Exception::~Exception() throw(){}const char * Exception::what() const throw(){ return message_.c_str();}void Exception::fillStackTrace(){ const int len = 200; void * buffer[len]; // 获取栈痕迹 存储在buffer数组中，这里len是数组的最大长度，而返回值nptrs是数组的实际长度 int nptrs = ::backtrace(buffer, len); //把buffer中的地址转化成字符串存储在strings中 //这里在生成strings的时候调用了malloc函数 动态分配了内存 因此后面需要free char** strings = ::backtrace_symbols(buffer, nptrs); if(strings){ int i; for(i = 0; i < nptrs; i++){ stack_.append(strings[i]); stack_.push_back(‘\n‘); } } free(strings);}const char *Exception::stackTrace() throw(){ return stack_.c_str();}1.5 打包成Echo网络库放进系统库文件中
1.5.1 步骤
a）执行make，生成echo头文件夹和静态库libecho.a；
将echo安装到/usr/include/下，将libecho.a放置/usr/lib/下。
编译的时候需要加上-std=c++0x -lecho -lpthread
b）Makefile
.PHONY:cleanCC=g++CFLAGS=-Wall -gBIN=libecho.aOBJS=InetAddress.o Socket.o Rio.o TcpConnection.o PollPoller.o TcpServer.o Thread.o Condition.o ThreadPool.o Exception.oSTARD=-std=c++0x -rdynamicLINKS=-lpthread$(BIN):$(OBJS) ar -crv $@ -o $^ chmod +x $@ rm -f *.o mkdir echo cp *.h echo/%.o:%.cpp $(CC) $(CFLAGS) -c $< -o $@ $(STARD)clean: rm -rf *.o $(BIN) echo
1.6 小结（重点）
a）NonCopyable、Copyable表示对象是否具有value语义（复制和赋值），Echo中除了InetAddress之外，其余均禁用掉了value语义，这是为了避免潜在的BUG；
b）Exception相比标准库的exception，增加了打印栈痕迹的功能；
c）ThreadPool系列，主要包含MutexLock、Condition、Thread、ThreadPool。其中大量采用了RAII技术，避免资源的泄露，对于Thread和ThreadPool，我们采用了function作为泛型技术，用户只需注册回调函数。
d）Timer，内部采用timerfd系列的定时器，不使用信号，而是使用fd可读作为定时器的触发事件，这使得Timer可以加入到IO复用模型，我们采用的是Poll模型。也可以单独把Timer放到一个线程，这就是TimerThread的产生。
e）用户注册事件与回调流程：先注册给TcpServer，然后是PollPoller，之后是TcpConnection，这样完成了事件的注册。回调函数由PollPoller触发，通过map寻找到Tcp连接，然后调用里面的回调函数。
f）TcpServer实质是一个IO复用模型，ThreadPool则是代笔多线程。用户在使用时，可以只选择其一。如果计算任务负担较重，可以将计算任务与Tcp回发封装成函数，交给线程池去计算。
g)此时，运行TcpServer的线程是一个IO线程，ThreadPool 里面的线程专注于CPU密集型计算。
2.Echo 网络库的使用
2.1 转换大小写回显服务器
2.1.1 转换大小写回显服务器分析：
a）根据现有的网络库采用类的组合可以编写出新的服务器类，这个服务器类要做的事情有，设置回调函数，开启服务器两件事。
b）第一件事在Server类的构造函数中设置，因此本类中需要有要绑定的回调函数，这些回调函数主要是用来处理连接发生和关闭以及传送数据时的状况。此外，对于任务型服务器，这里可以和线程池组合，将新任务放到任务队列中，供线程池调度，对于任务较轻的服务器可以不使用线程池。这里将转换大小写和回显操作封装成 compute 函数，每次来一个任务就加入到任务队列中。
2.1.2 程序示例
#ifndef __ECHO_SERVER_H#define __ECHO_SERVER_H#include <echo/TcpServer.h>#include <echo/ThreadPool.h>#include <echo/NonCopyable.h>class EchoServer : NonCopyable{ public: EchoServer(const InetAddress &addr); void start(); private: void onConnection(const TcpConnectionPtr &conn); void onMessage(const TcpConnectionPtr &conn); void onClose(const TcpConnectionPtr &conn); void compute(std::string s, const TcpConnectionPtr &conn); //转换大小写函数 TcpServer server_; ThreadPool pool_;};#endif /*__ECHO_SERVER_H*/#include "EchoServer.h"#include <functional>#include <iostream>using namespace std;using namespace placeholders;EchoServer::EchoServer(const InetAddress &addr) :server_(addr), pool_(1000, 4){// 任务数组的大小 和 线程池的大小 server_.setConnection(bind(&EchoServer::onConnection, this, _1)); server_.setMessage(bind(&EchoServer::onMessage, this, _1)); server_.setClose(bind(&EchoServer::onClose, this, _1));}void EchoServer::onConnection(const TcpConnectionPtr &conn){ conn->send("Hello! Welcome to Echo Server!\r\n");}void EchoServer::onMessage(const TcpConnectionPtr &conn){ string s = conn->receive(); cout << "recv : " << s; pool_.add_task(bind(&EchoServer::compute, this, s, conn));}void EchoServer::onClose(const TcpConnectionPtr &conn){ cout << "Client closed\r\n" << endl; conn->shutdown();}void EchoServer::compute(string s, const TcpConnectionPtr &conn){ int i; for(i = 0; i < s.size(); i++){ if(isupper(s[i])) s[i] = tolower(s[i]); else if(islower(s[i])) s[i] = toupper(s[i]); } conn->send(s);}void EchoServer::start(){ pool_.start(); server_.start();}#include "EchoServer.h"using namespace std;int main(int argc, const char *argv[]){ EchoServer server(InetAddress(9999)); server.start(); return 0;} 2.2 文本查询服务器
2.2.1 文本查询程序分析：
a）程序的要求是，用户输入一个单词，输出该单词的词频和每次出现该单词的一行文本；
b）数据结构，用一个vector数组存储要录入文本的每行内容，用map<int, set>组合来实现单词和行号集合的映射关系，这样每次查询时，会返回一个行号的集合set，根据set中的元素依次输出vector中的元素内容即可；
c）本程序在构造函数中录入文本，填充上述两个数据结构，给出查询接口，最后需要对查询结果拼接成字符串，这在之后的网络传输中比较方便。
d）最后使用Echo网络库将文本查询类和TcpServer以及线程池类组合，封装出本查询服务器类，这里相对上例只需要修改任务函数，此外还要注意，在用户输入的字符串在传输中加了\r\n，因此服务器需要对接收到的字符处理一下，否则将不匹配。
2.2.2 程序示例
#ifndef __TEXTQUERY_H__#define __TEXTQUERY_H__#include <string>#include <vector>#include <map>#include <set>class TextQuery{ public: typedef std::vector<std::string>::size_type LineNo; //行号的类型 TextQuery(const std::string &filename); std::string run_query(const std::string &word) const; std::string print_result(const std::string &word, const std::set<LineNo> &nums) const; private: void read_file(const std::string &filename); std::vector<std::string> lines_; //存储文档中每行的内容 std::map<std::string, std::set<LineNo> > word_map_; //单词和行号之间的映射};#endif /*__TEXTQUERY_H__*/#include "TextQuery.h"#include <sstream>#include <fstream>#include <echo/Exception.h>#include <stdio.h>using namespace std;TextQuery::TextQuery(const string &filename){ read_file(filename);}string TextQuery::run_query(const string &word) const{ map<string, set<LineNo> >::const_iterator loc = word_map_.find(word); if(loc == word_map_.end()) return print_result(word, set<LineNo>()); else return print_result(word, loc->second);}string TextQuery::print_result(const string &word, const set<LineNo> &nums) const{ string ret; char buf[16]; snprintf(buf, sizeof buf, "%u", nums.size()); ret += word + " occurs " + buf + " times\n"; set<LineNo>::const_iterator it = nums.begin(); for(; it != nums.end(); it++){ snprintf(buf, sizeof buf, "%u", *it + 1); ret += string("(line ") + buf + " ) "; ret += lines_[*it] + "\n"; } return ret;}void TextQuery::read_file(const string &filename){ ifstream in; in.open(filename.c_str()); if(!in) throw Exception("open file fail"); //读入每一行的内容到数组vector中 string line; while(getline(in, line)) lines_.push_back(line); in.close(); //将每一行的内容抽取出单词和行号 存入map中 for(LineNo num = 0; num != lines_.size(); num++){ istringstream line(lines_[num]); string word; while(line >> word){ word_map_[word].insert(num);//这里word_map_[word]是一个set类型 } }}#ifndef __QUERY_SERVER_H#define __QUERY_SERVER_H#include <echo/TcpServer.h>#include <echo/ThreadPool.h>#include <echo/NonCopyable.h>#include "TextQuery.h"class QueryServer : NonCopyable{ public: QueryServer(const InetAddress &addr, const std::string &filename); void start(); private: void onConnection(const TcpConnectionPtr &conn); void onMessage(const TcpConnectionPtr &conn); void onClose(const TcpConnectionPtr &conn); void query(const std::string s, const TcpConnectionPtr &conn); //转换大小写函数 TcpServer server_; TextQuery query_; ThreadPool pool_;};#endif /*__QUERY_SERVER_H*/#include "QueryServer.h"#include <functional>#include <iostream>#include <fstream>#include <set>#include <stdio.h>using namespace std;using namespace placeholders;QueryServer::QueryServer(const InetAddress &addr, const string &filename) :server_(addr), pool_(1000, 4), query_(filename){// 任务数组的大小 和 线程池的大小 server_.setConnection(bind(&QueryServer::onConnection, this, _1)); server_.setMessage(bind(&QueryServer::onMessage, this, _1)); server_.setClose(bind(&QueryServer::onClose, this, _1));}void QueryServer::onConnection(const TcpConnectionPtr &conn){ conn->send("Hello! Welcome to Echo Server!\r\n");}void QueryServer::onMessage(const TcpConnectionPtr &conn){ string s = conn->receive(); cout << "recv : " << s; pool_.add_task(bind(&QueryServer::query, this, s, conn));}void QueryServer::onClose(const TcpConnectionPtr &conn){ cout << "Client closed\r\n" << endl; conn->shutdown();}void QueryServer::query(const string s, const TcpConnectionPtr &conn){ string word = s; if(word.substr(word.size()-2, 2) == "\r\n"){ word.erase(word.size() - 1); word.erase(word.size() - 1); } string ret = query_.run_query(word); cout << ret; conn->send(ret + "\r\n");}void QueryServer::start(){ pool_.start(); server_.start();}#include "QueryServer.h"using namespace std;int main(int argc, const char *argv[]){ QueryServer server(InetAddress(9999), argv[1]); server.start(); return 0;}2.3 聊天室服务器
2.3.1 使用Echo网络库编写一个简单的聊天室，将一个人的消息转发给所有在线的用户，并且用一个静态变量限制最大连接数。
2.3.2 程序示例
#ifndef __CHAT_SERVER_H#define __CHAT_SERVER_H#include <echo/NonCopyable.h>#include <echo/TcpServer.h>#include <set>class ChatServer : NonCopyable{ public: ChatServer(const InetAddress &addr); void start(); private: void onConnection(const TcpConnectionPtr &conn); void onMessage(const TcpConnectionPtr &conn); void onClose(const TcpConnectionPtr &conn); TcpServer server_; std::set<TcpConnectionPtr> clients_; //用set而不用vector 保证唯一性 const static size_t kMaxClients_ = 3; //用来控制连接数};#endif /*__CHAT_SERVER_H*/#include "ChatServer.h"#include <iostream>using namespace std;using namespace placeholders;ChatServer::ChatServer(const InetAddress &addr) :server_(addr){ server_.setConnection(bind(&ChatServer::onConnection, this, _1)); server_.setMessage(bind(&ChatServer::onMessage, this, _1)); server_.setClose(bind(&ChatServer::onClose, this, _1));}void ChatServer::onConnection(const TcpConnectionPtr &conn){ if(clients_.size() >= kMaxClients_){ conn->send("Chat Server is full, please try later\r\n"); conn->shutdown(); return; } clients_.insert(conn); cout << "New Client online IP: " << conn->getPeerAddress().toIP() << " Port: " << conn->getPeerAddress().toPort() << endl; conn->send("Hello ! Welcome to ChatRoom!\r\n");//}void ChatServer::onMessage(const TcpConnectionPtr &conn){ string s = "recv from " + conn->getPeerAddress().toIP() + " Massage: "; s += conn->receive(); set<TcpConnectionPtr>::iterator it = clients_.begin(); for(; it != clients_.end(); it ++){ (*it)->send(s); }}void ChatServer::onClose(const TcpConnectionPtr &conn){ cout << "client " << conn->getPeerAddress().toIP() << " leave !" << endl; clients_.erase(conn); conn->shutdown();}void ChatServer::start(){ server_.start();}? chat_server? chat_server cat ChatServer.h ChatServer.cpp main.cpp#ifndef __CHAT_SERVER_H#define __CHAT_SERVER_H#include <echo/NonCopyable.h>#include <echo/TcpServer.h>#include <set>class ChatServer : NonCopyable{ public: ChatServer(const InetAddress &addr); void start(); private: void onConnection(const TcpConnectionPtr &conn); void onMessage(const TcpConnectionPtr &conn); void onClose(const TcpConnectionPtr &conn); TcpServer server_; std::set<TcpConnectionPtr> clients_; //用set而不用vector 保证唯一性 const static size_t kMaxClients_ = 3; //用来控制连接数};#endif /*__CHAT_SERVER_H*/#include "ChatServer.h"#include <iostream>using namespace std;using namespace placeholders;ChatServer::ChatServer(const InetAddress &addr) :server_(addr){ server_.setConnection(bind(&ChatServer::onConnection, this, _1)); server_.setMessage(bind(&ChatServer::onMessage, this, _1)); server_.setClose(bind(&ChatServer::onClose, this, _1));}void ChatServer::onConnection(const TcpConnectionPtr &conn){ if(clients_.size() >= kMaxClients_){ conn->send("Chat Server is full, please try later\r\n"); conn->shutdown(); return; } clients_.insert(conn); cout << "New Client online IP: " << conn->getPeerAddress().toIP() << " Port: " << conn->getPeerAddress().toPort() << endl; conn->send("Hello ! Welcome to ChatRoom!\r\n");//}void ChatServer::onMessage(const TcpConnectionPtr &conn){ string s = "recv from " + conn->getPeerAddress().toIP() + " Massage: "; s += conn->receive(); set<TcpConnectionPtr>::iterator it = clients_.begin(); for(; it != clients_.end(); it ++){ (*it)->send(s); }}void ChatServer::onClose(const TcpConnectionPtr &conn){ cout << "client " << conn->getPeerAddress().toIP() << " leave !" << endl; clients_.erase(conn); conn->shutdown();}void ChatServer::start(){ server_.start();}#include "ChatServer.h"int main(int argc, const char *argv[]){ ChatServer server(InetAddress(9999)); server.start(); return 0;}
声明：以上内容来自用户投稿及互联网公开渠道收集整理发布，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任，若内容有误或涉及侵权可进行投诉： 投诉/举报 工作人员会在5个工作日内联系你，一经查实，本站将立刻删除涉嫌侵权内容。