五种I/O模型和Java NIO源码分析

class EPollSelectorImpl extends SelectorImpl
{
    .....
    // The poll object
    EPollArrayWrapper pollWrapper;
    .....
    EPollSelectorImpl(SelectorProvider sp) throws IOException {
        .....
        pollWrapper = new EPollArrayWrapper();
        pollWrapper.initInterrupt(fd0, fd1);
        .....
    }
    .....
}

    private native int epollCreate();
    private native void epollCtl(int epfd, int opcode, int fd, int events);
    private native int epollWait(long pollAddress, int numfds, long timeout,
                                 int epfd) throws IOException;

//创建一个epoll句柄，size是这个监听的数目的最大值．
int epoll_create(int size);
//事件注册函数，告诉内核epoll监听什么类型的事件，参数是感兴趣的事件类型，回调和监听的fd
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
//等待事件的产生，类似于select调用，events参数用来从内核得到事件的集合
int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);

protected final SelectionKey register(AbstractSelectableChannel ch,
                                      int ops,
                                      Object attachment)
{
    if (!(ch instanceof SelChImpl))
        throw new IllegalSelectorException();
    //生成SelectorKey来存储到hashmap中，一共之后获取
    SelectionKeyImpl k = new SelectionKeyImpl((SelChImpl)ch, this);
    //attach用户想要存储的对象
    k.attach(attachment);
    //调用子类的implRegister方法
    synchronized (publicKeys) {
        implRegister(k);
    }
    //设置关注的option
    k.interestOps(ops);
    return k;
}

    protected void implRegister(SelectionKeyImpl ski) {
        if (closed)
            throw new ClosedSelectorException();
        SelChImpl ch = ski.channel;
        //获取Channel所对应的fd,因为在linux下socket会被当作一个文件，也会有fd
        int fd = Integer.valueOf(ch.getFDVal());
        fdToKey.put(fd, ski);
        //调用pollWrapper的add方法,将channel的fd添加到监控列表中
        pollWrapper.add(fd);
        //保存到HashSet中，keys是SelectorImpl的成员变量
        keys.add(ski);
    }

protected int doSelect(long timeout) throws IOException {
    .....
    try {
        ....
        //调用了poll方法,底层调用了native的epollCtl和epollWait方法
        pollWrapper.poll(timeout);
    } finally {
        ....
    }
    ....
    //更新selectedKeys,为之后的selectedKeys函数做准备
    int numKeysUpdated = updateSelectedKeys();
    ....
    return numKeysUpdated;
}

int poll(long timeout) throws IOException {
    updateRegistrations(); ////先调用epollCtl,更新关注的事件类型
    ////导致阻塞，等待事件产生
    updated = epollWait(pollArrayAddress, NUM_EPOLLEVENTS, timeout, epfd);
    .....
    return updated;
}

//是通过Util.ungrowableSet生成的,不能添加,只能减少
private Set<SelectionKey> publicSelectedKeys;
public Set<SelectionKey> selectedKeys() {
    ....
    return publicSelectedKeys;
}

private int updateSelectedKeys() {
    //更新了的keys的个数,或在说是产生的事件的个数
    int entries = pollWrapper.updated; 
    int numKeysUpdated = 0;
    for (int i=0; i<entries; i++) {
        //对应的channel的fd
        int nextFD = pollWrapper.getDescriptor(i);
        //通过fd找到对应的SelectionKey
        SelectionKeyImpl ski = fdToKey.get(Integer.valueOf(nextFD));
        if (ski != null) {
            int rOps = pollWrapper.getEventOps(i);
            //更新selectedKey变量,并通知响应的channel来做响应的处理
            if (selectedKeys.contains(ski)) {
                if (ski.channel.translateAndSetReadyOps(rOps, ski)) {
                    numKeysUpdated++;
                }
            } else {
                ski.channel.translateAndSetReadyOps(rOps, ski);
                if ((ski.nioReadyOps() & ski.nioInterestOps()) != 0) {
                    selectedKeys.add(ski);
                    numKeysUpdated++;
                }
            }
        }
    }
    return numKeysUpdated;
}