【Linux IO多路复用 】 Linux下select函数全解析：驾驭I-O复用的高效之道

struct timeval {
    long tv_sec;   // seconds
    long tv_usec;  // microseconds
};

#include <stdio.h>
#include <stdlib.h>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>
#include <sys/select.h>
int main(void) {
    fd_set readfds;
    struct timeval timeout;
    int ret, fd_max;
    // 创建两个管道
    int pipefds1[2];
    int pipefds2[2];
    pipe(pipefds1);
    pipe(pipefds2);
    // 向管道写入数据
    write(pipefds1[1], "Hello", 5);
    write(pipefds2[1], "World", 5);
    while (1) {
        FD_ZERO(&readfds);
        FD_SET(pipefds1[0], &readfds);
        FD_SET(pipefds2[0], &readfds);
        // 设置最大文件描述符
        fd_max = (pipefds1[0] > pipefds2[0]) ? pipefds1[0] : pipefds2[0];
        // 设置超时时间
        timeout.tv_sec = 5;
        timeout.tv_usec = 0;
        ret = select(fd_max + 1, &readfds, NULL, NULL, &timeout);
        if (ret == -1) {
            perror("select");
            exit(EXIT_FAILURE);
        } else if (ret == 0) {
            printf("Timeout!\n");
            break;
        } else {
            if (FD_ISSET(pipefds1[0], &readfds)) {
                char buf[6];
                read(pipefds1[0], buf, 5);
                buf[5] = '\0';
                printf("Data from pipe1: %s\n", buf);
            }
            if (FD_ISSET(pipefds2[0], &readfds)) {
                char buf[6];
                read(pipefds2[0], buf, 5);
                buf[5] = '\0';
                printf("Data from pipe2: %s\n", buf);
            }
            break;
        }
    }
    close(pipefds1[0]);
    close(pipefds1[1]);
    close(pipefds2[0]);
    close(pipefds2[1]);
    return 0;
}

#include <stdio.h>
#include <stdlib.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/select.h>
void set_non_blocking(int sockfd) {
    int flags = fcntl(sockfd, F_GETFL, 0);
    fcntl(sockfd, F_SETFL, flags | O_NONBLOCK);
}
int main(void) {
    int listen_sock, conn_sock;
    struct sockaddr_in server_addr, client_addr;
    socklen_t addr_len = sizeof(client_addr);
    fd_set readfds;
    struct timeval timeout;
    int ret, fd_max;
    listen_sock = socket(AF_INET, SOCK_STREAM, 0);
    set_non_blocking(listen_sock);
    server_addr.sin_family = AF_INET;
    server_addr.sin_port = htons(8888);
    server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    bind(listen_sock, (struct sockaddr *)&server_addr, sizeof(server_addr));
    listen(listen_sock, 5);
    while (1) {
        FD_ZERO(&readfds);
        FD_SET(listen_sock, &readfds);
        fd_max = listen_sock;
        // 设置超时时间
        timeout.tv_sec = 5;
        timeout.tv_usec = 0;
        ret = select(fd_max + 1, &readfds, NULL, NULL, &timeout);
        if (ret == -1) {
            perror("select");
            exit(EXIT_FAILURE);
        } else if (ret == 0) {
            printf("Timeout!\n");
            continue;
        } else {
            if (FD_ISSET(listen_sock, &readfds)) {
                conn_sock = accept(listen_sock, (struct sockaddr *)&client_addr, &addr_len);
                if (conn_sock != -1) {
                    printf("Accepted a connection from %s:%d\n", inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port));
                    close(conn_sock);
                }
            }
        }
    }
    close(listen_sock);
    return 0;
}

#include <stdio.h>
#include <stdlib.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/select.h>
typedef struct ready_fd_node {
    int fd;
    struct ready_fd_node *next;
} ready_fd_node;
void set_non_blocking(int sockfd) {
    int flags = fcntl(sockfd, F_GETFL, 0);
    fcntl(sockfd, F_SETFL, flags | O_NONBLOCK);
}
int main(void) {
    int listen_sock, conn_sock;
    struct sockaddr_in server_addr, client_addr;
    socklen_t addr_len = sizeof(client_addr);
    fd_set readfds;
    struct timeval timeout;
    int ret, fd_max;
    ready_fd_node *ready_fds_head = NULL, *current_node = NULL;
    listen_sock = socket(AF_INET, SOCK_STREAM, 0);
    set_non_blocking(listen_sock);
    server_addr.sin_family = AF_INET;
    server_addr.sin_port = htons(8888);
    server_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    bind(listen_sock, (struct sockaddr *)&server_addr, sizeof(server_addr));
    listen(listen_sock, 5);
    while (1) {
        FD_ZERO(&readfds);
        FD_SET(listen_sock, &readfds);
        fd_max = listen_sock;
        // 设置超时时间
        timeout.tv_sec = 5;
        timeout.tv_usec = 0;
        ret = select(fd_max + 1, &readfds, NULL, NULL, &timeout);
        if (ret == -1) {
            perror("select");
            exit(EXIT_FAILURE);
        } else if (ret == 0) {
            printf("Timeout!\n");
            continue;
        } else {
            if (FD_ISSET(listen_sock, &readfds)) {
                conn_sock = accept(listen_sock, (struct sockaddr *)&client_addr, &addr_len);
                if (conn_sock != -1) {
                    printf("Accepted a connection from %s:%d\n", inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port));
                    // 将新连接的文件描述符加入到链表中
                    ready_fd_node *new_node = (ready_fd_node *)malloc(sizeof(ready_fd_node));
                    new_node->fd = conn_sock;
                    new_node->next = NULL;
                    if (ready_fds_head == NULL) {
                        ready_fds_head = new_node;
                    } else {
                        current_node = ready_fds_head;
                        while (current_node->next != NULL) {
                            current_node = current_node->next;
                        }
                        current_node->next = new_node;
                    }
                }
            }
            // 处理链表中的文件描述符
            current_node = ready_fds_head;
            while (current_node != NULL) {
                // 这里可以处理每个已准备好的文件描述符，例如读取数据或者关闭连接
                // ...
                current_node = current_node->next;
            }
        }
    }
    // 释放链表资源
    current_node = ready_fds_head;
    while (current_node != NULL) {
        ready_fd_node *temp = current_node;
        current_node = current_node->next;
        free(temp);
    }
    close(listen_sock);
    return 0;
}

#ifndef SELECT_SERVER_H
#define SELECT_SERVER_H
#include <string>
#include <functional>
#include <vector>
#include <memory>
#include <thread>
#include <map>
#include <mutex>
class Client {
public:
    int sockfd;
    // 其他客户端属性
    Client(int sockfd) : sockfd(sockfd) {}
    ~Client() {}
};
class SelectServer {
public:
    // Constructor
    SelectServer();
    // Destructor
    ~SelectServer();
    // 设置服务器监听端口
    void setPort(unsigned short port);
    // 设置服务器监听地址
    void setAddress(const std::string& address);
    // 设置客户端连接最大数量
    void setMaxClients(unsigned int maxClients);
    // 启动服务器
    void start();
    // 停止服务器
    void stop();
    // 注册读取数据回调函数
    void setOnDataReceivedCallback(std::function<void(int, const std::vector<char>&)> callback);
    // 注册新连接建立回调函数
    void setOnClientConnectedCallback(std::function<void(int)> callback);
    // 注册连接断开回调函数
    void setOnClientDisconnectedCallback(std::function<void(int)> callback);
    // 向指定客户端发送数据
    void sendData(int clientId, const std::vector<char>& data);
    // 关闭指定客户端连接
    void closeClient(int clientId);
    // 获取客户端数量
    unsigned int getClientCount() const;
    // ... 其他函数和成员变量 ...
    std::map<int, std::shared_ptr<Client>> clients;
    std::mutex clientsMutex;    
private:
    // 初始化服务器
    void initServer();
    // 处理客户端连接
    void handleClientConnections();
    // 处理客户端数据
    void handleClientData(int clientId);
    // 处理客户端断开连接
    void handleClientDisconnection(int clientId);
    // 捕获并处理异常
    void handleException(const std::exception& e);
};
#endif // SELECT_SERVER_H

#include "SelectServer.h"
#include "Client.h"
#include <thread>
#include <stdexcept>
#include <iostream>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <fcntl.h>
// Constructor
SelectServer::SelectServer()
    : serverAddress("0.0.0.0"), serverPort(0), maxClients(10), serverSocket(-1) {}
// Destructor
SelectServer::~SelectServer() {
    if (serverSocket != -1) {
        close(serverSocket);
    }
}
// 设置服务器监听端口
// 参数: port - 服务器监听的端口号（范围：0-65535）
// 无返回值
void SelectServer::setPort(unsigned short port) {
    serverPort = port;
}
// 设置服务器监听地址
// 参数: address - 服务器监听的IP地址
// 无返回值
void SelectServer::setAddress(const std::string& address) {
    serverAddress = address;
}
// 设置客户端连接最大数量
// 参数: maxClients - 允许的最大客户端连接数
// 无返回值
void SelectServer::setMaxClients(unsigned int maxClients) {
    this->maxClients = maxClients;
}
// 启动服务器
// 无参数
// 无返回值
void SelectServer::start() {
    initServer();
    handleClientConnections();
}
// 停止服务器
// 无参数
// 无返回值
void SelectServer::stop() {
    if (serverSocket != -1) {
        close(serverSocket);
        serverSocket = -1;
    }
}
// 注册读取数据回调函数
// 参数: callback - 用于处理客户端数据的回调函数，接受两个参数：客户端ID和接收到的数据
// 无返回值
void SelectServer::setOnDataReceivedCallback(std::function<void(int, const std::vector<char>&)> callback) {
    onDataReceivedCallback = callback;
}
// 注册新连接建立回调函数
// 参数: callback - 当有新的客户端连接时，用于处理客户端连接的回调函数，接受一个参数：客户端ID
// 无返回值
void SelectServer::setOnClientConnectedCallback(std::function<void(int)> callback) {
    onClientConnectedCallback = callback;
}
// 注册连接断开回调函数
// 参数: callback - 当客户端断开连接时，用于处理客户端断开连接的回调函数，接受一个参数：客户端ID
// 无返回值
void SelectServer::setOnClientDisconnectedCallback(std::function<void(int)> callback) {
    onClientDisconnectedCallback = callback;
}
// 初始化服务器
// 无参数
// 无返回值
void SelectServer::initServer() {
    // 创建TCP套接字
    serverSocket = socket(AF_INET, SOCK_STREAM, 0);
    if (serverSocket == -1) {
        throw std::runtime_error("Failed to create socket");
    }
   // 设置套接字选项，允许地址和端口复用
    int opt = 1;
    if (setsockopt(serverSocket, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) == -1) {
        throw std::runtime_error("Failed to set socket options");
    }
    // 绑定地址和端口
    struct sockaddr_in addr;
    addr.sin_family = AF_INET;// IPv4地址
    addr.sin_addr.s_addr = inet_addr(serverAddress.c_str()); // 转换服务器IP地址为网络字节序
    addr.sin_port = htons(serverPort); // 转换服务器端口号为网络字节序
    if (bind(serverSocket, (struct sockaddr*)&addr, sizeof(addr)) == -1) {
        throw std::runtime_error("Failed to bind socket");
    }
    // 监听端口，等待客户端连接
    if (listen(serverSocket, 10) == -1) {
        throw std::runtime_error("Failed to listen on socket");
    }
}
// 处理客户端连接
// 无参数
// 无返回值
void SelectServer::handleClientConnections() {
    while (true) {
        struct sockaddr_in clientAddr;
        socklen_t clientAddrLen = sizeof(clientAddr);
        int clientSocket = accept(serverSocket, (struct sockaddr*)&clientAddr, &clientAddrLen);
        if (clientSocket == -1) {
            // 处理异常情况
            continue;
        }
        // 将客户端套接字设置为非阻塞模式
        int flags = fcntl(clientSocket, F_GETFL, 0);
        fcntl(clientSocket, F_SETFL, flags | O_NONBLOCK);
        // 将新客户端添加到客户端列表
        auto client = std::make_shared<Client>(clientSocket);
        {
            std::unique_lock<std::mutex> lock(clientsMutex);
            clients[clientSocket] = client;
        }
        // 创建子线程处理客户端连端连接
        std::thread clientThread(&SelectServer::handleClient, this, clientSocket);
        clientThread.detach();
    }
}
// 处理客户端的子线程函数
// 参数: clientSocket - 客户端的文件描述符
// 无返回值
void SelectServer::handleClient(int clientSocket) {
    // 在这里处理客户端的各种请求，例如读写数据等
    try {
        // 以下为示例，您可以根据实际需求编写逻辑
        char buffer[1024];
        ssize_t bytesRead;
        while ((bytesRead = recv(clientSocket, buffer, sizeof(buffer), 0)) > 0) {
            std::vector<char> data(buffer, buffer + bytesRead);
            // 调用注册的回调函数处理数据
            if (onDataReceivedCallback) {
                onDataReceivedCallback(clientSocket, data);
            }
        }
        // 客户端断开连接
        closeClient(clientSocket);
        {
            std::unique_lock<std::mutex> lock(clientsMutex);
            clients.erase(clientSocket);
        }
        // 调用客户端断开连接的回调函数
        if (onClientDisconnectedCallback) {
            onClientDisconnectedCallback(clientSocket);
        }
    } catch (const std::exception& e) {
        // 捕获并处理异常
        handleException(e);
    }
}
// 处理客户端数据
// 参数: clientId - 要处理数据的客户端ID
// 无返回值
void SelectServer::handleClientData(int clientId) {
    // 查找客户端对象
    std::shared_ptr<Client> client;
    {
        std::unique_lock<std::mutex> lock(clientsMutex);
        auto it = clients.find(clientId);
        if (it == clients.end()) {
            // 客户端不存在，返回
            return;
        }
        client = it->second;
    }
    // 读取客户端数据
    char buffer[1024];
    ssize_t bytesRead = recv(client->sockfd, buffer, sizeof(buffer), 0);
    if (bytesRead > 0) {
        // 将数据封装为 vector 对象
        std::vector<char> data(buffer, buffer + bytesRead);
        // 调用数据接收回调函数处理数据
        if (onDataReceivedCallback) {
            onDataReceivedCallback(clientId, data);
        }
    } else if (bytesRead == 0) {
        // 客户端断开连接
        handleClientDisconnection(clientId);
    } else {
        // 处理异常情况
    }
}
// 处理客户端断开连接
// 参数: clientId - 要处理断开连接的客户端ID
// 无返回值
void SelectServer::handleClientDisconnection(int clientId) {
    // 关闭客户端连接
    closeClient(clientId);
    // 移除客户端
    {
        std::unique_lock<std::mutex> lock(clientsMutex);
        clients.erase(clientId);
    }
    // 调用客户端断开连接回调函数
    if (onClientDisconnectedCallback) {
        onClientDisconnectedCallback(clientId);
    }
}
// 向指定客户端发送数据
// 参数: clientId - 要发送数据的客户端ID；data - 要发送的数据
// 无返回值
void SelectServer::sendData(int clientId, const std::vector<char>& data) {
    std::shared_ptr<Client> client;
    {
        std::unique_lock<std::mutex> lock(clientsMutex);
        auto it = clients.find(clientId);
        if (it == clients.end()) {
            // 客户端不存在，返回
            return;
        }
        client = it->second;
    }
    try {
        // 发送数据
        if (send(client->sockfd, data.data(), data.size(), 0) == -1) {
            // 发送失败，处理异常情况
            throw std::runtime_error("Send data failed.");
        }
    } catch (const std::exception& e) {
        // 捕获并处理异常
        handleException(e);
    }
}
// 关闭指定客户端连接
// 参数: clientId - 要关闭的客户端ID
// 无返回值
void SelectServer::closeClient(int clientId) {
    std::shared_ptr<Client> client;
    {
        std::unique_lock<std::mutex> lock(clientsMutex);
        auto it = clients.find(clientId);
        if (it == clients.end()) {
            // 客户端不存在，返回
            return;
        }
        client = it->second;
    }
    try {
        // 关闭客户端连接
        if (close(client->sockfd) == -1) {
            // 关闭失败，处理异常情况
            throw std::runtime_error("Close client failed.");
        }
    } catch (const std::exception& e) {
        // 捕获并处理异常
        handleException(e);
    }
}
// 获取客户端数量
// 无参数
// 返回值: 当前已连接的客户端数量
unsigned int SelectServer::getClientCount() const {
    std::unique_lock<std::mutex> lock(clientsMutex);
    return clients.size();
}
// 捕获并处理异常
// 参数: e - 异常引用
// 无返回值
void SelectServer::handleException(const std::exception& e) {
    // 输出异常信息
    std::cerr << "Exception caught: " << e.what() << std::endl;
    // 在这里实现其他异常处理逻辑，例如采取恢复措施或通知其他组件
    // ...
}