亮白风格-图解网络-小林coding-v2.0.pdf 暗黑风格-图解网络-小林coding-v2.0.pdf

select poll epoll 都是linux系统提供的 IO多路复用的函数

  1. #include<sys/select.h>
  2. #include<sys/poll.h>
  3. #include<sys/epoll.h>

1. 从操作系统角度看多路复用

点击查看【processon】

2. epoll与select, poll区别

  • epoll每次调用只会向多路复用器传入新需要关注的文件描述符
  • select和poll, 每次调用都会把所有需要关注的文件描述符都传进去, 非常浪费空间
  • select 入参是个数组, select返回的是含有整个句柄的数组，应用程序需要遍历整个数组才能发现哪些句柄发生了事件；
  • poll 入参是个链表
  • select 和 poll 是水平触发, 只要有IO数据可读, 就会触发通知
  • epoll开辟了一个内存空间, 专门保存需要监听的文件描述符(红黑树)
  • epoll是边缘触发, 只有新的IO活动到来, 才会触发通知; 即使通道中有数据也不会再触发

举个例子:
第一轮轮询

  • 需要关注100个文件描述符fd, epoll传入100个, select和poll也是传入100个

第二轮轮询

  • 我只需要在原来的100个基础上再多关注1个, 那么第二次轮询
  • epoll只需要传那新的一个就行了;
  • 而select和poll函数需要传入101个文件描述符

image.png

3. epoll_wait函数

函数声明:

  1. int epoll_wait(int epfd,struct epoll_event * events,int maxevents,int timeout)

该函数用于轮询I/O事件的发生；
参数：

  1. epfd :由epoll_create 生成的epoll专用的文件描述符；
  2. epoll_event:用于回传代处理事件的数组；
  3. maxevents:每次能处理的事件数；
  4. timeout:等待I/O事件发生的超时值(单位我也不太清楚)； -1相当于阻塞，0相当于非阻塞。一般用 -1 即可返回发生事件数。

4. 用C实现服务端

启动:

  1. socket(); 创建一个文件描述符, 例如4
  2. bind(); 将这个文件描述符, 绑定到端口上, 例如4绑定到9090
  3. listen(); 开始监听9090端口的网络事件
  4. epoll_create(); 创建一个多路复用器的空间, 例如这个空间用文件描述符8表示; 相当于java中的 Selector.open()
  5. 5. epoll_ctl(8,EPOLL_CTL_ADD,4,{EPOLLIN,{u32=4,u64=12412342345234345345345}})=0; 把监听网络的文件描述符4, 注册到文件描述符8的空间中, 并标识监听 EPOLLIN 事件; 相当于java的server.register(selector, SelectorKey.OP_ACCEPT);

当新客户端连接发生的时候:

  1. epoll_wait(8,{{EPOLLIN,{u32=4,u64=4234234234234234}},4096,-1})=1 入参最后一个参数是timeout,-1是阻塞,0相当于非阻塞, 大于0的数相当于等待时间, 也就是说可以阻塞也可以非阻塞,正好对应了 java中的: select.select(0); 注意: epoll_wait()函数的入参中有个是指针, 函数会把发生事件的文件描述符存入那个指针的epoll_event结构体中 相当于java中的 select.select(0); 返回值=1 是告诉你有几个新的连接
  2. accept(4, .......) = 9 //9就是新拿到的文件描述符, 相当于java中的 (ServerSocketChannel)(key.channel()).accept(); 方法那么拿到的这个9就是新的连接, 分配到的文件描述符!!! 在java中相当于拿到了一个新的客户端的连接

客户端发送数据后:

  1. epoll_ctl(8, EPOLL_CTL_ADD, 9, {EPOLLIN, {....}}) = 0; 8是多路复用器的文件描述符, 9是新的客户端连接的文件描述符; 这个操作由java中的 client.register(selector, SelectionKey.OP_READ, buffer); 函数完成; 功能是: 把客户端的文件描述符9, 放入多路复用器8的空间中, 注册监听9的读取事件

  2. 这时候客户端发送数据; 服务端: epoll_wait(7, 接收文件描述符的结构体指针, -1) =1; 相当于java的, select.select(); ;那么C的函数,会把发生事件的文件描述符, 存入那个指针指向的结构体实例中; 我们就能从selector.selectedKeys()中找到那个客户端的文件描述符

  3. 然后服务端就可以读取socket缓冲区的数据了; C函数: read(8, ....); ; java中, 是直接读取到了buffer中

服务端发送数据:

  1. 服务端向客户端发送数据 C函数: write(8, ....);

5. 示例

  1. #include <netinet/in.h> // sockaddr_in
  2. #include <sys/types.h>  // socket
  3. #include <sys/socket.h> // socket
  4. #include <arpa/inet.h>
  5. #include <unistd.h>
  6. #include <sys/epoll.h> // epoll
  7. #include <sys/ioctl.h>
  8. #include <sys/time.h>
  9. #include <iostream>
  10. #include <vector>
  11. #include <string>
  12. #include <cstdlib>
  13. #include <cstdio>
  14. #include <cstring>
  15. using namespace std;
  16. #define BUFFER_SIZE 1024
  17. #define EPOLLSIZE 100
  19. struct PACKET_HEAD
  20. {
  21.     int length;
  22. };
  24. class Server
  25. {
  26. private:
  27.     struct sockaddr_in server_addr;
  28.     socklen_t server_addr_len;
  29.     int listen_fd;                        // 监听的fd
  30.     int epfd;                             // epoll fd
  31.     struct epoll_event events[EPOLLSIZE]; // epoll_wait返回的就绪事件
  32. public:
  33.     Server(int port);
  34.     ~Server();
  35.     void Bind();
  36.     void Listen(int queue_len = 20);
  37.     void Accept();
  38.     void Run();
  39.     void Recv(int fd);
  40. };
  42. Server::Server(int port)
  43. {
  44.     bzero(&server_addr, sizeof(server_addr));
  45.     server_addr.sin_family = AF_INET;
  46.     server_addr.sin_addr.s_addr = htons(INADDR_ANY);
  47.     server_addr.sin_port = htons(port);
  48.     // create socket to listen
  49.     listen_fd = socket(PF_INET, SOCK_STREAM, 0);
  50.     if (listen_fd < 0)
  51.     {
  52.         cout << "Create Socket Failed!";
  53.         exit(1);
  54.     }
  55.     int opt = 1;
  56.     setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
  57. }
  59. Server::~Server()
  60. {
  61.     close(epfd);
  62. }
  64. void Server::Bind()
  65. {
  66.     if (-1 == (bind(listen_fd, (struct sockaddr *)&server_addr, sizeof(server_addr))))
  67.     {
  68.         cout << "Server Bind Failed!";
  69.         exit(1);
  70.     }
  71.     cout << "Bind Successfully.\n";
  72. }
  74. void Server::Listen(int queue_len)
  75. {
  76.     if (-1 == listen(listen_fd, queue_len))
  77.     {
  78.         cout << "Server Listen Failed!";
  79.         exit(1);
  80.     }
  81.     cout << "Listen Successfully.\n";
  82. }
  84. void Server::Accept()
  85. {
  86.     struct sockaddr_in client_addr;
  87.     socklen_t client_addr_len = sizeof(client_addr);
  89.     int new_fd = accept(listen_fd, (struct sockaddr *)&client_addr, &client_addr_len);
  90.     if (new_fd < 0)
  91.     {
  92.         cout << "Server Accept Failed!";
  93.         exit(1);
  94.     }
  96.     cout << "new connection was accepted.\n";
  98.     // 在epfd中注册新建立的连接
  99.     struct epoll_event event;
  100.     event.data.fd = new_fd;
  101.     event.events = EPOLLIN;
  103.     epoll_ctl(epfd, EPOLL_CTL_ADD, new_fd, &event);
  104. }
  106. void Server::Run()
  107. {
  108.     epfd = epoll_create(1); // 创建epoll句柄
  110.     struct epoll_event event;
  111.     event.data.fd = listen_fd;
  112.     event.events = EPOLLIN;
  113.     epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd, &event); // 注册listen_fd
  115.     while (1)
  116.     {
  117.         int nums = epoll_wait(epfd, events, EPOLLSIZE, -1);
  118.         if (nums < 0)
  119.         {
  120.             cout << "poll() error!";
  121.             exit(1);
  122.         }
  124.         if (nums == 0)
  125.         {
  126.             continue;
  127.         }
  129.         for (int i = 0; i < nums; ++i) // 遍历所有就绪事件
  130.         {
  131.             int fd = events[i].data.fd;
  132.             if ((fd == listen_fd) && (events[i].events & EPOLLIN))
  133.                 Accept(); // 有新的客户端请求
  134.             else if (events[i].events & EPOLLIN)
  135.                 Recv(fd); // 读数据
  136.             else
  137.                 ;
  138.         }
  139.     }
  140. }
  142. void Server::Recv(int fd)
  143. {
  144.     bool close_conn = false; // 标记当前连接是否断开了
  146.     PACKET_HEAD head;
  147.     recv(fd, &head, sizeof(head), 0); // 先接受包头，即数据总长度
  149.     char *buffer = new char[head.length];
  150.     bzero(buffer, head.length);
  151.     int total = 0;
  152.     while (total < head.length)
  153.     {
  154.         int len = recv(fd, buffer + total, head.length - total, 0);
  155.         if (len < 0)
  156.         {
  157.             cout << "recv() error!";
  158.             close_conn = true;
  159.             break;
  160.         }
  161.         total = total + len;
  162.     }
  164.     if (total == head.length) // 将收到的消息原样发回给客户端
  165.     {
  166.         int ret1 = send(fd, &head, sizeof(head), 0);
  167.         int ret2 = send(fd, buffer, head.length, 0);
  168.         if (ret1 < 0 || ret2 < 0)
  169.         {
  170.             cout << "send() error!";
  171.             close_conn = true;
  172.         }
  173.     }
  175.     delete buffer;
  177.     if (close_conn) // 当前这个连接有问题，关闭它
  178.     {
  179.         close(fd);
  180.         struct epoll_event event;
  181.         event.data.fd = fd;
  182.         event.events = EPOLLIN;
  183.         epoll_ctl(epfd, EPOLL_CTL_DEL, fd, &event); // Delete一个fd
  184.     }
  185. }
  187. int main()
  188. {
  189.     Server server(15000);
  190.     server.Bind();
  191.     server.Listen();
  192.     server.Run();
  193.     return 0;
  194. }

这有几个文章不错 https://www.cnblogs.com/apprentice89/p/3234677.html https://www.cnblogs.com/apprentice89/archive/2013/05/06/3063039.html https://zhuanlan.zhihu.com/p/422229233 https://gcmurphy.wordpress.com/2012/11/30/using-epoll-in-go/