前言
围绕Socket 的基础概念、I/O 模型，逐步实现阻塞 I/O 客户端 - 服务器、多进程 / 多线程服务端处理，以及基于 select、poll、epoll 的 I/O 多路复用服务端，侧重与如何实现。
什么是socket
在C++中，Socket编程是一种用于在网络上进行通信的技术，它允许不同主机上的应用程序通过TCP/IP协议族进行数据交换。Socket作为应用层与TCP/IP协议族通信的中间软件抽象层，提供了一组接口，隐藏了协议的复杂性，使得开发者只需要关注简单的接口即可实现网络通信。
I/O
I/O (Input/Output) - 输入输出：泛指计算机与外部世界的数据交换，socket是网络I/O；
进行I/O时分为两步：一是检测是否符合条件，二是进行操作。阻塞条件下不满足条件就会一直检测直到符合条件；非阻塞检测一次就会返回。
I/O多路复用就是只检测，找到符合条件的文件描述符，然后只处理这些文件描述符。现在常见的I/O多路复用有三种select、poll、epoll，在检测时select、poll是通过轮询进行检测，epoll是通过回调进行实现。还有就是select需要每次把关注的fd集合拷贝到内核，而epoll只需要第一次把fd拷贝到红黑树中；
阻塞 I/O（Blocking I/O）：满足发送/接收条件才输入输出
- 当进程执行I/O操作时，如果条件不满足，进程会被操作系统挂起（睡眠），直到条件满足才被唤醒继续执行。
应用程序 内核 网络 | | | | send() | | |--------------->| | | | wait buffer space | | |<----------------> | | | copy data | | | start send | | |------------------>| | return size | | |<---------------| | 应用程序 内核 网络| | || recv() | ||--------------->| || |check recv buffer|| | if null->wait || |<--------------->|| | else recv || |<----------------|| | copy data to || return size | user space ||<---------------| | 非阻塞 I/O（Non-blocking I/O）：
- 当进程执行I/O操作时，如果条件不满足，系统调用立即返回一个错误，而不是让进程进入睡眠状态。进程可以继续执行其他任务，稍后再重试。
- 非阻塞 I/O 的关键特征
- 立即返回：无论I/O是否完成，立即返回控制权
- 错误码指示：用特殊错误码表示"需要等待"
- 主动轮询：需要程序主动检查I/O状态
- 并行处理：可以在等待I/O时做其他事情
| 模型 | 特点 | 适用场景 |
|---|---|---|
| 阻塞 I/O | 条件不满足时进程挂起 | 简单应用，连接数少 |
| 非阻塞 I/O | 立即返回，需要轮询 | 需要实时响应的应用 |
| I/O 多路复用 | 一个线程管理多个 I/O | 高并发服务器 |
TCP-UDP
| 特性 | TCP | UDP |
|---|---|---|
| 连接 | 面向连接 | 无连接 |
| 可靠性 | 可靠，自动重传 | 不可靠，可能丢包 |
| 顺序性 | 保证数据顺序 | 不保证顺序 |
| 流量控制 | 有滑动窗口 | 无 |
| 头部开销 | 20-60字节 | 8字节 |
连接：tcp三次握手，四次挥手
- SYN (Synchronize)：同步序号，用于建立连接
- ACK (Acknowledgment)：确认标志
- FIN (Finish)：结束标志，用于关闭连接
- RST (Reset)：重置连接
- PSH (Push)：推送数据
- URG (Urgent)：紧急指针有效
连接-三次握手
Client Server| || 1. SYN (seq=x) ||--------------------------------------->|| || 2. SYN-ACK (seq=y, ack=x+1) ||<---------------------------------------|| || 3. ACK (ack=y+1) ||--------------------------------------->|| || 连接建立，开始数据传输 ||<======================================>|断开连接-四次挥手Client Server| || 1. FIN (seq=u) ||--------------------------------------->|| || 2. ACK (ack=u+1) ||<---------------------------------------|| || 服务器处理剩余数据 ||<======================================>|| || 3. FIN (seq=v, ack=u+1) ||<---------------------------------------|| || 4. ACK (ack=v+1) ||--------------------------------------->|| 双方关闭连接 |
send-recv
tcp使用send-recv收发消息；
- 发送消息- send
ssize_t send(int sockfd, // 目标socket文件描述符const void *buf, // 要发送的数据缓冲区size_t len, // 要发送的数据长度int flags); // 发送标志（通常为0）// 0 - 默认行为，阻塞发送
send(sockfd, buf, len, 0);// MSG_DONTWAIT - 非阻塞发送
send(sockfd, buf, len, MSG_DONTWAIT);// MSG_NOSIGNAL - 不生成SIGPIPE信号
send(sockfd, buf, len, MSG_NOSIGNAL);// MSG_OOB - 发送带外数据（紧急数据）
send(sockfd, buf, len, MSG_OOB);// MSG_MORE - 提示有更多数据要发送（用于UDP）
send(sockfd, buf, len, MSG_MORE);// 可以组合使用
send(sockfd, buf, len, MSG_DONTWAIT | MSG_NOSIGNAL);
- 接收消息- recv
ssize_t recv(int sockfd, // socket文件描述符void *buf, // 接收数据缓冲区size_t len, // 缓冲区长度int flags); // 接收标志（通常为0）// 常用的flags值：
// 0 - 默认行为，阻塞接收
recv(sockfd, buf, len, 0);// MSG_DONTWAIT - 非阻塞接收
recv(sockfd, buf, len, MSG_DONTWAIT);// MSG_PEEK - 查看数据但不从缓冲区移除
recv(sockfd, buf, len, MSG_PEEK);// MSG_OOB - 接收带外数据（紧急数据）
recv(sockfd, buf, len, MSG_OOB);// MSG_WAITALL - 等待接收所有请求的字节
recv(sockfd, buf, len, MSG_WAITALL);// MSG_TRUNC - 即使数据被截断也返回数据包长度（原始长度）
recv(sockfd, buf, len, MSG_TRUNC);// 可以组合使用（某些组合可能无效）
recv(sockfd, buf, len, MSG_DONTWAIT | MSG_PEEK);
TCP实例
简单实例
实现一个简单的单线程、单个连接的阻塞I/O的客户端-服务器程序
server
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>#define PORT 8080
#define BUFFER_SIZE 1024int main() {int server_fd, client_fd;struct sockaddr_in address, client_addr;socklen_t client_len = sizeof(client_addr);char buffer[BUFFER_SIZE];// 创建socketserver_fd = socket(AF_INET, SOCK_STREAM, 0);if (server_fd == 0) {perror("socket failed");return 1;}int opt = 1;if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {perror("setsockopt failed");close(server_fd);return 1;}// 设置地址address.sin_family = AF_INET;address.sin_addr.s_addr = INADDR_ANY;address.sin_port = htons(PORT);// 绑定if (bind(server_fd, (struct sockaddr*)&address, sizeof(address)) < 0) {perror("bind failed");return 1;}// 监听if (listen(server_fd, 3) < 0) {perror("listen");return 1;}printf("Server started on port %d\n", PORT);// 接受连接client_fd = accept(server_fd, (struct sockaddr*)&client_addr, &client_len);if (client_fd < 0) {perror("accept");return 1;}printf("Client connected\n");while (1) {// 读取数据int bytes_read = read(client_fd, buffer, BUFFER_SIZE - 1);if (bytes_read > 0) {buffer[bytes_read] = '\0';printf("Received: %s\n", buffer);// 发送响应const char* response = "Hello from server!\n";write(client_fd, response, strlen(response));}else if (bytes_read == 0) {printf("Client disconnected\n");break;} else {perror("Read failed");break;}}// 关闭连接close(client_fd);return 0;
}
- CMakeLists.txt
cmake_minimum_required(VERSION 3.0)project(socketServer)# include_directories(${CMAKE_SOURCE_DIR}/CMAKE_SOURCE_DIRinclude)
include_directories(include)add_compile_options(-g -std=c++11 -o2 -Wall)set(CMAKE_BUILD_TYPE Debug)
# 设置可执行文件输出目录（在 build/bin/）
set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${CMAKE_SOURCE_DIR}/lib)# 设置库文件输出目录（在 build/lib/）
set(CMAKE_LIBRARY_OUTPUT_DIRECTORY ${CMAKE_SOURCE_DIR}/lib)add_executable(server ./src/server.cpp)target_link_libraries(server PRIVATE pthread)client
#include <iostream>#include <string>#include <cstring>#include <cerrno>#include <unistd.h>#include <sys/socket.h>#include <netinet/in.h>#include <arpa/inet.h>#define SERVER_IP "127.0.0.1"#define PORT 8080#define BUFFER_SIZE 1024int main() {int sock = 0;struct sockaddr_in serv_addr;char buffer[BUFFER_SIZE] = {0};// 创建socketif ((sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {std::cerr << "Socket creation error: " << strerror(errno) << std::endl;return EXIT_FAILURE;}serv_addr.sin_family = AF_INET;serv_addr.sin_port = htons(PORT);// 转换IP地址if (inet_pton(AF_INET, SERVER_IP, &serv_addr.sin_addr) <= 0) {std::cerr << "Invalid address/Address not supported: " << strerror(errno) << std::endl;return EXIT_FAILURE;}// 连接服务器if (connect(sock, (struct sockaddr *)&serv_addr, sizeof(serv_addr)) < 0) {std::cerr << "Connection failed: " << strerror(errno) << std::endl;return EXIT_FAILURE;}std::cout << "Connected to server at " << SERVER_IP << ":" << PORT << std::endl;std::cout << "Type your messages (type 'exit' to quit):" << std::endl;while (true) {// 获取用户输入std::string input;//不使用do-while 如果直接按enter，程序会向下走，send发送为空，但是服务端接收不到，就会在read中等待//客户端也会在read中等待do{std::cout << "> ";std::getline(std::cin, input);}while(input.empty());// 检查退出命令if (input == "exit") {break;}// 发送消息到服务器if (send(sock, input.c_str(), input.length(), 0) < 0) {std::cerr << "Send failed: " << strerror(errno) << std::endl;break;}std::cout << "Message sent to server" << std::endl;// 接收服务器响应memset(buffer, 0, BUFFER_SIZE);int valread = read(sock, buffer, BUFFER_SIZE - 1);if (valread < 0) {std::cerr << "Read error: " << strerror(errno) << std::endl;break;} else if (valread == 0) {std::cout << "Server closed the connection" << std::endl;break;} else {buffer[valread] = '\0';std::cout << "Server response:\n" << buffer << std::endl;}}close(sock);std::cout << "Connection closed" << std::endl;return 1;}非阻塞方式
// 获取当前文件状态标志int flags = fcntl(fd, F_GETFL, 0);if (flags < 0) {perror("fcntl F_GETFL");return -1;}// 添加非阻塞标志if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) < 0) {perror("fcntl F_SETFL");return -1;}
int on = 1; // 1表示启用非阻塞if (ioctl(fd, FIONBIO, &on) < 0) {perror("ioctl FIONBIO");return -1;}
服务端接收多个客户端
- 多进程、多线程、select、poll、epoll
- 少量连接：多进程/多线程
- 中等并发：select/poll
- 高并发：epoll
多进程 fork()
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <signal.h>#define PORT 8080
#define BUFFER_SIZE 1024int main() {int server_fd, client_fd;struct sockaddr_in address, client_addr;socklen_t client_len = sizeof(client_addr);char buffer[BUFFER_SIZE];// 创建socketserver_fd = socket(AF_INET, SOCK_STREAM, 0);if (server_fd == 0) {perror("socket failed");return 1;}int opt = 1;if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {perror("setsockopt failed");close(server_fd);return 1;}// 设置地址address.sin_family = AF_INET;address.sin_addr.s_addr = INADDR_ANY;address.sin_port = htons(PORT);// 绑定if (bind(server_fd, (struct sockaddr*)&address, sizeof(address)) < 0) {perror("bind failed");return 1;}// 监听if (listen(server_fd, 3) < 0) {perror("listen");return 1;}signal(SIGCHLD, SIG_IGN);printf("Server started on port %d\n", PORT);// 只需要修改主循环部分while (true) {// 接受连接client_fd = accept(server_fd, (struct sockaddr*)&client_addr, &client_len);if (client_fd < 0) {perror("accept faild!");continue; // 继续等待下一个连接}printf("Client connected\n");// 创建子进程处理客户端pid_t pid = fork();if (pid < 0) {perror("fork failed");close(client_fd);continue;}if (pid == 0) { // 子进程close(server_fd); // 子进程不需要监听socket// 处理客户端连接（使用你现有的while循环）while (true) {int bytes_read = read(client_fd, buffer, BUFFER_SIZE - 1);if (bytes_read > 0) {buffer[bytes_read] = '\0';printf("PID %d Received: %s\n", getpid(), buffer);const char* response = "server received!\n";write(client_fd, response, strlen(response));}else if (bytes_read == 0) {printf("PID %d Client disconnected\n", getpid());break;}else {perror("Read failed");break;}}close(client_fd);exit(0); // 子进程退出}else { // 父进程close(client_fd); // 父进程关闭客户端socket，继续监听}}// 关闭连接close(client_fd);return 0;
}
多线程 std::thread
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <thread>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <signal.h>
#include <iostream>
#define PORT 8080
#define BUFFER_SIZE 1024// 客户端处理线程函数
void handle_client(int client_fd) {char buffer[BUFFER_SIZE];std::cout << "Thread " << std::this_thread::get_id() << ": Client connected" << std::endl;while (1) {memset(buffer, 0, BUFFER_SIZE);int bytes_read = read(client_fd, buffer, BUFFER_SIZE - 1);if (bytes_read > 0) {buffer[bytes_read] = '\0';const char * response = "receive over!";std::cout<< "received:" << buffer << std::endl;write(client_fd, response, strlen(response));}else if (bytes_read == 0) {printf("Thread %lu: Client disconnected\n", pthread_self());break;}else {perror("Read failed");break;}}close(client_fd);return;
}int main() {int server_fd, client_fd;struct sockaddr_in address, client_addr;socklen_t client_len = sizeof(client_addr);// 创建socketserver_fd = socket(AF_INET, SOCK_STREAM, 0);if (server_fd == 0) {perror("socket failed");return 1;}int opt = 1;if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {perror("setsockopt failed");close(server_fd);return 1;}// 设置地址address.sin_family = AF_INET;address.sin_addr.s_addr = INADDR_ANY;address.sin_port = htons(PORT);// 绑定if (bind(server_fd, (struct sockaddr*)&address, sizeof(address)) < 0) {perror("bind failed");return 1;}// 监听if (listen(server_fd, 3) < 0) {perror("listen");return 1;}signal(SIGCHLD, SIG_IGN);printf("Server started on port %d\n", PORT);while (true) {// 接受连接client_fd = accept(server_fd, (struct sockaddr*)&client_addr, &client_len);if (client_fd < 0) {perror("accept faild!");continue; // 继续等待下一个连接}char client_ip[INET_ADDRSTRLEN];inet_ntop(AF_INET, &client_addr.sin_addr, client_ip, sizeof(client_ip));int client_port = ntohs(client_addr.sin_port);std::cout << "New client connected from " << client_ip << ":" << client_port << std::endl;try {std::thread client_thread(handle_client, client_fd);client_thread.detach(); // 分离线程，让它在后台运行 }catch (const std::exception& e) {std::cerr << "Failed to create thread: " << e.what() << std::endl;close(client_fd);}}close(client_fd);std::cout << "Server stopped" << std::endl;return 0;
}
select多路复用
使用select 实现，并做简单的封装。
虽然使用多线程和多进程都可以实现对多个客户端监听，每一个进程/线程都会在send/recv处阻塞，等待发送/接收；
select、poll、epoll可以对文件描述符进行监控，当有事件触发时才会去调用。
select实现I/O多路复用就是对fd_set（位图）的操作，如果要监控可读操作，就声明一个read的位图，把已经连接到服务器的clientfd记录下来（不如clientfd=10，就是把位图中第十个值置为1）；通过select对位图进行监控，当有满足read的文件描述符时值不变，不满足的清空，所以要把连接的客户端存起来避免丢失。所以返回的fd_set就是满足可读的已连接客户端，对这些客户端进行操作就行了。
这里面存在的问题就是系统默认fd_set大小为1024，所以最多连接1020个客户端（0-标准输入流、1-标准输出流、2-标准错误流、3-服务端文件描述符），当客户端连接过多时会丢失或者出现错误。
还有就是fd_set会从用户态拷贝到内核，再拷贝出来，性能上不是很好。
select 是一种 I/O 多路复用技术，允许程序同时监视多个文件描述符（file descriptors），等待一个或多个描述符变为"就绪"状态（可读、可写或发生异常）
I/O多路复用就是使用一个线程管理多个io是否就绪。
-
先了解一下几个概念
-
fd_set是什么
fd_set是一个存放文件描述符的数组，大小为1024位； 结构体实际上就是定义一个 长整型的数组 long int fds_bits[FD_SETSIZE / NFDBITS];
1024 = int * 数组个数 * 8 位，能够储存1024个文件描述符；也就是实际能够储存1024个socket的文件描述符。
每个进程默认打开3个文件描述符，0-标准输入流、1-标准输出流、2-标准错误流
/* The fd_set member is required to be an array of longs. */
typedef long int __fd_mask;/* Some versions of <linux/posix_types.h> define this macros. */
#undef __NFDBITS
/* It's easier to assume 8-bit bytes than to get CHAR_BIT. */
#define __NFDBITS (8 * (int) sizeof (__fd_mask))
#define __FD_ELT(d) ((d) / __NFDBITS)
#define __FD_MASK(d) ((__fd_mask) (1UL << ((d) % __NFDBITS)))/* fd_set for select and pselect. */
typedef struct{/* XPG4.2 requires this member name. Otherwise avoid the namefrom the global namespace. */
#ifdef __USE_XOPEN__fd_mask fds_bits[__FD_SETSIZE / __NFDBITS];
# define __FDS_BITS(set) ((set)->fds_bits)
#else__fd_mask __fds_bits[__FD_SETSIZE / __NFDBITS];
# define __FDS_BITS(set) ((set)->__fds_bits)
#endif} fd_set;
- 文件描述符的分配规则
寻找最小的未使用的文件描述符，所以连接数较少时可以使用select进行实现。
- 文件描述符如何映射到socket
好像是：进程控制块-进程描述表->文件描述符表->文件对象；具体怎么映射的没了解
int select(int maxfdp, //最大文件描述符值加1fd_set *readfds, //指向可读文件描述符集合的指针 可读时接收fd_set *writefds, //指向可写文件描述符集合的指针 可写时发送fd_set *exceptfds, //指向异常文件描述符集合的指针 常见：带外数据到达（TCP紧急数据）struct timeval *timeout); //超时时间结构体指针
- 实现
#include <iostream>#include <sys/select.h>#include <sys/socket.h>#include <netinet/in.h>#include <arpa/inet.h>#include <unistd.h>#include <cstring>#include <vector>#include <algorithm>#define PORT 8080#define BUFFER_SIZE 1024#define MAX_CLIENTS 10class TCPServer {private:int server_socket;fd_set readfds;int client_socket[MAX_CLIENTS];int max_sd;public:TCPServer() {server_socket = -1;// 初始化客户端socket数组for (int i = 0; i < MAX_CLIENTS; i++) {client_socket[i] = 0;}}void init() {int opt = 1;struct sockaddr_in address;// 创建主socketif ((server_socket = socket(AF_INET, SOCK_STREAM, 0)) == 0) {perror("socket failed");exit(EXIT_FAILURE);}// 设置socket选项，允许地址重用if (setsockopt(server_socket, SOL_perror("setsockopt");exit(EXIT_FAILURE);}// 配置地址address.sin_family = AF_INET;address.sin_addr.s_addr = INADDR_ANY;address.sin_port = htons(PORT);// 绑定socketif (bind(server_socket, (struct sockaddr *)&address, sizeof(address)) < 0) {perror("bind failed");exit(EXIT_FAILURE);}std::cout << "Listener on port " << PORT << std::endl;// 开始监听if (listen(server_socket, MAX_CLIENTS) < 0) {perror("listen");exit(EXIT_FAILURE);}}void run() {int activity, new_socket, sd;int max_sd;struct sockaddr_in address;socklen_t addrlen = sizeof(address);char buffer[BUFFER_SIZE];std::cout << "Waiting for connections..." << std::endl;while (true) {// 清空文件描述符集合FD_ZERO(&readfds);// 添加主socket到集合FD_SET(server_socket, &readfds);max_sd = server_socket;// 添加客户端socket到集合for (int i = 0; i < MAX_CLIENTS; ++i) {sd = client_socket[i];if (sd > 0) {FD_SET(sd, &readfds);}max_sd = max_sd > sd ? max_sd : sd;}// 等待活动struct timeval timeout;timeout.tv_sec = 0; // 秒timeout.tv_usec = 500;activity = select(max_sd + 1, &readfds, NULL, NULL, &timeout);if (activity < 0 && errno != EINTR) {std::cerr << "select error" << std::endl;continue;}else if (activity == 0) {// 超时处理continue;}// 如果有新连接if (FD_ISSET(server_socket, &readfds)) {if ((new_socket = accept(server_socket,(struct sockaddr *)&address,&addrlen)) < 0) {perror("accept");exit(EXIT_FAILURE);}std::cout << "New connection, socket fd: " << new_socket<< ", IP: " << inet_ntoa(address.sin_addr)<< ", Port: " << ntohs(address.sin_port) << std::endl;// 添加到客户端数组for (int i = 0; i < MAX_CLIENTS; i++) {if (client_socket[i] == 0) {client_socket[i] = new_socket;std::cout << "Adding to list of sockets as " << i <<" client id: " << new_sock et << std::endl;break;}}}// 检查客户端socket的IO操作for (int i = 0; i < MAX_CLIENTS; i++) {sd = client_socket[i];memset(buffer,0,BUFFER_SIZE);if (FD_ISSET(sd, &readfds)) {// 检查是否断开连接int valread = read(sd, buffer, BUFFER_SIZE);if (valread == 0) {// 客户端断开连接getpeername(sd, (struct sockaddr*)&address, &addrlen);std::cout << "Host disconnected, IP: " << inet_ntoa(address.sin_addr)<< ", Port: " << ntohs(address.sin_port) << std::endl;close(sd);client_socket[i] = 0;} else {// 处理接收到的数据buffer[valread] = '\0';std::cout << "Received: " << buffer << std::endl;// 回声给客户端send(sd, buffer, strlen(buffer), 0);}}}}}~TCPServer() {close(server_socket);for (int i = 0; i < MAX_CLIENTS; ++i) {if (client_socket[i] > 0) {std::cout << "close client :"<< client_socket[i] << std::endl;close(client_socket[i]);}}}};int main() {TCPServer server;server.init();server.run();return 0;}poll多路复用
poll不再使用位图对文件描述符存储，而是通过pollfd进行控制，可以声明pollfd的数组（所以就没有了1024的限制）。
- poll改进
没有最大文件描述符限制：poll 使用数组，可以处理任意数量的文件描述符
更高效：不需要每次调用都重新设置文件描述符集合
更清晰的事件分离：每个文件描述符都有独立的事件输入和输出字段
- poll事件常量
| 事件常量 | 值（通常） | 说明 | 对象 |
|---|---|---|---|
POLLIN |
0x001 | 数据可读（普通数据） | events/revents |
POLLPRI |
0x002 | 紧急数据可读（带外数据） | events/revents |
POLLOUT |
0x004 | 数据可写，不阻塞 | events/revents |
POLLRDNORM |
0x040 | 普通数据可读 | events/revents |
POLLRDBAND |
0x080 | 优先级带数据可读 | events/revents |
POLLWRNORM |
0x100 | 普通数据可写 | events/revents |
POLLWRBAND |
0x200 | 优先级带数据可写 | events/revents |
POLLERR |
0x008 | 错误情况 | revents |
POLLHUP |
0x010 | 已挂起 | revents |
POLLNVAL |
0x020 | 无效轮询请求 | revents |
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <iostream>
#include <errno.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <poll.h>#define PORT 8080
#define MAX_CLIENTS 1000
#define BUFFER_SIZE 1024
#define TIMEOUT -1 // 无限等待class TCPServer{private:int server_fd;char buffer[BUFFER_SIZE];struct pollfd fds[MAX_CLIENTS + 1]; // +1 给服务器套接字int nfds;public:TCPServer(){server_fd = -1;nfds = 0;// 初始化所有 pollfd 结构for (int i = 0; i < MAX_CLIENTS + 1; i++) {fds[i].fd = -1; // 表示未使用fds[i].events = 0;fds[i].revents = 0;}};void init(){// 创建监听套接字if ((server_fd = socket(AF_INET, SOCK_STREAM, 0)) == 0) {perror("socket failed");exit(EXIT_FAILURE);}// 设置 SO_REUSEADDR 选项int opt = 1;if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt))) {perror("setsockopt failed");exit(EXIT_FAILURE);}struct sockaddr_in address;address.sin_family = AF_INET;address.sin_addr.s_addr = INADDR_ANY;address.sin_port = htons(PORT);// 绑定地址if (bind(server_fd, (struct sockaddr *)&address, sizeof(address)) < 0) {perror("bind failed");exit(EXIT_FAILURE);}// 开始监听 listen(server_fd, 10) 等待连接的最大个数if (listen(server_fd, 10) < 0) {perror("listen failed");exit(EXIT_FAILURE);}printf("Server listening on port %d\n", PORT);// 添加服务器套接字到 pollfd 数组fds[0].fd = server_fd;fds[0].events = POLLIN; // 监视可读事件（新连接）nfds = 1; // 当前监视的文件描述符数量};void run(){int current_size = 0;while (1) {// 调用 poll，等待事件发生int ret = poll(fds, nfds, TIMEOUT);if (ret < 0) {perror("poll failed");break;} else if (ret == 0) {// 超时（本例中不会发生，因为TIMEOUT=-1）continue;}current_size = nfds; // 保存当前大小，因为 nfds 可能在循环中改变// 检查所有文件描述符for (int i = 0; i < current_size; i++) {if (fds[i].fd < 0) continue; // 跳过未使用的文件描述符// 检查是否有事件发生if (fds[i].revents == 0) continue;// 检查是否是错误事件if (fds[i].revents & (POLLERR | POLLHUP | POLLNVAL)) {printf("Error on fd %d, closing connection\n", fds[i].fd);close(fds[i].fd);fds[i].fd = -1;fds[i].revents = 0;continue;}// 如果是服务器套接字，处理新连接if (fds[i].fd == server_fd) {if (fds[i].revents & POLLIN)handle_client_connect();}// 处理客户端套接字else {handle_client_event(i);}}}};void handle_client_connect(){// 接受新连接struct sockaddr_in address;socklen_t addrlen = sizeof(address);int new_socket = accept(server_fd, (struct sockaddr *)&address, &addrlen);if (new_socket < 0) {perror("accept failed");return ;}printf("New connection, socket fd is %d, IP: %s, port: %d\n",new_socket,inet_ntoa(address.sin_addr), ntohs(address.sin_port));// 查找空位添加新客户端int added = 0 ;for (int j = 1; j < MAX_CLIENTS + 1; ++j) {if (fds[j].fd == -1) {fds[j].fd = new_socket;fds[j].events = POLLIN | POLLRDHUP; // 监视可读和连接关闭fds[j].revents = 0;added = 1;nfds = j >= nfds ? j+1 : nfds;break;}}if (!added) {printf("Too many clients, rejecting connection\n");close(new_socket);}};void handle_client_event(int fd_index){// 检查连接是否关闭if (fds[fd_index].revents & POLLRDHUP) {printf("Client %d disconnected\n", fds[fd_index].fd);close(fds[fd_index].fd);fds[fd_index].fd = -1;fds[fd_index].revents = 0;return;}// 检查是否有数据可读if (fds[fd_index].revents & POLLIN) {memset(buffer,0,BUFFER_SIZE);int valread = read(fds[fd_index].fd, buffer, BUFFER_SIZE);if (valread == 0) {// 客户端正常关闭连接printf("Client %d closed connection\n", fds[fd_index].fd);close(fds[fd_index].fd);fds[fd_index].fd = -1;} else if (valread < 0) {// 读取错误perror("read failed");close(fds[fd_index].fd);fds[fd_index].fd = -1;} else {// 回显数据buffer[valread] = '\0';std::cout << "Received from client : " << fds[fd_index].fd <<" " << buffer << std: :endl;send(fds[fd_index].fd,buffer,strlen(buffer),0);}}}~TCPServer(){// 关闭所有连接for (int i = 0; i < nfds; i++) {if (fds[i].fd >= 0) {close(fds[i].fd);}}if (server_fd >= 0) {close(server_fd);}printf("Server shutdown complete\n");};
};int main() {TCPServer server;server.init();server.run();return 0;
}
- 事件
#define POLLIN 0x001 /* There is data to read. */
#define POLLPRI 0x002 /* There is urgent data to read. */
#define POLLOUT 0x004 /* Writing now will not block. */#if defined __USE_XOPEN || defined __USE_XOPEN2K8
/* These values are defined in XPG4.2. */
# define POLLRDNORM 0x040 /* Normal data may be read. */
# define POLLRDBAND 0x080 /* Priority data may be read. */
# define POLLWRNORM 0x100 /* Writing now will not block. */
# define POLLWRBAND 0x200 /* Priority data may be written. */
#endif#ifdef __USE_GNU
/* These are extensions for Linux. */
# define POLLMSG 0x400
# define POLLREMOVE 0x1000
# define POLLRDHUP 0x2000
#endif/* Event types always implicitly polled for. These bits need not be set in`events', but they will appear in `revents' to indicate the status ofthe file descriptor. */
#define POLLERR 0x008 /* Error condition. */
#define POLLHUP 0x010 /* Hung up. */
#define POLLNVAL 0x020 /* Invalid polling request. */
select/poll 缺点
- 每次调用时要重复地从用户态读入参数。
- 每次调用时要重复地扫描文件描述符。
- 每次在调用开始时，要把当前进程放入各个文件描述符的等待队列。在调用结束后，又把进程从各个等待队列中删除。
poll核心时间可分为可读、可写、异常事件。
epoll()多路复用
-
水平触发 ：fd事件没有被处理或者没有处理全部，下次还会报告这个fd
-
边缘触发：程序在处理文件描述符的就绪事件时，必须确保将其处理完毕，否则 epoll_wait 将不会重复通知该文件描述符的就绪状态。
-
基于红黑树+链表+回调函数
用户空间 内核空间│ ││ epoll_create() │├───────────────────────────────> │ 创建eventpoll结构│ │ • 初始化红黑树(rbr)│ │ • 初始化就绪链表(rdllist)│ ││ epoll_ctl(EPOLL_CTL_ADD, fd) │├───────────────────────────────> │ • 分配epitem结构│ │ • 插入红黑树(O(log n))│ │ • 设置文件回调函数│ ││ epoll_wait() │├───────────────────────────────> │ • 检查rdllist是否为空│ │ • 空则阻塞进程│ ││ 数据到达 │ • 网卡中断/定时器触发│ │ • 调用ep_poll_callback()│ │ • 将epitem加入rdllist│ │ • 唤醒等待进程│ ││ <───────────────────────────────┤ • 复制events到用户空间│ │ • 清空rdllist(LT模式)│ ││ 处理事件 ││───────────────────────────────> ││ │
#include <iostream>
#include <cstring>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <sys/epoll.h>
#include <fcntl.h>
#include <vector>
#include <cerrno>
#include <arpa/inet.h>#define MAX_EVENTS 64
#define BUFFER_SIZE 1024class EpollServer {
private:int server_fd;int epoll_fd;struct sockaddr_in server_addr;public:EpollServer(int port) {// 创建socket SOCK_STREAM (TCP) SOCK_DERAM (UDP)server_fd = socket(AF_INET, SOCK_STREAM, 0);if (server_fd < 0) {perror("socket creation failed");exit(EXIT_FAILURE);}// 设置socket选项int opt = 1;if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt))) {perror("setsockopt failed");exit(EXIT_FAILURE);}// 绑定地址memset(&server_addr, 0, sizeof(server_addr));server_addr.sin_family = AF_INET;server_addr.sin_addr.s_addr = INADDR_ANY;server_addr.sin_port = htons(port);if (bind(server_fd, (struct sockaddr*)&server_addr, sizeof(server_addr)) < 0) {perror("bind failed");exit(EXIT_FAILURE);}// 创建epoll实例epoll_fd = epoll_create1(0);if (epoll_fd < 0) {perror("epoll_create1 failed");exit(EXIT_FAILURE);}// 添加server_fd到epollstruct epoll_event event;event.data.fd = server_fd;event.events = EPOLLIN | EPOLLET; // 边缘触发模式epoll_ctl(epoll_fd, EPOLL_CTL_ADD, server_fd, &event);// 监听socketif (listen(server_fd, SOMAXCONN) < 0) {perror("listen failed");return;}std::cout << "Server listening on port " << ntohs(server_addr.sin_port) << std::endl;}~EpollServer() {close(server_fd);close(epoll_fd);}// 设置非阻塞模式void set_nonblocking(int fd) {int flags = fcntl(fd, F_GETFL, 0);fcntl(fd, F_SETFL, flags | O_NONBLOCK);}void run() {// 事件循环struct epoll_event events[MAX_EVENTS];while (true) {int nfds = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);if (nfds < 0) {perror("epoll_wait failed");break;}for (int i = 0; i < nfds; i++) {if (events[i].data.fd == server_fd) {// 处理新连接handle_new_connection();} else {// 处理客户端数据if (events[i].events & EPOLLIN) {handle_client_data(events[i].data.fd);}if (events[i].events & (EPOLLRDHUP | EPOLLHUP | EPOLLERR)) {// 处理连接关闭或错误close_client(events[i].data.fd);}}}}}private:void handle_new_connection() {struct sockaddr_in client_addr;socklen_t addr_len = sizeof(client_addr);// 接受所有待处理的连接do {int client_fd = accept(server_fd, (struct sockaddr*)&client_addr, &addr_len);if (client_fd < 0) {if (errno == EAGAIN || errno == EWOULDBLOCK) {// 没有更多连接break;} else {perror("accept failed");break;}}else{std::cout << "New connection from " << inet_ntoa(client_addr.sin_addr) << ":" << ntohs(client_addr.sin_port) << std::endl;// 设置非阻塞set_nonblocking(client_fd);// 添加到epollstruct epoll_event event;event.data.fd = client_fd;event.events = EPOLLIN | EPOLLET | EPOLLRDHUP;if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, client_fd, &event) < 0) {perror("epoll_ctl add client failed");close(client_fd);}else{std::cout << "Successfully added client fd " << client_fd << " to epoll" << std::endl;}break;}}while (true);}void handle_client_data(int client_fd) {char buffer[BUFFER_SIZE];// 边缘触发模式需要读取所有可用数据while (true) {memset(buffer,0,BUFFER_SIZE);ssize_t bytes_read = read(client_fd, buffer, BUFFER_SIZE - 1);if (bytes_read > 0) {buffer[bytes_read] = '\0';std::cout << "Received from client " << client_fd << ": " << buffer;// 回显数据send(client_fd, buffer, bytes_read, 0);// 如果读取的数据少于缓冲区大小，说明数据已读完if (bytes_read < BUFFER_SIZE - 1) {break;}} else if (bytes_read == 0) {// 客户端关闭连接close_client(client_fd);break;} else {if (errno == EAGAIN || errno == EWOULDBLOCK) {// 数据已读完break;} else {perror("read failed");close_client(client_fd);break;}}}}void close_client(int client_fd) {std::cout << "Client " << client_fd << " disconnected" << std::endl;epoll_ctl(epoll_fd, EPOLL_CTL_DEL, client_fd, nullptr);close(client_fd);}
};int main() {EpollServer server(8080);server.run();return 0;
}
UDP实例
udp使用sendto-recvfrom收发消息；
- server
#include <iostream>
#include <cstring>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>#define BUFFER_SIZE 1024class UDPServer {
private:int server_fd;struct sockaddr_in server_addr;public:UDPServer(int port) {// 创建UDP socketserver_fd = socket(AF_INET, SOCK_DGRAM, 0);if (server_fd < 0) {perror("socket creation failed");exit(EXIT_FAILURE);}// 设置地址重用int opt = 1;if (setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt))) {perror("setsockopt failed");close(server_fd);exit(EXIT_FAILURE);}// 绑定地址memset(&server_addr, 0, sizeof(server_addr));server_addr.sin_family = AF_INET;server_addr.sin_addr.s_addr = INADDR_ANY; // 监听所有接口server_addr.sin_port = htons(port);if (bind(server_fd, (struct sockaddr*)&server_addr, sizeof(server_addr)) < 0) {perror("bind failed");close(server_fd);exit(EXIT_FAILURE);}std::cout << "UDP Server listening on port " << port << std::endl;}~UDPServer() { if (server_fd >= 0) {close(server_fd);}}void run() {char buffer[BUFFER_SIZE];struct sockaddr_in client_addr;socklen_t addr_len = sizeof(client_addr);while (true) {// 清空缓冲区memset(buffer, 0, BUFFER_SIZE);memset(&client_addr, 0, sizeof(client_addr));// 接收数据ssize_t recv_len = recvfrom(server_fd, buffer, BUFFER_SIZE - 1, 0,(struct sockaddr*)&client_addr, &addr_len);if (recv_len < 0) {perror("recvfrom failed");continue;}// 获取客户端信息char client_ip[INET_ADDRSTRLEN];inet_ntop(AF_INET, &client_addr.sin_addr, client_ip, INET_ADDRSTRLEN);std::cout << "Received from " << client_ip << ":" << ntohs(client_addr.sin_port) << " (" << recv_len << " bytes): " << buffer << std::endl;// 准备回显消息std::string echo_msg = "Echo: ";echo_msg += buffer;// 发送回显（发送到刚才接收的地址）sendto(server_fd, echo_msg.c_str(), echo_msg.length(), 0,(struct sockaddr*)&client_addr, addr_len);}}void run_with_select() {fd_set read_fds;struct timeval timeout;char buffer[BUFFER_SIZE];std::cout << "UDP Server with select() running..." << std::endl;while (true) {// 设置文件描述符集合FD_ZERO(&read_fds);FD_SET(server_fd, &read_fds);// 设置超时（5秒）timeout.tv_sec = 5;timeout.tv_usec = 0;// 使用select等待数据int ready = select(server_fd + 1, &read_fds, NULL, NULL, &timeout);if (ready < 0) {perror("select failed");break;} else if (ready == 0) {// 超时std::cout << "Timeout, waiting for data..." << std::endl;continue;}// 有数据可读if (FD_ISSET(server_fd, &read_fds)) {struct sockaddr_in client_addr;socklen_t addr_len = sizeof(client_addr);ssize_t recv_len = recvfrom(server_fd, buffer, BUFFER_SIZE - 1, 0,(struct sockaddr*)&client_addr, &addr_len);if (recv_len > 0) {buffer[recv_len] = '\0';std::cout << "Received: " << buffer << std::endl;// 回显sendto(server_fd, buffer, recv_len, 0,(struct sockaddr*)&client_addr, addr_len);}}}}
};int main() {UDPServer server(8080);server.run(); // 或 server.run_with_select();return 0;
}
- client
#include <iostream>
#include <cstring>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <string>class UDPClient {
private:int client_fd;struct sockaddr_in server_addr;public:UDPClient(const std::string& server_ip, int port) {// 创建UDP socketclient_fd = socket(AF_INET, SOCK_DGRAM, 0);if (client_fd < 0) {perror("socket creation failed");exit(EXIT_FAILURE);}// 设置服务器地址memset(&server_addr, 0, sizeof(server_addr));server_addr.sin_family = AF_INET;server_addr.sin_port = htons(port);// 转换IP地址if (inet_pton(AF_INET, server_ip.c_str(), &server_addr.sin_addr) <= 0) {std::cerr << "Invalid address: " << server_ip << std::endl;close(client_fd);exit(EXIT_FAILURE);}std::cout << "UDP Client ready to send to " << server_ip << ":" << port << std::endl;}~UDPClient() {if (client_fd >= 0) {close(client_fd);}}// 发送单条消息bool send_message(const std::string& message) {ssize_t sent = sendto(client_fd, message.c_str(), message.length(), 0,(struct sockaddr*)&server_addr, sizeof(server_addr));if (sent < 0) {perror("sendto failed");return false;}std::cout << "Sent " << sent << " bytes: " << message << std::endl;return true;}// 发送并接收回复（阻塞）bool send_and_receive(const std::string& message, int timeout_sec = 5) {// 发送消息if (!send_message(message)) {return false;}// 设置接收超时struct timeval timeout;timeout.tv_sec = timeout_sec;timeout.tv_usec = 0;if (setsockopt(client_fd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)) < 0) {perror("setsockopt timeout failed");}// 接收回复char buffer[1024];struct sockaddr_in from_addr;socklen_t addr_len = sizeof(from_addr);ssize_t recv_len = recvfrom(client_fd, buffer, sizeof(buffer) - 1, 0,(struct sockaddr*)&from_addr, &addr_len);if (recv_len < 0) {if (errno == EAGAIN || errno == EWOULDBLOCK) {std::cout << "Timeout: No response received" << std::endl;} else {perror("recvfrom failed");}return false;}buffer[recv_len] = '\0';char from_ip[INET_ADDRSTRLEN];inet_ntop(AF_INET, &from_addr.sin_addr, from_ip, INET_ADDRSTRLEN);std::cout << "Received from " << from_ip << ":" << ntohs(from_addr.sin_port) << " (" << recv_len << " bytes): " << buffer << std::endl;return true;}// 广播消息bool broadcast(const std::string& message, int port) {// 开启广播选项int broadcast_enable = 1;if (setsockopt(client_fd, SOL_SOCKET, SO_BROADCAST, &broadcast_enable, sizeof(broadcast_enable)) < 0) {perror("setsockopt broadcast failed");return false;}// 设置广播地址struct sockaddr_in broadcast_addr;memset(&broadcast_addr, 0, sizeof(broadcast_addr));broadcast_addr.sin_family = AF_INET;broadcast_addr.sin_port = htons(port);broadcast_addr.sin_addr.s_addr = htonl(INADDR_BROADCAST); // 255.255.255.255// 发送广播ssize_t sent = sendto(client_fd, message.c_str(), message.length(), 0,(struct sockaddr*)&broadcast_addr, sizeof(broadcast_addr));if (sent < 0) {perror("broadcast sendto failed");return false;}std::cout << "Broadcast " << sent << " bytes" << std::endl;return true;}
};int main() {UDPClient client("127.0.0.1", 8080);// 发送测试消息client.send_and_receive("Hello UDP Server!");client.send_and_receive("Another message");// 发送多条消息for (int i = 0; i < 5; i++) {std::string msg = "Message " + std::to_string(i);client.send_and_receive(msg);}return 0;
}
组播
组播地址范围：224.0.0.0 - 239.255.255.255
- 224.0.0.0 - 224.0.0.255：本地网络控制块
- 224.0.1.0 - 238.255.255.255：全球范围组播地址
- 239.0.0.0 - 239.255.255.255：本地管理组播地址
#include <iostream>
#include <cstring>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>class MulticastServer {
private:int sock_fd;struct sockaddr_in multicast_addr;public:MulticastServer(const std::string& multicast_ip, int port) {// 创建UDP socketsock_fd = socket(AF_INET, SOCK_DGRAM, 0);if (sock_fd < 0) {perror("socket creation failed");exit(EXIT_FAILURE);}// 设置组播地址memset(&multicast_addr, 0, sizeof(multicast_addr));multicast_addr.sin_family = AF_INET;multicast_addr.sin_port = htons(port);if (inet_pton(AF_INET, multicast_ip.c_str(), &multicast_addr.sin_addr) <= 0) {std::cerr << "Invalid multicast address" << std::endl;close(sock_fd);exit(EXIT_FAILURE);}// 设置TTL（生存时间）unsigned char ttl = 1; // 只在本地网络if (setsockopt(sock_fd, IPPROTO_IP, IP_MULTICAST_TTL, &ttl, sizeof(ttl)) < 0) {perror("setsockopt TTL failed");}std::cout << "Multicast Server ready for " << multicast_ip << ":" << port << std::endl;}void send_message(const std::string& message) {ssize_t sent = sendto(sock_fd, message.c_str(), message.length(), 0,(struct sockaddr*)&multicast_addr, sizeof(multicast_addr));if (sent < 0) {perror("multicast sendto failed");} else {std::cout << "Multicast sent " << sent << " bytes" << std::endl;}}
};#include <iostream>
#include <cstring>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>class MulticastClient {
private:int sock_fd;struct sockaddr_in local_addr;struct ip_mreq mreq;public:MulticastClient(const std::string& multicast_ip, int port) {// 创建UDP socketsock_fd = socket(AF_INET, SOCK_DGRAM, 0);if (sock_fd < 0) {perror("socket creation failed");exit(EXIT_FAILURE);}// 允许地址重用int reuse = 1;if (setsockopt(sock_fd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)) < 0) {perror("setsockopt reuseaddr failed");}// 绑定到任意地址和指定端口memset(&local_addr, 0, sizeof(local_addr));local_addr.sin_family = AF_INET;local_addr.sin_addr.s_addr = htonl(INADDR_ANY);local_addr.sin_port = htons(port);if (bind(sock_fd, (struct sockaddr*)&local_addr, sizeof(local_addr)) < 0) {perror("bind failed");close(sock_fd);exit(EXIT_FAILURE);}// 加入多播组mreq.imr_multiaddr.s_addr = inet_addr(multicast_ip.c_str());mreq.imr_interface.s_addr = htonl(INADDR_ANY);if (setsockopt(sock_fd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof(mreq)) < 0) {perror("setsockopt add membership failed");close(sock_fd);exit(EXIT_FAILURE);}std::cout << "Joined multicast group " << multicast_ip << " on port " << port << std::endl;}void receive_messages() {char buffer[1024];std::cout << "Listening for multicast messages..." << std::endl;while (true) {struct sockaddr_in from_addr;socklen_t addr_len = sizeof(from_addr);ssize_t recv_len = recvfrom(sock_fd, buffer, sizeof(buffer) - 1, 0,(struct sockaddr*)&from_addr, &addr_len);if (recv_len > 0) {buffer[recv_len] = '\0';char from_ip[INET_ADDRSTRLEN];inet_ntop(AF_INET, &from_addr.sin_addr, from_ip, INET_ADDRSTRLEN);std::cout << "Multicast from " << from_ip << ": " << buffer << std::endl;}}}
};