多路转接select目录多路转接select一、初识select1.1.select的概念1.2.select函数1.3.select的特点1.4.Select服务器实现二、初识poll2.1.poll的作用与定位2.2.poll的接口2.3.poll的特点2.4.Poll服务器实现一、初识select1.1.select的概念一种通过等待多个文件描述符的就绪事件通知机制一旦有一个或多个文件描述符的事件就绪select就会通知上层告诉调用方哪些文件描述符可以IO用来实现多路复用的模型底层有数据读事件就绪底层有空间写事件就绪对于最开始的文件描述符接收和发送缓冲区都是空的所以一般默认读事件是不就绪的默认写事件是就绪的1.2.select函数参数1需要监视的最大文件描述符值 1参数2只关心读事件fd_set内核提供给用户的位图结构一次可以给fd_set添加多个文件描述符输入型比特位的位置表示文件描述符的编号比特位的内容表示用户告诉内核要关心哪些文件描述符上的读事件输出型比特位的位置表示文件描述符的编号比特位的内容表示内核告诉用户要关心的文件描述符哪些读事件就绪参数3只关心写事件参数4只关心异常事件参数5设置超时时间秒数微秒1s 1000000μms输入型设为正常事件等待多个文件描述符超过该时间后没有文件描述符就绪立即返回设为NULL则表示一直阻塞不返回直到有任一文件描述符就绪阻塞设为{00}在检测完一轮后如果没有文件描述符就绪就会立即返回非阻塞输出型如果在设置时间内没有文件描述符就绪select返回后就会被设为{00}如果在设置时间内有文件描述符就绪select返回后就会设为剩余时间返回值大于0表示几个文件描述符就绪等于0表示超时在设定时间内没有文件描述符就绪小于0检测到非法的文件描述符直接报错FD_CLR从集合中移除指定的文件描述符FD_ISSET判断文件描述符是否在集合里FD_SET在指定集合设置一个文件描述符FD_ZERO清空指定的集合1.3.select的特点每次调用select都需要手动对fds进行重置每次调用select都需要把fds从用户态拷贝到内核态每次调用select都需要在内核遍历所有的文件描述符select支持的文件描述符数量有限且数量太小1.4.Select服务器实现SelectServer.hpp#pragma once #include iostream #include memory #include unistd.h #include Socket.hpp #include Log.hpp using namespace SocketModule; using namespace LogModule; class SelectServer { const static int size sizeof(fd_set) * 8; const static int defaultfd -1; public: SelectServer(int port) : _listensock(std::make_uniqueTcpSocket()), _isrunning(false) { _listensock-BuildTcpSocketMethod(port); for (int i 0; i size; i) { _fd_array[i] defaultfd; } _fd_array[0] _listensock-Fd(); } void Start() { _isrunning true; while (_isrunning) { // listensockfd也是一个fd, Accept如果没有收到连接也会发生阻塞 // 需要将listensockfd添加到select函数设置关心读事件是否就绪 // auto res _listensock-Accept(); fd_set rfds; // 定义fds集合 FD_ZERO(rfds); // 清空fds集合 int maxfd defaultfd; for (int i 0; i size; i) { if (_fd_array[i] defaultfd) { continue; } // 重置rfds FD_SET(_fd_array[i], rfds); // 更新最大文件描述符 if (maxfd _fd_array[i]) { maxfd _fd_array[i]; } } PrintFd(); // FD_SET(_listensock-Fd(), rfds); // struct timeval timeout {2, 0}; // 每次调用select, 都要对rfds重置 // 最大文件描述符一定是变化的 int n select(maxfd 1, rfds, nullptr, nullptr, nullptr); // 事件就绪检测 switch (n) { case -1: LOG(LogLevel::ERROR) select error; break; case 0: LOG(LogLevel::INFO) time out...; break; default: LOG(LogLevel::DEBUG) 有事件就绪... n : n; Dispatcher(rfds); // 处理就绪的事件 break; } } _isrunning false; } // 事件派发器 void Dispatcher(fd_set rfds) { for (int i 0; i size; i) { // 判断文件描述符合法 if (_fd_array[i] defaultfd) { continue; } // 判断文件描述符就绪 if (FD_ISSET(_fd_array[i], rfds)) { if (_fd_array[i] _listensock-Fd()) { // 1. 新连接到来 Accepter(); } else { // 2. 读事件就绪 Recver(_fd_array[i], i); } } } } // IO处理器 void Recver(int fd, int pos) { char buffer[1024]; ssize_t n recv(fd, buffer, sizeof(buffer) - 1, 0); if (n 0) { buffer[n] 0; std::cout client say buffer std::endl; } else if (n 0) { LOG(LogLevel::INFO) client quit...; // 在辅助数组中移除文件描述符 _fd_array[pos] defaultfd; // 关闭文件描述符 close(fd); } else { LOG(LogLevel::ERROR) recv error; // 在辅助数组中移除文件描述符 _fd_array[pos] defaultfd; // 关闭文件描述符 close(fd); } } // 连接管理器 void Accepter() { InetAddr client; int sockfd _listensock-Accept(client); if (sockfd 0) { // 获取新连接 LOG(LogLevel::INFO) get a new link, sockfd: sockfd , client is client.StringAddr(); // 将新连接托管给select, 只需将新的文件描述符添加到辅助数组 int pos 0; for (; pos size; pos) { if (_fd_array[pos] defaultfd) { break; } } if (pos size) { LOG(LogLevel::WARNING) select server full; close(sockfd); } else { _fd_array[pos] sockfd; } } } void PrintFd() { std::cout fd_array[]: ; for (int i 0; i size; i) { if (_fd_array[i] defaultfd) { continue; } std::cout _fd_array[i] ; } std::cout \r\n; } void Stop() { _isrunning false; } ~SelectServer() { } private: std::unique_ptrSocket _listensock; bool _isrunning; int _fd_array[size]; // 辅助数组 };socket.hpp#pragma once #include iostream #include string #include unistd.h #include sys/socket.h #include sys/types.h #include netinet/in.h #include arpa/inet.h #include cstdlib #include Log.hpp #include Common.hpp #include InetAddr.hpp namespace SocketModule { using namespace LogModule; const static int gbacklog 16; // 设计模式: 模板方法模式 // 基类socket // 大部分方法为纯虚方法 class Socket { public: virtual ~Socket() {} // 析构函数 virtual void SocketOrDie() 0; // 创建套接字 virtual void BindOrDie(uint16_t port) 0; // 绑定IP和端口号 virtual void ListenOrDie(int backlog) 0; // 监听 // virtual std::shared_ptrSocket Accept(InetAddr *client) 0; // 获取 virtual int Accept(InetAddr *client) 0; // 获取 virtual void Close() 0; // 关闭 virtual int Recv(std::string *out) 0; // 读取 virtual int Send(const std::string message) 0; // 写入 virtual int Connect(const std::string server_ip, uint16_t port) 0; // 连接 virtual int Fd() 0; public: void BuildTcpSocketMethod(uint16_t port, int backlog gbacklog) { SocketOrDie(); BindOrDie(port); ListenOrDie(backlog); } void BuildTcpClientSocketMethod() { SocketOrDie(); } }; const static int defaultfd -1; class TcpSocket : public Socket { public: TcpSocket() : _sockfd(defaultfd) { } TcpSocket(int fd) : _sockfd(fd) { } void SocketOrDie() override { _sockfd ::socket(AF_INET, SOCK_STREAM, 0); if (_sockfd 0) { LOG(LogLevel::FATAL) socket error; exit(SOCKET_ERR); } int opt 1; setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, opt, sizeof(opt)); LOG(LogLevel::INFO) socket success: _sockfd; } void BindOrDie(uint16_t port) override { InetAddr localaddr(port); int n ::bind(_sockfd, localaddr.NetAddrPtr(), localaddr.NetAddrLen()); if (n 0) { LOG(LogLevel::FATAL) bind error; exit(BIND_ERR); } LOG(LogLevel::INFO) bind success; } void ListenOrDie(int backlog) override { int n ::listen(_sockfd, backlog); if (n 0) { LOG(LogLevel::FATAL) listen error; exit(LISTEN_ERR); } int opt 1; setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, opt, sizeof(opt)); // 设置地址复用 LOG(LogLevel::INFO) listen success; } // std::shared_ptrSocket Accept(InetAddr *client) override int Accept(InetAddr *client) override { struct sockaddr_in peer; socklen_t len sizeof(peer); int fd ::accept(_sockfd, CONV(peer), len); if (fd 0) { LOG(LogLevel::WARNING) accept warning...; return -1; } // client-SetAddr(peer); // return std::make_sharedTcpSocket(fd); return fd; } void Close() override { if (_sockfd 0) { ::close(_sockfd); } } int Recv(std::string *out) override { // 流式读取不用关心读到的是什么 char buffer[4096]; ssize_t n ::recv(_sockfd, buffer, sizeof(buffer) - 1, 0); if (n 0) { buffer[n] 0; *out buffer; } return n; } int Send(const std::string message) override { return ::send(_sockfd, message.c_str(), message.size(), 0); } int Connect(const std::string server_ip, uint16_t port) override { InetAddr server(server_ip, port); return ::connect(_sockfd, server.NetAddrPtr(), server.NetAddrLen()); } int Fd() override { return _sockfd; } ~TcpSocket() { } private: int _sockfd; }; };Main.cc#include SelectServer.hpp int main(int argc, char *argv[]) { if (argc ! 2) { std::cout Usage: argv[0] port std::endl; exit(USAGE_ERR); } Enable_Console_Log_Strategy(); uint16_t port std::stoi(argv[1]); std::unique_ptrSelectServer svr std::make_uniqueSelectServer(port); svr-Start(); return 0; }二、初识poll一个进程能打开的文件描述符个数是有上限的但linux内核支持对文件描述符表进行动态扩展2.1.poll的作用与定位poll只负责等待一次可以等待多个文件描述符事件就绪对上层进行事件通知2.2.poll的接口参数1数组的起始地址参数2数组元素的个数参数3设置超时的时间当poll调用时fd和events有效用户告诉内核需要关心fd上面的events事件当poll返回时fd和revents有效内核告诉用户要关心的fd上的events事件已经就绪events和events取值事件位图poll输入与输出参数分离不用在poll之前重置参数poll的文件描述符数量没有上限用户可以进行扩容内核依旧会对文件描述符遍历如果非法会自动不关心2.3.poll的特点poll没有最大数量的限制使用pollfd指针对文件描述符与事件进行管理每次调用poll需要把大量的pollfd结构体从用户拷贝到内核poll返回后也需要通过轮询来获取pollfd中就绪的文件描述符同时连接大量客户端但在某时刻只有很少处于就绪状态效率会下降2.4.Poll服务器实现#pragma once #include iostream #include memory #include unistd.h #include sys/poll.h #include Socket.hpp #include Log.hpp using namespace SocketModule; using namespace LogModule; class PollServer { const static int size 4096; const static int defaultfd -1; public: PollServer(int port) : _listensock(std::make_uniqueTcpSocket()), _isrunning(false) { _listensock-BuildTcpSocketMethod(port); for (int i 0; i size; i) { _fds[i].fd defaultfd; _fds[i].events 0; _fds[i].revents 0; } _fds[0].fd _listensock-Fd(); _fds[0].events | POLLIN; } void Start() { int timeout 1000; // ms _isrunning true; while (_isrunning) { PrintFd(); int n poll(_fds, size, timeout); switch (n) { case -1: LOG(LogLevel::ERROR) poll error; break; case 0: LOG(LogLevel::INFO) poll time out...; break; default: LOG(LogLevel::DEBUG) 有事件就绪... n : n; Dispatcher(); // 处理就绪的事件 break; } } _isrunning false; } // 事件派发器 void Dispatcher() { for (int i 0; i size; i) { // 判断文件描述符合法 if (_fds[i].fd defaultfd) { continue; } // 判断文件描述符就绪 if (_fds[i].revents POLLIN) { if (_fds[i].fd _listensock-Fd()) { // 1. 新连接到来 Accepter(); } else { // 2. 读事件就绪 Recver(i); } } } } // IO处理器 void Recver(int pos) { char buffer[1024]; ssize_t n recv(_fds[pos].fd, buffer, sizeof(buffer) - 1, 0); if (n 0) { buffer[n] 0; std::cout client say buffer std::endl; } else if (n 0) { LOG(LogLevel::INFO) client quit...; // 关闭文件描述符 close(_fds[pos].fd); // 不再关心该描述符 _fds[pos].fd defaultfd; _fds[pos].events 0; _fds[pos].revents 0; } else { LOG(LogLevel::ERROR) recv error; // 关闭文件描述符 close(_fds[pos].fd); // 不再关心该描述符 _fds[pos].fd defaultfd; _fds[pos].events 0; _fds[pos].revents 0; } } // 连接管理器 void Accepter() { InetAddr client; int sockfd _listensock-Accept(client); if (sockfd 0) { // 获取新连接 LOG(LogLevel::INFO) get a new link, sockfd: sockfd , client is client.StringAddr(); // 将新连接托管给select, 只需将新的文件描述符添加到辅助数组 int pos 0; for (; pos size; pos) { if (_fds[pos].fd defaultfd) { break; } } if (pos size) { LOG(LogLevel::WARNING) poll server full; close(sockfd); // 如果用扩容方案, 就可以进行扩容 } else { _fds[pos].fd sockfd; _fds[pos].events | POLLIN; _fds[pos].revents 0; } } } void PrintFd() { std::cout _fds[]: ; for (int i 0; i size; i) { if (_fds[i].fd defaultfd) { continue; } std::cout _fds[i].fd ; } std::cout \r\n; } void Stop() { _isrunning false; } ~PollServer() { } private: std::unique_ptrSocket _listensock; bool _isrunning; struct pollfd _fds[size]; // struct pollfd *_fds; 动态扩容 };