基于IO多路复用的异步非阻塞网络编程模型
2022-10-305.8k 阅读
一、理解IO多路复用
在传统的网络编程中，每个连接通常需要一个独立的线程或进程来处理I/O操作。这种方式在连接数量较少时表现良好，但随着连接数的增加，系统资源的消耗会急剧上升，因为每个线程或进程都需要占用一定的内存空间，并且线程间的上下文切换也会带来额外的开销。
IO多路复用技术则提供了一种更高效的解决方案。它允许一个线程或进程同时监控多个文件描述符（在网络编程中，套接字就是一种文件描述符），当其中任何一个文件描述符就绪（即可以进行读或写操作）时，程序能够被通知并进行相应处理。这大大减少了系统资源的占用，提高了应用程序的并发处理能力。
常见的IO多路复用机制有 select、poll 和 epoll 等。下面我们分别来看它们的特点。
1.1 select
select 是最早出现的IO多路复用机制，它通过一个 fd_set 结构体来管理文件描述符集合，通过 select 函数来监控这些文件描述符。select 函数的原型如下：
int select(int nfds, fd_set *readfds, fd_set *writefds,
fd_set *exceptfds, struct timeval *timeout);
nfds：需要检查的文件描述符集合中最大文件描述符的值加1。readfds、writefds、exceptfds：分别是需要检查的可读、可写和异常的文件描述符集合。timeout：设置等待的超时时间，如果为NULL则表示无限等待。
select 函数返回时，会修改这三个 fd_set 集合，只保留就绪的文件描述符。应用程序需要遍历整个集合来找出哪些文件描述符就绪。这种方式的缺点很明显，首先 fd_set 集合的大小有限制（通常为1024），其次每次调用 select 都需要将整个集合从用户空间拷贝到内核空间，并且遍历集合的时间复杂度为 O(n)，随着文件描述符数量的增加，性能会急剧下降。
下面是一个简单的 select 示例代码：
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <string.h>
#include <sys/types.h>
#include <sys/time.h>
#define PORT 8888
#define MAX_CLIENTS 10
int main() {
int server_socket, client_socket;
struct sockaddr_in server_addr, client_addr;
socklen_t client_addr_len = sizeof(client_addr);
char buffer[1024];
// 创建套接字
server_socket = socket(AF_INET, SOCK_STREAM, 0);
if (server_socket == -1) {
perror("Socket creation failed");
exit(EXIT_FAILURE);
}
// 初始化服务器地址
server_addr.sin_family = AF_INET;
server_addr.sin_port = htons(PORT);
server_addr.sin_addr.s_addr = INADDR_ANY;
// 绑定套接字到地址
if (bind(server_socket, (struct sockaddr *)&server_addr, sizeof(server_addr)) == -1) {
perror("Bind failed");
close(server_socket);
exit(EXIT_FAILURE);
}
// 监听连接
if (listen(server_socket, MAX_CLIENTS) == -1) {
perror("Listen failed");
close(server_socket);
exit(EXIT_FAILURE);
}
fd_set read_fds;
FD_ZERO(&read_fds);
FD_SET(server_socket, &read_fds);
fd_set tmp_fds = read_fds;
while (1) {
int activity = select(server_socket + 1, &tmp_fds, NULL, NULL, NULL);
if (activity == -1) {
perror("Select error");
break;
} else if (activity > 0) {
if (FD_ISSET(server_socket, &tmp_fds)) {
// 有新连接
client_socket = accept(server_socket, (struct sockaddr *)&client_addr, &client_addr_len);
if (client_socket == -1) {
perror("Accept failed");
continue;
}
printf("New client connected: %s:%d\n", inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port));
FD_SET(client_socket, &read_fds);
}
for (int i = 0; i <= server_socket; i++) {
if (FD_ISSET(i, &tmp_fds)) {
if (i != server_socket) {
// 客户端有数据可读
int bytes_read = recv(i, buffer, sizeof(buffer), 0);
if (bytes_read <= 0) {
// 客户端关闭连接
printf("Client disconnected\n");
close(i);
FD_CLR(i, &read_fds);
} else {
buffer[bytes_read] = '\0';
printf("Received from client: %s\n", buffer);
send(i, buffer, strlen(buffer), 0);
}
}
}
}
}
tmp_fds = read_fds;
}
close(server_socket);
return 0;
}
1.2 poll
poll 是对 select 的改进，它使用一个 pollfd 结构体数组来管理文件描述符，poll 函数的原型如下：
int poll(struct pollfd *fds, nfds_t nfds, int timeout);
fds：一个pollfd结构体数组，每个pollfd结构体包含文件描述符、需要检查的事件和实际发生的事件。nfds：数组中元素的个数。timeout：等待的超时时间，单位为毫秒。
与 select 相比，poll 没有了文件描述符数量的限制，并且每次调用 poll 只需要将新添加或修改的 pollfd 结构体从用户空间拷贝到内核空间，性能有所提升。但它仍然需要遍历整个数组来找出就绪的文件描述符，时间复杂度依然为 O(n)。
以下是一个 poll 的示例代码：
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <string.h>
#include <poll.h>
#define PORT 8888
#define MAX_CLIENTS 10
int main() {
int server_socket, client_socket;
struct sockaddr_in server_addr, client_addr;
socklen_t client_addr_len = sizeof(client_addr);
char buffer[1024];
// 创建套接字
server_socket = socket(AF_INET, SOCK_STREAM, 0);
if (server_socket == -1) {
perror("Socket creation failed");
exit(EXIT_FAILURE);
}
// 初始化服务器地址
server_addr.sin_family = AF_INET;
server_addr.sin_port = htons(PORT);
server_addr.sin_addr.s_addr = INADDR_ANY;
// 绑定套接字到地址
if (bind(server_socket, (struct sockaddr *)&server_addr, sizeof(server_addr)) == -1) {
perror("Bind failed");
close(server_socket);
exit(EXIT_FAILURE);
}
// 监听连接
if (listen(server_socket, MAX_CLIENTS) == -1) {
perror("Listen failed");
close(server_socket);
exit(EXIT_FAILURE);
}
struct pollfd fds[MAX_CLIENTS + 1];
fds[0].fd = server_socket;
fds[0].events = POLLIN;
int nfds = 1;
while (1) {
int activity = poll(fds, nfds, -1);
if (activity == -1) {
perror("Poll error");
break;
} else if (activity > 0) {
if (fds[0].revents & POLLIN) {
// 有新连接
client_socket = accept(server_socket, (struct sockaddr *)&client_addr, &client_addr_len);
if (client_socket == -1) {
perror("Accept failed");
continue;
}
printf("New client connected: %s:%d\n", inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port));
fds[nfds].fd = client_socket;
fds[nfds].events = POLLIN;
nfds++;
}
for (int i = 1; i < nfds; i++) {
if (fds[i].revents & POLLIN) {
// 客户端有数据可读
int bytes_read = recv(fds[i].fd, buffer, sizeof(buffer), 0);
if (bytes_read <= 0) {
// 客户端关闭连接
printf("Client disconnected\n");
close(fds[i].fd);
for (int j = i; j < nfds - 1; j++) {
fds[j] = fds[j + 1];
}
nfds--;
i--;
} else {
buffer[bytes_read] = '\0';
printf("Received from client: %s\n", buffer);
send(fds[i].fd, buffer, strlen(buffer), 0);
}
}
}
}
}
close(server_socket);
return 0;
}
1.3 epoll
epoll 是 Linux 特有的IO多路复用机制，它在性能上比 select 和 poll 有了质的飞跃。epoll 采用事件驱动的方式，当有文件描述符就绪时，内核会直接将就绪的文件描述符通知给应用程序，而不需要应用程序遍历整个文件描述符集合。
epoll 有三个主要的函数：epoll_create、epoll_ctl 和 epoll_wait。
epoll_create：创建一个 epoll 实例，返回一个 epoll 文件描述符。
int epoll_create(int size);
epoll_ctl：用于控制某个 epoll 实例，例如添加、修改或删除文件描述符。
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
- `epfd`：epoll 文件描述符。
- `op`：操作类型，如 `EPOLL_CTL_ADD`、`EPOLL_CTL_MOD`、`EPOLL_CTL_DEL`。
- `fd`：要操作的文件描述符。
- `event`：指向 `epoll_event` 结构体的指针，用于指定事件和数据。
epoll_wait：等待事件的发生，返回就绪的文件描述符数量。
int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);
- `epfd`：epoll 文件描述符。
- `events`：用于存放就绪事件的数组。
- `maxevents`：数组的大小。
- `timeout`：等待的超时时间，单位为毫秒。
下面是一个 epoll 的示例代码：
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <string.h>
#include <sys/epoll.h>
#define PORT 8888
#define MAX_CLIENTS 10
#define EVENTS_SIZE 10
int main() {
int server_socket, client_socket;
struct sockaddr_in server_addr, client_addr;
socklen_t client_addr_len = sizeof(client_addr);
char buffer[1024];
// 创建套接字
server_socket = socket(AF_INET, SOCK_STREAM, 0);
if (server_socket == -1) {
perror("Socket creation failed");
exit(EXIT_FAILURE);
}
// 初始化服务器地址
server_addr.sin_family = AF_INET;
server_addr.sin_port = htons(PORT);
server_addr.sin_addr.s_addr = INADDR_ANY;
// 绑定套接字到地址
if (bind(server_socket, (struct sockaddr *)&server_addr, sizeof(server_addr)) == -1) {
perror("Bind failed");
close(server_socket);
exit(EXIT_FAILURE);
}
// 监听连接
if (listen(server_socket, MAX_CLIENTS) == -1) {
perror("Listen failed");
close(server_socket);
exit(EXIT_FAILURE);
}
int epoll_fd = epoll_create(MAX_CLIENTS + 1);
if (epoll_fd == -1) {
perror("Epoll create failed");
close(server_socket);
exit(EXIT_FAILURE);
}
struct epoll_event event;
event.data.fd = server_socket;
event.events = EPOLLIN;
if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, server_socket, &event) == -1) {
perror("Epoll add server socket failed");
close(server_socket);
close(epoll_fd);
exit(EXIT_FAILURE);
}
struct epoll_event events[EVENTS_SIZE];
while (1) {
int num_events = epoll_wait(epoll_fd, events, EVENTS_SIZE, -1);
if (num_events == -1) {
perror("Epoll wait failed");
break;
}
for (int i = 0; i < num_events; i++) {
if (events[i].data.fd == server_socket) {
// 有新连接
client_socket = accept(server_socket, (struct sockaddr *)&client_addr, &client_addr_len);
if (client_socket == -1) {
perror("Accept failed");
continue;
}
printf("New client connected: %s:%d\n", inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port));
event.data.fd = client_socket;
event.events = EPOLLIN;
if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, client_socket, &event) == -1) {
perror("Epoll add client socket failed");
close(client_socket);
}
} else {
// 客户端有数据可读
client_socket = events[i].data.fd;
int bytes_read = recv(client_socket, buffer, sizeof(buffer), 0);
if (bytes_read <= 0) {
// 客户端关闭连接
printf("Client disconnected\n");
close(client_socket);
epoll_ctl(epoll_fd, EPOLL_CTL_DEL, client_socket, NULL);
} else {
buffer[bytes_read] = '\0';
printf("Received from client: %s\n", buffer);
send(client_socket, buffer, strlen(buffer), 0);
}
}
}
}
close(server_socket);
close(epoll_fd);
return 0;
}
二、异步非阻塞网络编程模型
异步非阻塞网络编程模型结合了异步和非阻塞I/O的特性，进一步提升了网络应用程序的性能和并发处理能力。
2.1 非阻塞I/O
在传统的阻塞I/O模式下，当执行I/O操作（如 read 或 write）时，程序会一直阻塞，直到操作完成。这意味着在等待I/O操作的过程中，线程无法执行其他任务，造成资源浪费。
非阻塞I/O则不同，当执行I/O操作时，如果操作不能立即完成，函数会立即返回，并返回一个错误码（如 EAGAIN 或 EWOULDBLOCK），表示操作需要稍后重试。这样，线程可以在等待I/O操作的同时执行其他任务，提高了资源利用率。
要将套接字设置为非阻塞模式，可以使用 fcntl 函数：
int flags = fcntl(sockfd, F_GETFL, 0);
flags |= O_NONBLOCK;
fcntl(sockfd, F_SETFL, flags);
2.2 异步I/O
异步I/O是指应用程序发起I/O操作后，不需要等待操作完成，操作系统会在操作完成后通知应用程序。在Linux中，可以通过 aio_read 和 aio_write 等函数实现异步I/O。
异步I/O与非阻塞I/O的区别在于，非阻塞I/O虽然不会阻塞线程，但仍然需要应用程序主动轮询检查I/O操作是否完成；而异步I/O则由操作系统负责通知应用程序I/O操作的完成，应用程序可以在I/O操作进行的同时执行其他任务。
2.3 结合IO多路复用的异步非阻塞模型
将IO多路复用与异步非阻塞I/O结合，可以构建一个高效的网络编程模型。以epoll为例，在这种模型下，我们可以将套接字设置为非阻塞模式，然后通过epoll监听这些套接字的事件。当某个套接字就绪时，epoll会通知应用程序，应用程序可以在非阻塞的情况下进行I/O操作。
下面是一个基于epoll的异步非阻塞网络编程示例代码：
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <string.h>
#include <sys/epoll.h>
#include <fcntl.h>
#define PORT 8888
#define MAX_CLIENTS 10
#define EVENTS_SIZE 10
void set_nonblocking(int fd) {
int flags = fcntl(fd, F_GETFL, 0);
flags |= O_NONBLOCK;
fcntl(fd, F_SETFL, flags);
}
int main() {
int server_socket, client_socket;
struct sockaddr_in server_addr, client_addr;
socklen_t client_addr_len = sizeof(client_addr);
char buffer[1024];
// 创建套接字
server_socket = socket(AF_INET, SOCK_STREAM, 0);
if (server_socket == -1) {
perror("Socket creation failed");
exit(EXIT_FAILURE);
}
// 初始化服务器地址
server_addr.sin_family = AF_INET;
server_addr.sin_port = htons(PORT);
server_addr.sin_addr.s_addr = INADDR_ANY;
// 绑定套接字到地址
if (bind(server_socket, (struct sockaddr *)&server_addr, sizeof(server_addr)) == -1) {
perror("Bind failed");
close(server_socket);
exit(EXIT_FAILURE);
}
// 监听连接
if (listen(server_socket, MAX_CLIENTS) == -1) {
perror("Listen failed");
close(server_socket);
exit(EXIT_FAILURE);
}
set_nonblocking(server_socket);
int epoll_fd = epoll_create(MAX_CLIENTS + 1);
if (epoll_fd == -1) {
perror("Epoll create failed");
close(server_socket);
exit(EXIT_FAILURE);
}
struct epoll_event event;
event.data.fd = server_socket;
event.events = EPOLLIN;
if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, server_socket, &event) == -1) {
perror("Epoll add server socket failed");
close(server_socket);
close(epoll_fd);
exit(EXIT_FAILURE);
}
struct epoll_event events[EVENTS_SIZE];
while (1) {
int num_events = epoll_wait(epoll_fd, events, EVENTS_SIZE, -1);
if (num_events == -1) {
perror("Epoll wait failed");
break;
}
for (int i = 0; i < num_events; i++) {
if (events[i].data.fd == server_socket) {
// 有新连接
while ((client_socket = accept(server_socket, (struct sockaddr *)&client_addr, &client_addr_len)) != -1) {
set_nonblocking(client_socket);
printf("New client connected: %s:%d\n", inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port));
event.data.fd = client_socket;
event.events = EPOLLIN | EPOLLET;
if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, client_socket, &event) == -1) {
perror("Epoll add client socket failed");
close(client_socket);
}
}
if (errno != EAGAIN && errno != EWOULDBLOCK) {
perror("Accept error");
}
} else {
// 客户端有数据可读
client_socket = events[i].data.fd;
int bytes_read;
while ((bytes_read = recv(client_socket, buffer, sizeof(buffer), 0)) > 0) {
buffer[bytes_read] = '\0';
printf("Received from client: %s\n", buffer);
send(client_socket, buffer, strlen(buffer), 0);
}
if (bytes_read == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
continue;
} else if (bytes_read == -1) {
perror("Recv error");
} else if (bytes_read == 0) {
// 客户端关闭连接
printf("Client disconnected\n");
close(client_socket);
epoll_ctl(epoll_fd, EPOLL_CTL_DEL, client_socket, NULL);
}
}
}
}
close(server_socket);
close(epoll_fd);
return 0;
}
在这个示例中，我们将服务器套接字和客户端套接字都设置为非阻塞模式，并使用epoll的边缘触发（EPOLLET）模式。边缘触发模式下，只有在事件状态发生变化时才会通知应用程序，这可以减少不必要的通知，提高性能。但在边缘触发模式下，应用程序需要一次性处理完所有数据，否则可能会错过后续的数据读取。
三、应用场景和优势
3.1 应用场景
- 高并发网络服务器：如Web服务器、游戏服务器等，需要处理大量的并发连接。基于IO多路复用的异步非阻塞网络编程模型可以在不占用大量系统资源的情况下，高效地处理这些连接，提高服务器的并发处理能力。
- 实时通信系统：如即时通讯软件、实时监控系统等，需要及时响应客户端的请求和数据。这种模型能够快速地处理I/O事件，保证系统的实时性。
- 分布式系统：在分布式系统中，各个节点之间需要进行频繁的网络通信。采用这种模型可以提高节点之间的通信效率，提升整个分布式系统的性能。
3.2 优势
- 资源利用率高：通过IO多路复用，一个线程或进程可以监控多个文件描述符，减少了线程或进程的数量，降低了系统资源的消耗。同时，异步非阻塞I/O使得线程在等待I/O操作时可以执行其他任务，进一步提高了资源利用率。
- 并发处理能力强：能够同时处理大量的并发连接，适合构建高并发的网络应用程序。相比于传统的多线程或多进程模型，这种模型在处理大量连接时性能优势更加明显。
- 响应速度快：异步非阻塞的特性使得应用程序能够及时响应I/O事件，提高了系统的响应速度，适合实时性要求较高的应用场景。
四、挑战与应对
4.1 编程复杂度增加
异步非阻塞网络编程模型虽然性能优越，但编程复杂度也相对较高。在这种模型下，代码逻辑更加复杂，需要处理更多的异步操作和状态管理。例如，在处理边缘触发模式下的I/O事件时，需要确保一次性读取或写入所有数据，否则可能会出现数据丢失或错误。
应对方法：编写清晰的代码结构，使用合适的设计模式来管理异步操作和状态。例如，可以使用状态机来管理套接字的不同状态，使得代码逻辑更加清晰易懂。同时，使用一些成熟的网络编程框架（如libevent、libuv等）也可以降低编程复杂度，这些框架封装了底层的IO多路复用和异步非阻塞操作，提供了更简洁易用的接口。
4.2 调试困难
由于异步非阻塞编程模型的复杂性，调试起来也更加困难。异步操作可能导致程序的执行流程难以预测，错误定位也更加困难。
应对方法：使用日志记录和调试工具来辅助调试。在关键的异步操作点添加详细的日志记录，以便在出现问题时能够追踪程序的执行流程。同时，使用调试工具（如gdb）来调试异步代码，通过设置断点、查看变量值等方式来定位问题。
4.3 数据完整性和一致性
在异步非阻塞模型下，由于I/O操作可能被中断或延迟，数据的完整性和一致性可能会受到影响。例如，在写入数据时，如果网络出现问题，可能导致部分数据丢失。
应对方法：在应用层实现数据校验和重传机制。例如，在发送数据时，添加校验和字段，接收方在接收到数据后进行校验，如果校验失败则请求重传。同时，合理地设置缓冲区和超时时间，确保数据能够完整、一致地传输。
五、总结
基于IO多路复用的异步非阻塞网络编程模型是一种高效的网络编程方式，它通过结合IO多路复用、非阻塞I/O和异步I/O的特性，大大提升了网络应用程序的性能和并发处理能力。在高并发、实时性要求较高的应用场景中，这种模型具有明显的优势。然而，它也带来了编程复杂度增加、调试困难以及数据完整性和一致性等方面的挑战。通过合理的代码设计、使用成熟的框架以及采用有效的调试和数据处理策略，我们可以充分发挥这种模型的优势，构建出高效、稳定的网络应用程序。在实际开发中，需要根据具体的应用场景和需求，权衡利弊，选择合适的网络编程模型。