全栈攻城狮

Appearance

Selector / EPoll 多路复用源码精读 ​

概述 ​

Selector 让一个线程管理成千上万连接，是 Reactor 模式（Netty 的核心）的基础。Linux 上 Java 的 Selector 默认走 epoll：一个 epollfd 注册所有关注的事件，epoll_wait 一次性返回就绪事件列表，O(1) 复杂度与"事件驱动"的结合让它远超传统轮询。本文基于 OpenJDK 21 源码拆解 EPollSelectorImpl 与 WindowsSelectorImpl。

一、EPollSelectorImpl 的整体结构 ​

java
// sun.nio.ch.EPollSelectorImpl
public final class EPollSelectorImpl extends SelectorImpl {
    private final EPollArrayWrapper pollWrapper;   // epoll 封装（fd + 事件数组）
    private final int fd0, fd1;                    // 唤醒管道（pipe[0]/pipe[1]）
    private final int[] registeredChannels;        // 已注册通道映射

    public EPollSelectorImpl(SelectorProvider sp) {
        super(sp);
        pollWrapper = new EPollArrayWrapper();     // 创建 epollfd
        fd0 = pollWrapper.epollfd;                 // epoll 实例 fd
        // 创建唤醒管道
        try {
            fd1 = Net.socket(true);                // socketpair / pipe
            fd0 = Net.pipe(true);                  // 管道的两个端
        } ...
    }
}
java
// sun.nio.ch.EPollArrayWrapper
public final class EPollArrayWrapper {
    private final int epfd;                        // epoll 实例句柄
    private int outgoingInterruptFD;               // 唤醒管道写端
    private EPollEventArray events;                // 事件结果数组

    EPollArrayWrapper() throws IOException {
        // ① epoll_create：创建 epoll 实例
        epfd = epollCreate();
        // ② 预分配事件数组（默认 128 个事件容量）
        events = new EPollEventArray(SIZE_DEFAULT);
    }
}
epoll 三个系统调用：
  epoll_create()      → 创建 epoll 实例，返回 epfd
  epoll_ctl(epfd, op, fd, event) → 注册/修改/删除关注事件
  epoll_wait(epfd, events, max, timeout) → 等待就绪事件

二、select() 与 doSelect() 的核心循环 ​

java
// java.nio.channels.Selector（JDK 21 用 select() 与 selectNow()）
public int select() throws IOException {
    return select(0);            // 永久阻塞
}
public int select(long timeout) throws IOException {
    ...
    lockAndDoSelect((timeout == 0) ? -1 : timeout);   // -1 表示无限等待
}

// SelectorImpl
public int select(long timeout) throws IOException {
    synchronized (this) {
        ...
        lockAndDoSelect(timeout);
    }
}
java
// EPollSelectorImpl.doSelect
protected int doSelect(Consumer<SelectionKey> action, long timeout) throws IOException {
    ...
    // ① 更新兴趣事件到内核：把新注册/修改的事件 epoll_ctl 到 epfd
    pollWrapper.updateRegistrations();

    // ② 阻塞等待就绪事件
    long to = timeout;
    int numEntries;
    if (to == 0) {                       // 永久阻塞模式
        numEntries = pollWrapper.poll(timeout);   // epollWait(epfd, events, -1)
    } else {
        numEntries = pollWrapper.poll(timeout);   // epollWait(epfd, events, timeout)
    }
    ...
    // ③ 处理唤醒管道（中断信号）
    if (pollWrapper.interrupted()) { ... }

    // ④ 处理就绪事件：遍历 events 数组
    int numKeysUpdated = 0;
    if (numEntries > 0) {
        for (int i = 0; i < numEntries; i++) {
            int fd = pollWrapper.getDescriptor(i);        // 就绪的 fd
            int ev = pollWrapper.getEventOps(i);          // 就绪的事件
            SelectionKeyImpl ski = fdToKey.get(fd);       // 反向映射到 key
            ...
            ski.readyOps(ev);                             // 更新 readyOps
            numKeysUpdated++;
        }
    }
    return numKeysUpdated;
}
doSelect 完整流程：
  1. updateRegistrations：把注册队列中的新通道 epoll_ctl 加进 epfd
  2. epollWait(epfd, events, timeout)：阻塞等待内核就绪事件
  3. 检查唤醒管道（wakeup 信号）
  4. 遍历就绪事件数组 → fd 反查 SelectionKey → 更新 readyOps
  5. 返回就绪 key 数量
java
// EPollArrayWrapper.poll
int poll(long timeout) throws IOException {
    // epollWait native
    updateRegistrations();
    int n = epollWait(epfd, events.pollArray, events.size, timeout);
    return n;
}

关键点：epollWait 返回的是就绪 fd 列表（不是全量扫描），这正是 epoll 相比 select/poll 的 O(1) 优势——select/poll 每次都要全量遍历所有 fd 检查状态。

三、register 注册事件 ​

java
// 用户侧调用链：
// channel.register(selector, ops, attachment)
// → SelectorImpl.register(SelectableChannel, int, Object)
// → implRegister(SelectionKeyImpl)

// EPollSelectorImpl.implRegister
protected void implRegister(SelectionKeyImpl ski) {
    SelectableChannel ch = ski.channel();
    fdToKey.put(ski.getFDVal(), ski);       // fd → key 反向映射
    pollWrapper.add(ski.getFDVal());        // 记录到注册列表（待内核注册）
    registeredChannels[ski.getIndex()] = ski.getFDVal();
}

// EPollArrayWrapper.add
void add(int fd) {
    // 放入更新队列，select 时统一 epoll_ctl_add
    updateRegistrations();
    ...
}
注册流程：
  ① implRegister：记录 fd→key 映射，fd 加入待更新队列
  ② 下次 select → updateRegistrations → epoll_ctl_add(epfd, fd, interestOps)
  ③ 内核监听该 fd 的 EPOLLIN/EPOLLOUT 事件
java
// EPollArrayWrapper.updateRegistrations 核心
private void updateRegistrations() {
    synchronized (updateLock) {
        int j = 0;
        while (j < updateCount) {
            int fd = updateDescriptors[j];
            // 已注册 → epoll_ctl_mod；新注册 → epoll_ctl_add
            int opcode = (registeredEvents[fd] == 0) ? EPOLL_CTL_ADD : EPOLL_CTL_MOD;
            epollCtl(epfd, opcode, fd, events);   // native epoll_ctl
            registeredEvents[fd] = events;
            j++;
        }
        updateCount = 0;
    }
}

四、deregister / cancel 删除事件 ​

java
// SelectionKeyImpl.cancel（用户取消注册）
public void cancel() {
    ...
    // 通知 Selector 删除该通道的关注事件
    ((AbstractSelector)selector).cancel(this);
}

// EPollSelectorImpl.implDereg
protected void implDereg(SelectionKeyImpl ski) {
    int fd = ski.getFDVal();
    fdToKey.remove(fd);                 // 移除映射
    pollWrapper.release(fd);            // 从 epoll 删除
    registeredChannels[ski.getIndex()] = 0;
}

// EPollArrayWrapper.release
void release(int fd) {
    // 加入删除队列，select 时执行 epoll_ctl_del
    epollCtl(epfd, EPOLL_CTL_DEL, fd, null);   // native：从 epoll 摘除
    registeredEvents[fd] = 0;
}
删除语义：
  用户 cancel → 不立即删除（惰性）→ 下次 select 统一 epoll_ctl_del
  删除后：该 fd 不再产生就绪事件（即使有数据）
  key 被取消后仍可读 selectedKeys 中的旧事件（已处理完）

五、wakeup() 管道唤醒 ​

java
// EPollSelectorImpl.wakeup
public Selector wakeup() {
    synchronized (interruptLock) {
        if (!interruptTriggered) {
            pollWrapper.interrupt();     // 往管道写一个字节
            interruptTriggered = true;
        }
    }
    return this;
}

// EPollArrayWrapper.interrupt
void interrupt() {
    // 写 1 字节到管道写端
    int n = write1(outgoingInterruptFD, 1);
    ...
}
wakeup 机制：
  阻塞在 epollWait 的线程看不到普通唤醒 → 用一个"管道"作为唤醒信号
  wakeup() → 往 pipe 写 1 字节 → 内核 epollWait 因管道可读而返回
  下次 select 时读掉该字节（interrupted 处理）恢复阻塞

价值：另一个线程可以在任何时刻打断 select 的阻塞等待
（例如：关闭 selector、添加紧急任务、超时控制）

六、WindowsSelectorImpl 的实现差异 ​

java
// sun.nio.ch.WindowsSelectorImpl
public final class WindowsSelectorImpl extends SelectorImpl {
    private PollArrayWrapper pollWrapper;      // 轮询数组（fd 列表）
    private final int[] readFds, writeFds, exceptFds;  // 三个 fd 集合
    private final int[] channelArray;         // fd → channel 映射
    private final int threadCount;            // 子线程数
    private final int[] thread0Array;         // 子线程负责的 fd 段
}
Windows 实现差异：
  早期：select() → WSAPoll（同 poll 模型，全量扫描）
  现代：WaitForMultipleObjects + 子线程辅助
        → 主线程用事件对象等待（可被唤醒）
        → 子线程负责 WSAPoll 各自 fd 段，就绪后通知主线程
  vs Linux：epoll 事件驱动 O(1)；Windows 仍是轮询分摊
java
// WindowsSelectorImpl.doSelect
protected int doSelect(Consumer<SelectionKey> action, long timeout) throws IOException {
    ...
    // 主线程等待事件对象 + 子线程轮询结果
    // 子线程：每个负责 channelArray 的一段 → WSAPoll
    // 主线程：WaitForMultipleObjects(事件 + 唤醒事件)
}

性能对比：Linux epoll 在海量连接、少量活跃场景下远优于 Windows 的轮询模型；这也是生产环境网络服务多部署 Linux 的原因之一。

七、selectedKeys() 的迭代 ​

java
// SelectorImpl.selectedKeys（JDK 21 直接返回集合）
public Set<SelectionKey> selectedKeys() { return selectedKeys; }

// 用户迭代模式
public void process(Selector selector) {
    Set<SelectionKey> keys = selector.selectedKeys();
    Iterator<SelectionKey> it = keys.iterator();
    while (it.hasNext()) {
        SelectionKey key = it.next();
        it.remove();                       // 必须移除，否则下次会重复处理
        if (key.isAcceptable()) { ... }
        else if (key.isReadable()) { ... }
    }
}
selectedKeys 迭代要点：
  每次 select 后把就绪 key 加入 selectedKeys（不自动清理）
  用户处理完必须 it.remove() → 否则同一事件反复处理
  readyOps 在下次 select 时被覆盖更新
  键集合非线程安全 → 需在 select 线程内处理

八、实现要点 ​

Selector / EPoll 核心：
  EPollSelectorImpl：epoll_create + EPollArrayWrapper（事件数组）
  doSelect：updateRegistrations → epollWait → 遍历就绪 fd 反查 key
  register：implRegister 记录 → 下次 select 统一 epoll_ctl_add/mod
  cancel：惰性删除 → epoll_ctl_del
  wakeup：管道写 1 字节打断 epollWait
  Windows：WSAPoll + 子线程分摊（轮询模型）
  selectedKeys：迭代时手动 remove，防重复处理

常见陷阱：
  selectedKeys 忘记 remove → 事件重复处理/死循环
  在 select 线程外修改 interestOps → 需 wakeup
  海量连接但活跃少 → epoll 优势明显；活跃多 → 与轮询差异缩小
  register 后未 select 事件不生效（更新是惰性的）
  多线程 select 同一 selector → 不允许（同一时刻仅一个线程 select）