网络IO与IO多路复用

一、网络IO基础

• 系统对象：
  
  
  
  • 网络IO涉及用户空间调用IO的历程或线程以及内核空间的内核系统。
    
  • 例如，当进行read操作时，会经历两个阶段：
    
    
    • 期待数据准备就绪。
      
    • 将数据从内核拷贝到历程或线程中。
      
    
    
  
  
• 多种网络IO模子的出现缘故起因：由于上述两个阶段的不同环境，出现了多种网络IO模子。
  

二、阻塞IO（blocking IO）

• 特点：
  
  
  
  • 在Linux中，默认所有socket都是阻塞的。
    
  • 以读操作为例，当用户历程调用read系统调用，kernel开始IO的第一阶段（准备数据），对于网络IO，数据大概未到，kernel要期待，用户历程会被阻塞。
    
  • 直到kernel比及数据准备好，将数据从kernel拷贝到用户内存并返回结果，用户历程才解除阻塞状态。
    
  • 即阻塞IO在期待数据和拷贝数据两个阶段都阻塞历程。
    
  • 例如，使用listen()、send()、recv()等接口构建服务器/客户机模子，这些接口大多是阻塞型的。
    
  • 代码示例：
    
  
  

1. // 创建服务器端的socket
2. int serverSocket = socket(AF_INET, SOCK_STREAM, 0);
3. // 绑定地址
4. struct sockaddr_in serverAddr;
5. serverAddr.sin_family = AF_INET;
6. serverAddr.sin_port = htons(8080);
7. serverAddr.sin_addr.s_addr = INADDR_ANY;
8. bind(serverSocket, (struct sockaddr*)&serverAddr, sizeof(serverAddr));
9. // 监听连接
10. listen(serverSocket, 5);
11. // 接受连接
12. int clientSocket = accept(serverSocket, NULL, NULL);
13. // 接收数据，在此处进程会阻塞，直到接收到数据或出错
14. char buffer[1024];
15. recv(clientSocket, buffer, sizeof(buffer), 0);
16. // 发送数据，在此处进程也可能阻塞
17. send(clientSocket, buffer, strlen(buffer), 0);

复制代码

1. - 解释：上述代码首先创建了一个服务器端的socket，然后绑定地址和端口，接着监听连接。当调用`accept`时，如果没有连接请求，会阻塞等待。一旦有连接，调用`recv`接收数据时，会阻塞直到有数据到达。发送数据时也可能阻塞，例如网络拥堵或接收方接收缓冲区满。

复制代码

• 多线程/多历程改进方案：
  
  
  
  • 为解决单个毗连阻塞影响其他毗连的问题，可在服务器端使用多线程或多历程。
    
  • 多线程可使用pthread_create()创建，多历程使用fork()创建。
    
  • 多线程开销相对小，适合为较多客户机服务；历程更安全，适合单个服务执行体必要大量CPU资源的环境。
    
  • 例如，服务器端为多个客户机提供服务时，可在主线程期待毗连请求，有毗连时创建新线程或新历程提供服务。
    
  • 但当必要同时响应大量（成百上千）毗连请求时，多线程或多历程会严重占用系统资源，导致系统服从降落和假死。
    
  
  

三、非阻塞IO（non-blocking IO）

• 特点：
  
  
  
  • 通过设置socket为非阻塞，如使用fcntl( fd, F_SETFL, O_NONBLOCK );。
    
  • 当用户历程发出read操作，若kernel数据未准备好，不会阻塞用户历程，而是立刻返回error。
    
  • 用户历程可根据返回结果判断数据是否准备好，未准备好可再次发送read操作。
    
  • 当数据准备好，kernel会将数据拷贝到用户内存并返回。
    
  • 例如，recv()接口在非阻塞状态下调用后立刻返回，返回值有不同含义：
    
    
    
    • recv()返回值大于 0，表示担当数据完毕，返回值是接收到的字节数。
      
    • recv()返回 0，表示毗连正常断开。
      
    • recv()返回 -1，且errno即是EAGAIN，表示recv操作未完成。
      
    • recv()返回 - 1，且errno不即是EAGAIN，表示recv操作碰到系统错误errno。
      
    
    
  • 代码示例：
    
  
  

1. // 创建服务器端的socket
2. int serverSocket = socket(AF_INET, SOCK_STREAM, 0);
3. // 设置为非阻塞
4. fcntl(serverSocket, F_SETFL, O_NONBLOCK);
5. // 绑定地址
6. struct sockaddr_in serverAddr;
7. serverAddr.sin_family = AF_INET;
8. serverAddr.sin_port = htons(8080);
9. serverAddr.sin_addr.s_addr = INADDR_ANY;
10. bind(serverSocket, (struct sockaddr*)&serverAddr, sizeof(serverAddr));
11. // 监听连接
12. listen(serverSocket, 5);
13. // 接受连接
14. int clientSocket = accept(serverSocket, NULL, NULL);
15. // 接收数据，此处不会阻塞，会根据返回值判断状态
16. char buffer[1024];
17. int recvResult;
18. while ((recvResult = recv(clientSocket, buffer, sizeof(buffer), 0)) == -1 && errno == EAGAIN) {
19.     // 数据未准备好，可进行其他操作或再次尝试接收
20. }
21. if (recvResult > 0) {
22.     // 处理接收到的数据
23. } else if (recvResult == 0) {
24.     // 连接断开
25. } else {
26.     // 其他错误
27. }

复制代码

1. - 解释：在这个示例中，将服务器端socket设置为非阻塞后，调用`recv`时不会阻塞进程。如果`recv`返回-1且`errno`为`EAGAIN`，说明数据还未准备好，程序可继续执行其他操作或过段时间再尝试接收，而不是像阻塞IO那样一直等待。

复制代码

四、多路复用IO（IO multiplexing）

• 特点：
  
  
  
  • 又称变乱驱动IO(event driven IO)，如select/epoll。
    
  • 单个历程可同时处置惩罚多个网络毗连的IO，基本原理是select/epoll函数不断轮询所负责的所有socket，当某个socket有数据到达，关照用户历程。
    
  • 流程：用户历程调用select会被阻塞，kernel监视select负责的socket，当有socket数据准备好，select返回，用户历程再调用read操作将数据从kernel拷贝到用户历程。
    
  • 固然使用select必要两个系统调用（select和read），但上风是可在一个线程内处置惩罚多个socket的IO请求。
    
  • 代码示例：
    
  
  

1. #include <stdio.h>
2. #include <sys/select.h>
3. #include <sys/socket.h>
4. #include <netinet/in.h>
5. #include <unistd.h>
7. int main() {
8.     int serverSocket = socket(AF_INET, SOCK_STREAM, 0);
9.     struct sockaddr_in serverAddr;
10.     serverAddr.sin_family = AF_INET;
11.     serverAddr.sin_port = htons(8080);
12.     serverAddr.sin_addr.s_addr = INADDR_ANY;
13.     bind(serverSocket, (struct sockaddr*)&serverAddr, sizeof(serverAddr));
14.     listen(serverSocket, 5);
15.     fd_set readfds;
16.     FD_ZERO(&readfds);
17.     FD_SET(serverSocket, &readfds);
18.     int maxFd = serverSocket;
19.     while (1) {
20.         fd_set tempFds = readfds;
21.         int activity = select(maxFd + 1, &tempFds, NULL, NULL, NULL);
22.         if (FD_ISSET(serverSocket, &tempFds)) {
23.             int clientSocket = accept(serverSocket, NULL, NULL);
24.             FD_SET(clientSocket, &readfds);
25.             if (clientSocket > maxFd) {
26.                 maxFd = clientSocket;
27.             }
28.         }
29.         for (int i = 0; i <= maxFd; i++) {
30.             if (FD_ISSET(i, &tempFds) && i!= serverSocket) {
31.                 char buffer[1024];
32.                 int valread = recv(i, buffer, 1024, 0);
33.                 if (valread == 0) {
34.                     close(i);
35.                     FD_CLR(i, &readfds);
36.                 } else {
37.                     // 处理接收到的数据
38.                 }
39.             }
40.         }
41.     }
42.     return 0;
43. }

复制代码

1. - 解释：首先创建服务器socket，绑定并监听。`FD_ZERO`和`FD_SET`用于初始化和设置文件描述符集合。`select`函数会阻塞，等待集合中文件描述符的可读事件。当`serverSocket`有新连接时，添加新连接的socket到集合，并更新最大文件描述符。当其他socket可读时，接收数据并处理。

复制代码

• 接口原型：
  
  
  
  • FD_ZERO(int fd, fd_set* fds)：初始化fd_set。
    
  • FD_SET(int fd, fd_set* fds)：将句柄添加到fd_set。
    
  • FD_ISSET(int fd, fd_set* fds)：查抄句柄是否在fd_set中。
    
  • FD_CLR(int fd, fd_set* fds)：从fd_set中移除句柄。
    
  • int select(int nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout)：用于探测多个文件句柄的状态变化，readfds、writefds和exceptfds作为输入和输出参数，可设置超时时间。
    
  
  
• 问题：
  
  
  
  • 当必要探测的句柄值较大时，select()接口本身必要大量时间轮询，很多操作系统提供更高效接口如epoll（Linux）、kqueue（BSD）、/dev/poll（Solaris）等。
    
  • 该模子将变乱探测和变乱响应混合，若变乱响应执行体庞大，会降低变乱探测的实时性。
    
  • 可使用变乱驱动库如libevent、libev库解决上述问题。
    
  
  

五、异步IO（Asynchronous I/O）

• 特点：
  
  
  
  • Linux下的异步IO主要用于磁盘IO读写操作，从内核2.6版本开始引入。
    
  • 用户历程发起read操作后可做其他事，kernel收到asynchronous read后立刻返回，不会阻塞用户历程。
    
  • kernel期待数据准备完成，将数据拷贝到用户内存，完成后给用户历程发送信号关照。
    
  • 代码示例（使用aio_read）：
    
  
  

1. #include <aio.h>
2. #include <stdio.h>
3. #include <stdlib.h>
4. #include <unistd.h>
5. #include <fcntl.h>
7. void aio_completion_handler(union sigval sigval) {
8.     struct aiocb *req = (struct aiocb *)sigval.sival_ptr;
9.     if (aio_error(req) == 0) {
10.         char *buffer = (char *)malloc(aio_return(req));
11.         // 处理读取到的数据
12.         free(buffer);
13.     }
14.     aio_destroy(req);
15. }
17. int main() {
18.     int fd = open("test.txt", O_RDONLY);
19.     if (fd < 0) {
20.         perror("open");
21.         return 1;
22.     }
23.     struct aiocb my_aiocb;
24.     memset(&my_aiocb, 0, sizeof(struct aiocb));
25.     my_aiocb.aio_fildes = fd;
26.     my_aiocb.aio_buf = malloc(1024);
27.     my_aiocb.aio_nbytes = 1024;
28.     my_aiocb.aio_offset = 0;
29.     my_aiocb.aio_sigevent.sigev_notify = SIGEV_THREAD;
30.     my_aiocb.aaiocb.sigev_notify_function = aio_completion_handler;
31.     my_aiocb.aio_sigevent.sigev_notify_attributes = NULL;
32.     my_aiocb.aio_sigevent.sigev_value.sival_ptr = &my_aiocb;
33.     if (aio_read(&my_aiocb) < 0) {
34.         perror("aio_read");
35.         close(fd);
36.         return 1;
37.     }
38.     // 进程可做其他事情
39.     sleep(1);
40.     return 0;
41. }

复制代码

1. - 解释：上述代码使用`aio_read`进行异步读操作。首先打开文件，设置`aiocb`结构（包含文件描述符、缓冲区、字节数等），并设置信号通知方式和处理函数。调用`aio_read`后，进程可以继续做其他事情，当读操作完成，会调用`aio_completion_handler`处理结果。

复制代码

• 重要性：异步IO是真正非阻塞的，对高并发网络服务器实现至关重要。
  

六、信号驱动IO（signal driven I/O， SIGIO）

• 特点：
  
  
  
  • 允许套接口进行信号驱动I/O并安装信号处置惩罚函数，历程继续运行不阻塞。
    
  • 当数据准备好，历程收到SIGIO信号，可在信号处置惩罚函数中调用I/O操作函数处置惩罚数据。
    
  • 上风在于期待数据报到达期间，历程可继续执行，制止了select的阻塞与轮询。
    
  
  

七、服务器模子Reactor与Proactor

• Reactor模子：
  
  
  
  • 是一种变乱驱动机制，用于同步I/O。
    
  • 应用步伐将处置惩罚I/O变乱的接口注册到Reactor上，若相应变乱发生，Reactor调用注册的接口（回调函数）。
    
  • 三个重要组件：
    
    
    • 多路复用器：如select、poll、epoll等系统调用。
      
    • 变乱分发器：将多路复用器返回的就绪变乱分到对应的处置惩罚函数。
      
    • 变乱处置惩罚器：负责处置惩罚特定变乱的处置惩罚函数。
      
    
    
  • 具体流程：
    
    
    • 注册读就绪变乱和相应变乱处置惩罚器。
      
    • 变乱分离器期待变乱。
      
    • 变乱到来，激活分离器，分离器调用变乱对应的处置惩罚器。
      
    • 变乱处置惩罚器完成实际的读操作，处置惩罚数据，注册新变乱，返还控制权。
      
    
    
  • 长处：响应快，编程相对简朴，可扩展性和可复用性好。
    
  • 缺点：当步伐必要使用多核资源时会有局限，因为通常是单线程的。
    
  • 代码示例：
    
  
  

1. #include <iostream>
2. #include <sys/epoll.h>
3. #include <sys/socket.h>
4. #include <netinet/in.h>
5. #include <unistd.h>
6. #include <fcntl.h>
7. #include <string.h>
9. class EventHandler {
10. public:
11.     virtual void handleEvent(int fd) = 0;
12. };
14. class Reactor {
15. private:
16.     int epollFd;
17.     std::vector<EventHandler*> handlers;
18. public:
19.     Reactor() {
20.         epollFd = epoll_create1(0);
21.     }
22.     ~Reactor() {
23.         close(epollFd);
24.     }
25.     void registerHandler(int fd, EventHandler* handler) {
26.         struct epoll_event event;
27.         event.data.fd = fd;
28.         event.events = EPOLLIN;
29.         epoll_ctl(epollFd, EPOLL_CTL_ADD, fd, &event);
30.         handlers.push_back(handler);
31.     }
32.     void handleEvents() {
33.         struct epoll_event events[10];
34.         int numEvents = epoll_wait(epollFd, events, 10, -1);
35.         for (int i = 0; i < numEvents; i++) {
36.             int fd = events[i].data.fd;
37.             for (EventHandler* handler : handlers) {
38.                 handler->handleEvent(fd);
39.             }
40.         }
41.     }
42. };
44. class EchoHandler : public EventHandler {
45. public:
46.     void handleEvent(int fd) override {
47.         char buffer[1024];
48.         int bytesRead = recv(fd, buffer, sizeof(buffer), 0);
49.         if (bytesRead > 0) {
50.             send(fd, buffer, bytesRead, 0);
51.         }
52.     }
53. };
55. int main() {
56.     Reactor reactor;
57.     int serverSocket = socket(AF_INET, SOCK_STREAM, 0);
58.     struct sockaddr_in serverAddr;
59.     serverAddr.sin_family = AF_INET;
60.     serverAddr.sin_port = htons(8080);
61.     serverAddr.sin_addr.s_addr = INADDR_ANY;
62.     bind(serverSocket, (struct sockaddr*)&serverAddr, sizeof(serverAddr));
63.     listen(serverSocket, 5);
64.     EchoHandler handler;
65.     reactor.registerHandler(serverSocket, &handler);
66.     while (1) {
67.         reactor.handleEvents();
68.     }
69.     return 0;
70. }

复制代码

1. - 解释：上述C++代码中，`Reactor`类负责创建`epoll`，注册和处理事件。`EventHandler`是抽象基类，`EchoHandler`是具体处理接收和发送的派生类。`main`函数创建服务器socket，注册`EchoHandler`到`Reactor`，并不断调用`handleEvents`处理事件。

复制代码

• Proactor模子：
  
  
  
  • 最大特点是使用异步I/O，所有I/O操作都交由系统的异步I/O接口执行，工作线程只负责业务逻辑。
    
  • 具体流程：
    
    
    • 处置惩罚器发起异步操作并关注I/O完成变乱。
      
    • 变乱分离器期待操作完成变乱。
      
    • 分离器期待时，内核并行执行实际I/O操作并将结果存入用户缓冲区，关照分离器读操作完成。
      
    • I/O完成后，通过变乱分离器召唤处置惩罚器。
      
    • 变乱处置惩罚器处置惩罚用户缓冲区中的数据。
      
    
    
  • 增加了编程复杂度，但给工作线程带来更高服从，可使用系统态的读写优化。
    
  • 在Windows上常用IOCP支持高并发，Linux上因aio性能不佳，主要以Reactor模子为主。
    
  • 也可使用Reactor模拟Proactor，但在读写并行本领上会有区别。
    
  
  

八、同步I/O和异步I/O的区别总结

• 阻塞与非阻塞IO的区别：
  
  
  
  • 调用阻塞IO会阻塞历程直到操作完成，非阻塞IO在kernel准备数据时会立刻返回。
    
  
  
• 同步与异步IO的区别：
  
  
  
  • 同步IO在做“IO operation”（如read系统调用）时会阻塞历程。阻塞IO、非阻塞IO、多路复用IO都属于同步IO。
    
  • 非阻塞IO在数据准备好时的拷贝数据阶段会阻塞历程；而异步IO在整个过程中，历程不会被阻塞，历程发起IO操作后直接做其他事，直到kernel发送信号关照完成。
    
  
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。