Linux poll 多路复用详解：select 的改进与局限

Linux poll 多路复用详解：select 的改进与局限

在之前的讨论中，我们深入分析了 select 机制及其存在的四个主要缺陷。poll 正是针对其中最棘手的问题——文件描述符数量上限——做出的改进。它通过更合理的数据结构替代了位图，不仅让接口更加清晰，也彻底去掉了 1024 的限制。不过，poll 并没有从根本上解决 select 的所有问题，核心的'每次全量拷贝 + O(n) 遍历'依然存在。

本文将深度解析 poll 的接口设计，讲清楚它相比 select 的进步在哪里，局限在哪里，最后用 poll 实现一个完整的服务器。理解了 poll，后面对 epoll 的学习会更有感觉——因为你能看清楚每一步改进背后的动机。

一、select 的痛点回顾

1.1 select 的问题在哪里？

学习 poll 之前，先把 select 的缺陷再明确一下，因为 poll 的设计就是奔着解决这些问题去的：

问题 1：fd 数量上限 1024

// select 用位图，fd_set 大小固定
// FD_SETSIZE = 1024（多数系统）
// 超过 1024 个连接直接没辙

问题 2：接口设计不友好

// select 用三个独立的位图，输入输出混在一起
// 每次调用前必须手动重建集合
// 读就绪、写就绪、异常三个 fd_set 分开管理，麻烦
fd_set readfds, writefds, exceptfds;
FD_ZERO(&readfds);
FD_SET(fd,&readfds);
// select 返回后 readfds 被修改，必须重建...

问题 3 & 4：每次拷贝 + O(n) 遍历（核心性能问题，poll 没解决）

poll 主要解决了问题 1 和 2，问题 3 和 4 要等 epoll 来解决。

二、poll 函数接口详解

2.1 函数原型

#include <poll.h>
int poll(struct pollfd *fds, nfds_t nfds, int timeout);

和 select 相比，参数少了一个（不需要分开传三个 fd_set），接口更简洁。

2.2 核心数据结构：pollfd

poll 的关键在于 pollfd 结构体：

struct pollfd {
    int fd;       /* 要监控的文件描述符 */
    short events; /* 关注的事件（输入参数）*/
    short revents;/* 实际发生的事件（输出参数）*/
};

events 和 revents 的取值：

宏名	值	含义
POLLIN	0x0001	数据可读（包括普通数据和优先数据）
POLLPRI	0x0002	高优先级数据可读（带外数据）
POLLOUT	0x0004	数据可写
POLLERR	0x0008	发生错误（仅 revents 有效）
POLLHUP	0x0010	挂断（仅 revents 有效）
POLLNVAL	0x0020	非法的 fd（仅 revents 有效）

关键设计：events 和 revents 分开！

• events：你设置，告诉内核你关注什么（输入）
• revents：内核设置，告诉你实际发生了什么（输出）

poll 返回后，events 不会被修改，只有 revents 被更新。这解决了 select 每次要重建集合的问题：你只需要检查 revents，而 events 始终保持你的设置，下次调用时不需要重新赋值（但 revents 需要清零）。

2.3 参数详解

• 参数 fds：pollfd 结构体数组的首地址，每个元素对应一个要监控的 fd。
• 参数 nfds：fds 数组的长度，即监控的 fd 数量。
• 参数 timeout：超时时间（毫秒）。

timeout 值	行为
-1	无限等待（永远阻塞）
0	立即返回，只检查当前状态
> 0	等待最多 timeout 毫秒

2.4 返回值

int ret = poll(fds, nfds, timeout);
// ret > 0：就绪的 fd 数量
// ret == 0：超时
// ret < 0：出错，查 errno

三、poll vs select：对比分析

3.1 数据结构对比

select 的位图方式：
fd_set：[bit0, bit1, bit2, ..., bit1023] 最多 1024 个 fd

poll 的 pollfd 数组：
pollfd[0]：{fd=3, events=POLLIN, revents=0}
pollfd[1]：{fd=5, events=POLLIN|POLLOUT, revents=0}
pollfd[2]：{fd=7, events=POLLIN, revents=0}
... 数组大小由用户决定，理论上无上限

poll 就像把 select 的位图升级成了一个更富有表达力的结构体数组。每个 fd 的信息自成一体，不需要在三个独立的位图之间查找。

3.2 使用方式对比

select 的使用（每次循环都很麻烦）：

// select：每次循环必须重建三个 fd_set
for(;;){
    fd_set readfds, writefds;
    FD_ZERO(&readfds);
    FD_ZERO(&writefds);
    for(int i = 0; i < n; i++)
        FD_SET(fds[i],&readfds);
    select(max_fd + 1,&readfds,&writefds,NULL,&timeout);
    // 检查结果
    for(int i = 0; i < n; i++){
        if(FD_ISSET(fds[i],&readfds)){
            /* 处理 */
        }
    }
}

poll 的使用（清晰多了）：

// poll：只需要清零 revents，events 保持不变
struct pollfd pfds[MAX_FDS];
// 初始化一次就好：
pfds[0]={fd1, POLLIN, 0};
pfds[1]={fd2, POLLIN | POLLOUT, 0};
for(;;){
    // 清零所有 revents（可选，但推荐）
    for(int i = 0; i < n; i++)
        pfds[i].revents = 0;
    poll(pfds, n,-1);
    // 检查结果
    for(int i = 0; i < n; i++){
        if(pfds[i].revents & POLLIN){
            /* 处理读 */
        }
        if(pfds[i].revents & POLLOUT){
            /* 处理写 */
        }
    }
}

3.3 优缺点总结

poll 相比 select 的优点：

改进项	select	poll
fd 数量限制	1024（固定）	无上限（数组大小可动态扩展）
接口设计	三个位图，输入输出混用	pollfd 结构体，events/revents 分离
重建集合	每次必须重建	events 保持不变，只需清零 revents
事件表达	三个集合（读/写/异常）	单结构体内用 events/revents 标志

poll 与 select 共同的缺点（核心性能问题）：

问题	select	poll
用户态到内核态拷贝	每次拷贝整个 fd_set	每次拷贝整个 pollfd 数组
内核查找就绪 fd	遍历所有 fd，O(n)	遍历所有 pollfd，O(n)

结论：poll 是 select 的改良版，解决了接口设计问题和数量限制，但没有从根本上解决性能问题。当连接数成千上万时，poll 和 select 都会因为 O(n) 遍历而性能下降。

四、poll 执行过程图解

4.1 一次 poll 调用的完整流程

用户态 内核态
|||
初始化 pollfd 数组
pfds[0]={3, POLLIN, 0}
pfds[1]={5, POLLIN, 0}
pfds[2]={7, POLLIN, 0}
||||
poll(pfds, 3, -1)
--- 拷贝 3 个 pollfd 到内核 --->
||||
(程序阻塞在 poll)
轮询每个 fd 的状态
fd=3：未就绪
fd=5：就绪！(有数据)
fd=7：未就绪
||||
设置就绪的 revents:
pfds[1].revents = POLLIN
||
<--- 返回 1（1 个 fd 就绪）-----
||||
检查 pfds[i].revents
pfds[1].revents & POLLIN → 处理 fd=5

五、最简单的 poll 示例

5.1 使用 poll 监控标准输入

#include <stdio.h>
#include <unistd.h>
#include <poll.h>

int main(){
    // 定义一个 pollfd，监控标准输入（fd=0）
    struct pollfd poll_fd;
    poll_fd.fd = 0;
    poll_fd.events = POLLIN; // 关注可读事件
    poll_fd.revents = 0;
    
    for(;;){
        // 超时 1000ms（1 秒）
        int ret = poll(&poll_fd, 1, 1000);
        if(ret < 0){
            perror("poll");
            continue;
        }
        if(ret == 0){
            printf("poll timeout（1 秒内无输入）\n");
            continue;
        }
        // 检查是否有读事件就绪
        if(poll_fd.revents & POLLIN){
            char buf[1024]={0};
            read(0, buf,sizeof(buf)-1);
            printf("stdin: %s", buf);
        }
        // 清零 revents，为下一轮做准备（poll 不会自动清零）
        poll_fd.revents = 0;
    }
    return 0;
}

运行效果：

• 1 秒内没有输入 → 打印 "poll timeout"
• 有输入 → 打印输入内容

六、完整的 PollServer 实现

6.1 设计思路

用 poll 实现服务器，核心思路和 select 版本一样：

1. 维护一个 pollfd 数组，代替 select 的 fd_set
2. 用 -1 标记"这个槽位空闲"（因为 poll 数组可能有空洞）
3. 新连接来了，找一个空闲槽位放入；连接断开，把那个槽位标记为 -1

教学版使用阻塞 socket + poll，适合短消息；生产环境通常配合 非阻塞 fd + 发送缓冲，避免慢连接在 send() 上阻塞整个事件循环。

pollfd 数组的管理：
[0]: {listensock, POLLIN, 0} ← 监听 socket，一直在
[1]: {fd=4, POLLIN, 0} ← 客户端 A
[2]: {-1, 0, 0} ← 空闲槽位
[3]: {fd=6, POLLIN, 0} ← 客户端 B
[4]: {-1, 0, 0} ← 空闲槽位

6.2 完整代码

// poll_server.hpp
#pragma once
#include <iostream>
#include <string>
#include <memory>
#include <poll.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <cerrno>
#include <cstring>

const static int g_default_port = 8888;
const static int g_backlog = 8;
const static int g_max_fds = 1024; // pollfd 数组的初始大小

/**
 * 简单的 TCP Socket 封装（复用自上一篇）
 */
class TcpSocket{
public:
    TcpSocket(int fd = -1):fd_(fd){}
    int GetFd() const{return fd_;}
    bool Build(int port){
        fd_ = socket(AF_INET, SOCK_STREAM, 0);
        if(fd_ < 0) return false;
        int opt = 1;
        setsockopt(fd_, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = INADDR_ANY;
        if(bind(fd_, (struct sockaddr*)&addr, sizeof(addr)) < 0) return false;
        if(listen(fd_, g_backlog) < 0) return false;
        return true;
    }
    int AcceptConnection(std::string* ip = nullptr, uint16_t* port = nullptr){
        struct sockaddr_in peer;
        socklen_t len = sizeof(peer);
        int sock = accept(fd_, (struct sockaddr*)&peer, &len);
        if(sock < 0) return -1;
        if(ip) *ip = inet_ntoa(peer.sin_addr);
        if(port) *port = ntohs(peer.sin_port);
        return sock;
    }
    int GetSockFd() const{return fd_;}
private:
    int fd_;
};

/**
 * 基于 poll 的 TCP 服务器
 */
class PollServer{
public:
    PollServer(int port = g_default_port):
        _port(port),
        _listen_sock(std::make_unique<TcpSocket>()),
        _is_running(false),
        _num(g_max_fds){}

    void InitServer(){
        // 初始化监听 socket
        if(!_listen_sock->Build(_port)){
            perror("build listen socket failed");
            exit(1);
        }
        printf("[PollServer] 服务器初始化完成，监听端口 %d\n", _port);
        // 初始化 pollfd 数组
        _rfds = new struct pollfd[_num];
        for(int i = 0; i < _num; i++){
            _rfds[i].fd = -1; // -1 表示空闲槽位
            _rfds[i].events = 0;
            _rfds[i].revents = 0;
        }
        // 把监听 socket 放入数组第 0 号位
        _rfds[0].fd = _listen_sock->GetSockFd();
        _rfds[0].events = POLLIN;
    }

    void Loop(){
        _is_running = true;
        while(_is_running){
            PrintDebug();
            int timeout = -1; // 永久阻塞
            int n = poll(_rfds, _num, timeout);
            switch(n){
                case 0: printf("[PollServer] poll 超时\n"); break;
                case -1: perror("poll error"); break;
                default: // 有 n 个 fd 就绪
                    HandleEvent(n); break;
            }
        }
        _is_running = false;
    }

    void Stop(){
        _is_running = false;
    }

    ~PollServer(){
        delete[] _rfds;
    }

private:
    /**
     * 处理就绪事件
     */
    void HandleEvent(int ready_count){
        for(int i = 0; i < _num; i++){
            if(_rfds[i].fd == -1) continue; // 跳过空闲槽位
            int fd = _rfds[i].fd;
            short revents = _rfds[i].revents;
            // 只处理读就绪事件
            if(!(revents & POLLIN)) continue;
            
            if(fd == _listen_sock->GetSockFd()){
                // 监听 socket 就绪：有新连接
                HandleNewConnection();
            } else {
                // 普通 socket 就绪：有数据可读
                HandleData(i, fd);
            }
            // 清零 revents（poll 不会自动清零）
            _rfds[i].revents = 0;
        }
    }

    /**
     * 处理新连接
     */
    void HandleNewConnection(){
        std::string client_ip;
        uint16_t client_port;
        int sock = _listen_sock->AcceptConnection(&client_ip, &client_port);
        if(sock == -1){
            perror("accept error");
            return;
        }
        printf("[PollServer] 新连接：%s:%d, fd=%d\n", client_ip.c_str(), client_port, sock);
        // 在 pollfd 数组中找一个空闲槽位
        int pos = FindEmptySlot();
        if(pos == -1){
            // 数组满了，可以扩容或拒绝
            printf("[PollServer] 服务器已满，拒绝连接 fd=%d\n", sock);
            close(sock);
            return;
        }
        // 将新连接加入 pollfd 数组
        _rfds[pos].fd = sock;
        _rfds[pos].events = POLLIN;
        _rfds[pos].revents = 0;
        printf("[PollServer] fd=%d 加入监控，位置 pos=%d\n", sock, pos);
    }

    /**
     * 处理普通连接的数据
     */
    void HandleData(int pos, int fd){
        char buffer[1024]={0};
        ssize_t n = recv(fd, buffer, sizeof(buffer)-1, 0);
        if(n > 0){
            // 正常数据
            buffer[n]='\0';
            printf("[PollServer] fd=%d 收到：%s\n", fd, buffer);
            // 简单的回显服务：原样返回
            std::string response = std::string("服务器收到：") + buffer;
            send(fd, response.c_str(), response.size(), 0);
        } else if(n == 0){
            // 客户端正常关闭
            printf("[PollServer] fd=%d 正常断开\n", fd);
            CloseConnection(pos);
        } else {
            // 出错
            if(errno != EINTR){
                perror("recv error");
                printf("[PollServer] fd=%d 出错，关闭\n", fd);
                CloseConnection(pos);
            }
        }
    }

    /**
     * 关闭连接，清理 pollfd 槽位
     */
    void CloseConnection(int pos){
        close(_rfds[pos].fd);
        _rfds[pos].fd = -1;
        _rfds[pos].events = 0;
        _rfds[pos].revents = 0;
    }

    /**
     * 在数组中找第一个空闲槽位（fd == -1）
     */
    int FindEmptySlot(){
        for(int i = 1; i < _num; i++){
            // 从 1 开始，0 是 listen_sock
            if(_rfds[i].fd == -1){
                return i;
            }
        }
        return -1; // 没有空闲槽位
    }

    /**
     * 打印当前监控的 fd 列表（调试用）
     */
    void PrintDebug(){
        printf("[PollServer] 当前监控的 fd：");
        for(int i = 0; i < _num; i++){
            if(_rfds[i].fd != -1){
                printf("%d ", _rfds[i].fd);
            }
        }
        printf("\n");
    }

private:
    int _port;
    std::unique_ptr<TcpSocket> _listen_sock;
    bool _is_running;
    struct pollfd* _rfds; // pollfd 数组
    int _num; // 数组大小
};

主函数：

// poll_main.cc
#include "poll_server.hpp"

int main(int argc, char* argv[]){
    int port = (argc > 1) ? atoi(argv[1]) : g_default_port;
    PollServer server(port);
    server.InitServer();
    server.Loop();
    return 0;
}

编译运行：

# 编译
g++ -std=c++14 poll_main.cc -o poll_server
# 运行（监听 8888 端口）
./poll_server 8888
# 另一个终端测试
nc 127.0.0.1 8888
hello world

6.3 关键实现细节解析

1. 为什么用 -1 标记空闲槽位？

poll 传入的是一个数组，内核会遍历 [0, nfds) 的每个元素。如果一个槽位不再使用但没有清理，内核会继续处理它，可能会产生意外行为。

用 fd = -1 标记空闲，poll 会自动忽略 fd 为负数的 pollfd（POSIX 标准保证），这是一个优雅的处理方式。

// poll 的行为：fd < 0 的元素会被忽略，revents 保持 0
// 所以可以安全地在数组中留下 fd = -1 的元素

2. events 不会被修改，但 revents 不会自动清零

// poll 返回后：
// events：不变，始终是你设置的关注事件
// revents：被内核设置为实际发生的事件
// 需要注意：revents 在下一次 poll 前应该清零
// 否则上次的结果会干扰判断
_rfds[i].revents = 0; // 处理完事件后清零

3. 扩容逻辑（生产环境应考虑）

// 当数组满了时，可以扩容
void Expand(){
    int new_num = _num * 2;
    struct pollfd* new_fds = new struct pollfd[new_num];
    // 拷贝旧数组内容
    memcpy(new_fds, _rfds, _num * sizeof(struct pollfd));
    // 初始化新增部分
    for(int i = _num; i < new_num; i++){
        new_fds[i].fd = -1;
        new_fds[i].events = 0;
        new_fds[i].revents = 0;
    }
    delete[] _rfds;
    _rfds = new_fds;
    _num = new_num;
    printf("[PollServer] 扩容至 %d 个槽位\n", _num);
}

七、select vs poll vs epoll 完整对比

7.1 三者对比总结（重要，面试必背）

比较项	select	poll	epoll
fd 数量限制	1024（FD_SETSIZE）	无限制	无限制
数据结构	三个位图（fd_set）	pollfd 数组	红黑树 + 就绪队列
用户到内核拷贝	每次全量拷贝	每次全量拷贝	只在 ctl 时拷贝
查找就绪 fd	遍历所有，O(n)	遍历所有，O(n)	回调机制，O(k)
集合重建	每次必须重建	events 保留，revents 清零	内核维护，无需重建
工作模式	LT 模式	LT 模式	LT + ET 模式
跨平台	所有平台支持	类 Unix 平台支持	Linux 专属
适用场景	连接数少（<100）	连接数中等	高并发（万级以上）

7.2 性能对比直觉

假设服务器有 10000 个连接，每次只有 10 个有数据：
select/poll 的工作：
- 每次拷贝 10000 个 fd 信息到内核
- 内核遍历 10000 个 fd，找到 10 个就绪的
- 再拷贝回用户态
- O(10000) 的工作量

epoll 的工作：
- 内核维护红黑树，新 fd 只需注册一次
- 有数据时通过回调直接加入就绪队列
- epoll_wait 只返回 10 个就绪的 fd
- epoll_wait 返回就绪列表，遍历成本与就绪数 k 相关（O(k)），避免每次全量扫描 n 个 fd。
连接数越多，差距越大

八、poll 的使用场景与选择建议

8.1 什么时候选 poll 而不是 select？

• 需要监控超过 1024 个 fd（虽然现在这种场景更应该用 epoll）
• 代码已经用了 select，且 fd 数量接近上限，需要简单升级
• 目标平台不支持 epoll（非 Linux 系统）

8.2 什么时候应该直接用 epoll 而不是 poll？

• 连接数超过几百个
• 需要高并发性能
• 在 Linux 系统上开发

决策树：
要监控多个 fd？
→ 连接数少（<100）且需要跨平台？ → 用 select
→ 不需要跨平台，Linux 系统？ → 直接用 epoll（跳过 poll）
→ 需要跨平台，连接数中等？ → 用 poll

九、常见问题解答

9.1 poll 能同时监控读和写吗？

可以，在 events 中同时设置 POLLIN | POLLOUT：

pfds[i].events = POLLIN | POLLOUT; // 同时关注读和写
// poll 返回后
if(pfds[i].revents & POLLIN){
    /* 有数据可读 */
}
if(pfds[i].revents & POLLOUT){
    /* 发送缓冲区有空间 */
}

实践建议：不要一直开启 POLLOUT 监控。发送缓冲区通常都有空间，POLLOUT 几乎总是就绪，这样 poll 会不停返回，浪费 CPU。只在发送缓冲区满了、数据没发完的时候才开启 POLLOUT 监控，发完了再关掉。

9.2 poll 超时精度如何？

poll 的 timeout 参数单位是毫秒，精度受系统时钟分辨率影响，通常精度在 10ms 级别。如果需要更高精度（微秒级），需要使用 epoll_wait 或者其他高精度定时器。

9.3 POLLHUP 和 POLLERR 需要手动监控吗？

不需要。POLLHUP（挂断）和 POLLERR（错误）不需要在 events 中设置，内核会自动在 revents 中设置它们，即使你的 events 没有包含这两个标志。

// 不需要：
// pfds[i].events = POLLIN | POLLERR | POLLHUP;
// 只需要：
pfds[i].events = POLLIN;
// poll 返回后，POLLERR 和 POLLHUP 也可能被设置在 revents 中
if(pfds[i].revents & POLLERR){
    /* 处理错误 */
}
if(pfds[i].revents & POLLHUP){
    /* 对端关闭 */
}

十、总结

10.1 核心要点

#	要点	关键点
1	pollfd 结构体	fd + events（输入）+ revents（输出），设计比 select 清晰
2	无数量限制	数组大小由用户决定，可动态扩容
3	-1 标记空闲	poll 自动忽略 fd < 0 的条目
4	revents 需手动清零	poll 不会自动清零，每次处理后需手动清零
5	仍是 O(n)	全量拷贝 + 全量遍历的问题未解决

10.2 记忆技巧

poll = select 的升级版：
select 的位图 → poll 的 pollfd 结构体
1024 上限 → 无上限（数组大小由你定）
三个集合混乱 → 一个结构体，events/revents 分离

但 poll 没解决的：
每次全量拷贝到内核 → epoll 来解决
O(n) 遍历找就绪 fd → epoll 来解决

总结：poll 是 select 的进化版，核心改进是用 pollfd 结构体替代位图，消除了 fd 数量上限，让接口更清晰。但 O(n) 遍历和每次全量拷贝的性能问题依然存在。下一篇，我们终于要进入重头戏——epoll，它用红黑树 + 就绪队列 + 回调机制彻底解决了这两个问题，并引入了 LT/ET 两种工作模式。epoll 是 Linux 高性能服务器的基石，也是面试最高频的考点，务必吃透。