Linux poll 多路复用：select 的改良版及其局限

Linux poll 多路复用：select 的改良版及其局限

导读：在理解了 select 之后，我们来看看它的改进版本——poll。它通过更合理的数据结构解决了 fd 数量上限和接口混乱的问题，但核心的性能瓶颈依然存在。本文将深入解析 poll 的设计细节、对比分析以及完整服务器实现。

一、select 的痛点回顾

学习 poll 之前，有必要再次明确 select 的缺陷，因为 poll 的设计初衷就是为了解决这些问题：

1. fd 数量上限 1024

select 使用位图（bitmask），fd_set 大小固定。大多数系统下 FD_SETSIZE = 1024，超过这个数就无法监控了。

2. 接口设计不友好

select 需要三个独立的位图分别管理读、写和异常事件。每次调用前必须手动重建集合，且返回后集合会被修改，导致逻辑复杂。

// select 需要三个集合，每次都要重建
fd_set readfds, writefds;
FD_ZERO(&readfds);
FD_SET(fd, &readfds);
// ... select 返回后 readfds 被修改，下次循环前必须重新初始化

3. 核心性能问题未解决

poll 主要解决了问题 1 和 2，但问题 3（每次全量拷贝 + O(n) 遍历）依然保留，这需要等到 epoll 来解决。

二、poll 函数接口详解

1. 函数原型

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

相比 select，poll 的参数更少，不需要分开传递三个集合，接口更加简洁。

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 需要清零）。

3. 参数详解

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

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

4. 返回值

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

三、poll vs select：对比分析

1. 数据结构对比

select 使用位图方式，fd 信息分散在三个集合中；poll 则使用结构体数组，每个 fd 的信息自成一体。

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 的信息自成一体，不需要在三个独立的位图之间查找。

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. 优缺点总结

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 执行过程图解

一次 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 示例

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 实现

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}              ← 空闲槽位

2. 完整代码

这里给出一个基于 C++ 的简单实现，包含 Socket 封装和事件循环。

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

3. 关键实现细节解析

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

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

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

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

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

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

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

// 当数组满了时，可以扩容
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 完整对比

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）	连接数中等	高并发（万级以上）

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 的使用场景与选择建议

1. 什么时候选 poll 而不是 select？

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

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

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

决策树：

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

九、常见问题解答

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 监控，发完了再关掉。

2. poll 超时精度如何？

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

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) {
    /* 对端关闭 */
}

十、总结

1. 核心要点

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

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 高性能服务器的基石，也是面试最高频的考点，务必吃透。