引言
在多线程编程中，I/O 多路复用（如 select、poll 或 epoll）可以与多线程结合使用，以提高系统的并发处理能力和效率。结合多线程和 I/O 多路复用，可以实现高性能的网络服务器和客户端。以下是一些常见的多线程和 I/O 多路复用结合使用的策略和示例：
常见策略
-
单线程使用 I/O 多路复用：
在一个单独的线程中使用 I/O 多路复用来监视所有的文件描述符。当有文件描述符变得就绪时，在该线程中处理 I/O 操作。这种方法简单且易于实现，但可能会因为单线程的处理能力受限，无法充分利用多核 CPU 的性能。 -
多线程使用 I/O 多路复用：
在主线程中使用 I/O 多路复用来监视所有的文件描述符。当有文件描述符变得就绪时，调用或创建线程来处理 I/O 操作。这种多线程与 I/O 多路复用相结合的方式，可以提高服务器的并发处理能力，充分利用多核 CPU 的性能，是实现高性能网络服务器的一种有效方法。
示例代码
下面是一个简单的例子，展示了如何在多线程环境中结合使用 I/O 多路复用和工作线程池来处理网络连接：
#include <iostream>
#include <vector>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <sys/select.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <cstring>#define PORT 8080
#define MAX_CLIENTS 30std::mutex queue_mutex;
std::condition_variable condition;
std::queue<int> client_queue;void worker_thread() {while (true) {std::unique_lock<std::mutex> lock(queue_mutex);condition.wait(lock, []{ return !client_queue.empty(); });int client_socket = client_queue.front();client_queue.pop();lock.unlock();char buffer[1024] = {0};int valread = read(client_socket, buffer, 1024);if (valread > 0) {buffer[valread] = '\0';//可对字符串进行处理send(client_socket, buffer, valread, 0);}close(client_socket);}
}int main() {int server_fd, new_socket;struct sockaddr_in address;fd_set readfds;int client_sockets[MAX_CLIENTS] = {0};server_fd = socket(AF_INET, SOCK_STREAM, 0);address.sin_family = AF_INET;address.sin_addr.s_addr = INADDR_ANY;address.sin_port = htons(PORT);bind(server_fd, (struct sockaddr*)&address, sizeof(address));listen(server_fd, 3);// 创建工作线程池std::vector<std::thread> workers;for (int i = 0; i < 4; ++i) {workers.emplace_back(worker_thread);}while (true) {FD_ZERO(&readfds);FD_SET(server_fd, &readfds);int max_sd = server_fd;for (int i = 0; i < MAX_CLIENTS; i++) {int sd = client_sockets[i];if (sd > 0)FD_SET(sd, &readfds);if (sd > max_sd)max_sd = sd;}int activity = select(max_sd + 1, &readfds, NULL, NULL, NULL);if (FD_ISSET(server_fd, &readfds)) {new_socket = accept(server_fd, NULL, NULL);for (int i = 0; i < MAX_CLIENTS; i++) {if (client_sockets[i] == 0) {client_sockets[i] = new_socket;break;}}}for (int i = 0; i < MAX_CLIENTS; i++) {int sd = client_sockets[i];if (FD_ISSET(sd, &readfds)) {std::unique_lock<std::mutex> lock(queue_mutex);client_queue.push(sd);lock.unlock();condition.notify_one();client_sockets[i] = 0;}}}for (std::thread &worker : workers) {worker.join();}return 0;
}
代码解释
-
主线程： 主线程负责使用
select监视服务器套接字和已连接的客户端套接字。select函数返回就绪的套接字后，主线程将新的客户端连接或已就绪的客户端套接字加入任务队列，并通知工作线程池处理这些套接字。 -
工作线程池： 多个工作线程从任务队列中获取套接字，并处理 I/O 操作，如读取数据、发送响应和关闭连接。
-
同步机制： 使用
std::mutex和std::condition_variable实现线程间的同步，确保安全地访问和修改任务队列。
