Java NIO Selector解析

用法

Selector的主要方法有三个：

Selector.open():Selector
主要用于创建一个selector
select():int
该方法有两种模式：在阻塞模式下，该方法会一直阻塞，直到超时或者有注册的事件发生，返回发生事件的数量；在非阻塞模式下，会立即返回，如果没有事件发生，则返回0。
selectedKeys():Set
获取已经就绪的事件的key的集合。

Selector.open()原理

通过阅读java.nio.channels.Selector的源码（如下），可以发现，其调用了SelectorProvider.provider()的openSelector()函数。

1
2
3

public static Selector More ...open() throws IOException {
    return SelectorProvider.provider().openSelector();
}

先看SelectorProvider.provider()，在这个函数中，保证了provider有且仅有一个。

public static SelectorProvider provider() {
    synchronized (lock) {
        if (provider != null)   //保证只有一个
            return provider;
        return AccessController.doPrivileged(
            new PrivilegedAction<SelectorProvider>() {
                public SelectorProvider run() {
                    if (loadProviderFromProperty())
                        return provider;
                    if (loadProviderAsService())
                        return provider;
                    provider = sun.nio.ch.DefaultSelectorProvider.create();
                    return provider;
                }
            });
    }
}

再看create()函数，在该函数中，区分了操作系统，在Windows下，使用的是sun.nio.ch.PollSelectorProvider类，作为Selector。也就是说，SelectorProvider.provider()返回了一个PollSelectorProvider对象。

public static SelectorProvider create() {
    String osname = AccessController
        .doPrivileged(new GetPropertyAction("os.name"));
    if (osname.equals("SunOS"))
        return createProvider("sun.nio.ch.DevPollSelectorProvider");
    if (osname.equals("Linux"))
        return createProvider("sun.nio.ch.EPollSelectorProvider");
    return new sun.nio.ch.PollSelectorProvider();
}

在sun.nio.ch.PollSelectorProvider的openSelector()函数（如下）中，返回了一个PollSelectorImpl的对象。

public class PollSelectorProvider
    extends SelectorProviderImpl
{
    public AbstractSelector openSelector() throws IOException {
        return new PollSelectorImpl(this);
    }

    public Channel inheritedChannel() throws IOException {
        return InheritedChannel.getChannel();
    }
}

在PollSelectorImpl的构造器（如下）中，我发现，其创建了一个pipe，创建函数是IOUtil.makePipe()，但是这个函数是native的，我估计应该是OS提供的一个函数。这个pipe的主要作用是，在select()函数被阻塞的时候，用于唤醒线程使用。

需要注意的是，虽然函数名称叫做makePipe，但是在Windows平台下实现中，并没有使用pipe，而是使用的TCP连接。

同时还要注意super函数的参数，第一个1是channels总数，第二个1是偏移量，说明，在初始化过程中，就已经有一个channel在这个数组中了，channels[0]就是刚刚创建的pipe。这个channel就是用来唤醒select()的。

PollSelectorImpl(SelectorProvider sp) {
    super(sp, 1, 1);
    long pipeFds = IOUtil.makePipe(false);  //创建pipe
    fd0 = (int) (pipeFds >>> 32);
    fd1 = (int) pipeFds;
    pollWrapper = new PollArrayWrapper(INIT_CAP);
    pollWrapper.initInterrupt(fd0, fd1);
    channelArray = new SelectionKeyImpl[INIT_CAP];
}

综上所述，Selector.open()函数的作用是：保证了provider有且仅有一个，并返回了一个PollSelectorImpl对象，同时创建了一条pipe。

select()原理

select的执行体是在sun.nio.ch.PollSelectorImpl中的doSelect函数（如下）。

protected int doSelect(long timeout)
    throws IOException
{
    if (channelArray == null)
        throw new ClosedSelectorException();
    processDeregisterQueue();
    try {
        begin();
        pollWrapper.poll(totalChannels, 0, timeout);    //在这里轮询
    } finally {
        end();
    }
    processDeregisterQueue();
    int numKeysUpdated = updateSelectedKeys();
    if (pollWrapper.getReventOps(0) != 0) {
        // Clear the wakeup pipe
        pollWrapper.putReventOps(0, 0);
        synchronized (interruptLock) {
            IOUtil.drain(fd0);
            interruptTriggered = false;
        }
    }
    return numKeysUpdated;
}

updateSelectedKeys()函数（如下）的实现是在sun.nio.ch.AbstractPollSelectorImpl中。
通过getReventOps()函数来判断通道是否有事件发生，若发生，将其加入到selectedKeys中。

/**
 * Copy the information in the pollfd structs into the opss
 * of the corresponding Channels. Add the ready keys to the
 * ready queue.
 */
protected int updateSelectedKeys() {
    int numKeysUpdated = 0;
    // Skip zeroth entry; it is for interrupts only
    // 第0个是自带的，用于唤醒select()的，因此从1开始循环。
    for (int i=channelOffset; i<totalChannels; i++) {
        int rOps = pollWrapper.getReventOps(i);
        if (rOps != 0) {
            SelectionKeyImpl sk = channelArray[i];
            pollWrapper.putReventOps(i, 0);
            if (selectedKeys.contains(sk)) {
                if (sk.channel.translateAndSetReadyOps(rOps, sk)) {
                    numKeysUpdated++;
                }
            } else {
                sk.channel.translateAndSetReadyOps(rOps, sk);
                if ((sk.nioReadyOps() & sk.nioInterestOps()) != 0) {
                    selectedKeys.add(sk);
                    numKeysUpdated++;
                }
            }
        }
    }
    return numKeysUpdated;
}

selectedKeys()原理

该方法直接返回publicSelectedKeys这个Set。

public Set<SelectionKey> selectedKeys() {
    if (!isOpen() && !Util.atBugLevel("1.4"))
        throw new ClosedSelectorException();
    return publicSelectedKeys;
}

publicSelectedKeys在SelectorImpl中的初始化中，与selectedKeys做了绑定，并在updateSelectedKeys()函数中，进行更新。

protected SelectorImpl(SelectorProvider sp) {
    super(sp);
    keys = new HashSet<SelectionKey>();
    selectedKeys = new HashSet<SelectionKey>();
    if (Util.atBugLevel("1.4")) {
        publicKeys = keys;
        publicSelectedKeys = selectedKeys;
    } else {
        publicKeys = Collections.unmodifiableSet(keys);
        publicSelectedKeys = Util.ungrowableSet(selectedKeys);
    }
}

Hello, world!