Java NIO选择器【源码笔记】

一、I/O复用模型回顾
二、创建选择器
    1.Pipe函数
    2.Poll函数
三、注册Channel到Selector
四、Select返回就绪事件集合
五、本文总结六、系列文章

Selector selector = Selector.open();

PollSelectorImpl(SelectorProvider sp) {
    super(sp, 1, 1);
    // Native调用pipe()函数返回两个文件描述符
    long pipeFds = IOUtil.makePipe(false); // @1
    // 文件描述符fd0只读数据
    fd0 = (int) (pipeFds >>> 32);
    // 文件描述符fd1只写数据
    fd1 = (int) pipeFds;
    // pollfd结构类型数组
    pollWrapper = new PollArrayWrapper(INIT_CAP); // @2
    pollWrapper.initInterrupt(fd0, fd1);
    channelArray = new SelectionKeyImpl[INIT_CAP];
}

long pipeFds = IOUtil.makePipe(false);
staticnativelongmakePipe(boolean blocking);
//IOUtil.c#Java_sun_nio_ch_IOUtil_makePipe()
pipe(fd)

intpipe(int fildes[2]);

The pipe() function creates a data pipe and places two file descriptors, one each into the arguments fildes[0] and fildes[1], that refer to the open file descriptions for the read and write ends of the pipe, respectively. Their integer values will be the two lowest available at the time of the pipe() call. The O_NONBLOCK and FD_CLOEXEC flags will be clear on both descriptors. NOTE: these flags can, however, be set by the fcntl() function.

Data can be written to the file descriptor fildes[1] and read from file descriptor fildes[0]. A read on the file descriptor fildes[0] will access data written to the file descriptor fildes[1] on a first-in-first-out basis. File descriptor fildes[0] is open for reading only. File descriptor fildes[1] is open for writing only.

intpoll(struct pollfd fds[],
          nfds_t nfds,
          int timeout)

The poll() function is used to enable an application to multiplex I/O over a set of descriptors. For each member of the array pointed to by fds, poll() will examine the given descriptor for the event(s) specified. nfds is the number of pollfd structures in the fds array. The poll() function will determine which descriptors can read or write data, or whether certain events have occured on the descriptors within the timeout period.

 structpollfd {
     int    fd;         /* Descriptor        */
     short  events;     /* Requested events  */
     short  revents;    /* Returned events   */
  };

pollfd 结构体数组元素个数。

发生错误时，poll函数的返回值为-1；若定时器到时没有任何描述符就绪，则返回0；否则返回就绪描述符的个数，即：revents成员值非0的描述符个数。

// 示例代码
serverSocket.bind(new InetSocketAddress("localhost",8089));
serverChannel.configureBlocking(false);
serverChannel.register(selector, SelectionKey.OP_ACCEPT);
// JDK代码
publicfinal SelectionKey register(Selector sel, int ops,
                                       Object att)
    throws ClosedChannelException{
    synchronized (regLock) {
        // 通道需开启状态
        if (!isOpen())
            throw new ClosedChannelException();
        // 操作合法性校验
        if ((ops & ~validOps()) != 0)
            throw new IllegalArgumentException();
        // 通道需非阻塞
        if (blocking)
            throw new IllegalBlockingModeException();
        SelectionKey k = findKey(sel);
        if (k != null) {
            // 注册操作并转换Poll Native支持的事件
            k.interestOps(ops);
            k.attach(att);
        }
        if (k == null) {
            // New registration
            synchronized (keyLock) {
                if (!isOpen())
                    throw new ClosedChannelException();
                // 注册channel到selector并返回选择SelectionKey
                k = ((AbstractSelector)sel).register(this, ops, att);
                addKey(k);
            }
        }
        return k;
    }
}

// 示例代码
int n = selector.select();
// JDK代码
protectedintdoSelect(long timeout)
        throws IOException {
  ...
  pollWrapper.poll(totalChannels, 0, timeout);
  ...
  int numKeysUpdated = updateSelectedKeys();
  ...
}
// JDK代码
protectedintupdateSelectedKeys() {
int numKeysUpdated = 0;
for (int i=channelOffset; i<totalChannels; i++) {
    // 获取通道就绪操作类型（可读、可写、错误等）
    int rOps = pollWrapper.getReventOps(i);
    ...
    // 将ReventOps就绪的操作类型转换到SelectionKeyImpl
    if (sk.channel.translateAndSetReadyOps(rOps, sk)) {
        numKeysUpdated++;
    }
    ...
}
return numKeysUpdated;
}