nio Selector 阻塞 唤醒 原理

Selector：java nio无阻塞io实现的关键。

阻塞io和无阻塞io：
阻塞io是指jdk1.4之前版本面向流的io，服务端需要对每个请求建立一堆线程等待请求，而客户端发送请求后，先咨询服务端是否有线程相应，如果没有则会一直等待或者遭到拒 绝请求，如果有的话，客户端会线程会等待请求结束后才继续执行。
当并发量大，而后端服务或客户端处理数据慢时就会产生产生大量线程处于等待中，即上述的阻塞。

 

无阻塞io是使用单线程或者只使用少量的多线程，每个连接共用一个线程，当处于等待（没有事件）的时候线程资源可以释放出来处理别的请求，通过事件驱动模型当有accept/read/write等事件发生后通知（唤醒）主线程分配资源来处理相关事件。java.nio.channels.Selector就是在该模型中事件的观察者，可以将多个SocketChannel的事件注册到一个Selector上，当没有事件发生时Selector处于阻塞状态，当SocketChannel有accept/read/write等事件发生时唤醒Selector。
 

 这个Selector是使用了单线程模型，主要用来描述事件驱动模型，要优化性能需要一个好的线程模型来使用，目前比较好的nio框架有Netty，apache的mina等。线程模型这块后面再分享，这里重点研究Selector的阻塞和唤醒原理。

 

先看一段简单的Selector使用的代码

 

selector = Selector.open();

ServerSocketChannel ssc = ServerSocketChannel.open();
ssc.configureBlocking(false);
ssc.socket().bind(new InetSocketAddress(port));

ssc.register(selector, SelectionKey.OP_ACCEPT);

while (true) {

    // select()阻塞，等待有事件发生唤醒
    int selected = selector.select();

    if (selected > 0) {
        Iterator<SelectionKey> selectedKeys = selector.selectedKeys().iterator();
        while (selectedKeys.hasNext()) {
            SelectionKey key = selectedKeys.next();
            if ((key.readyOps() & SelectionKey.OP_ACCEPT) == SelectionKey.OP_ACCEPT) {
                // 处理 accept 事件
            } else if ((key.readyOps() & SelectionKey.OP_READ) == SelectionKey.OP_READ) {
                // 处理 read 事件
            } else if ((key.readyOps() & SelectionKey.OP_WRITE) == SelectionKey.OP_WRITE) {
                // 处理 write 事件
            }
            selectedKeys.remove();
        }
    }
}

代码中关键的几个点在：
Selector.open();
selector.select();
阻塞后唤醒可以通过注册在selector上的socket有事件发生 或者 selector.select(timeOut)超时 或者 selector.wakeup()主动唤醒；

整个阻塞和唤醒的过程涉及到的点非常多，先上一张梳理出的整体图，再进入源码会比较容易理解

 

 

 现在通过openjdk中的源码来解析上图中的每一个环节：

1. Selector.open()

 

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

 先看看SelectorProvider.provider()做了什么:

 

 

SelectorProvider.java
-----
public static SelectorProvider provider() {
synchronized (lock) {
    if (provider != null)
    return provider;
    return (SelectorProvider)AccessController
    .doPrivileged(new PrivilegedAction() {
        public Object run() {
            if (loadProviderFromProperty())
            return provider;
            if (loadProviderAsService())
            return provider;
            provider = sun.nio.ch.DefaultSelectorProvider.create();
            return provider;
        }
        });
}
}

 其中provider = sun.nio.ch.DefaultSelectorProvider.create();会根据操作系统来返回不同的实现类，windows平台就返回WindowsSelectorProvider；
这里主要以windows的实现来梳理整个流程，拿到provider后来看openSelector()中的实现

 

 

WindowsSelectorProvider.java
----
public AbstractSelector openSelector() throws IOException {
    return new WindowsSelectorImpl(this);
}

WindowsSelectorImpl.java
----
WindowsSelectorImpl(SelectorProvider sp) throws IOException {
    super(sp);
    pollWrapper = new PollArrayWrapper(INIT_CAP);
    wakeupPipe = Pipe.open();
    wakeupSourceFd = ((SelChImpl)wakeupPipe.source()).getFDVal();

    // Disable the Nagle algorithm so that the wakeup is more immediate
    SinkChannelImpl sink = (SinkChannelImpl)wakeupPipe.sink();
    (sink.sc).socket().setTcpNoDelay(true);
    wakeupSinkFd = ((SelChImpl)sink).getFDVal();

    pollWrapper.addWakeupSocket(wakeupSourceFd, 0);
}

 这段代码中做了如下几个事情
Pipe.open()打开一个管道（打开管道的实现后面再看）；拿到wakeupSourceFd和wakeupSinkFd两个文件描述符；把唤醒端的文件描述符（wakeupSourceFd）放到pollWrapper里；
那么为什么需要一个管道，这个管道是怎么实现的？接下来看Pipe.open()做了什么

 

 

Pipe.java
----
public static Pipe open() throws IOException {
return SelectorProvider.provider().openPipe();
}

 同样，SelectorProvider.provider()也是获取操作系统相关的实现

 

 

SelectorProvider.java
----
public Pipe openPipe() throws IOException {
    return new PipeImpl(this);
}

 这里还是看windows下的实现

 

 

PipeImpl.java
----
PipeImpl(final SelectorProvider sp) throws IOException {
    try {
        AccessController.doPrivileged(new Initializer(sp));
    } catch (PrivilegedActionException x) {
        throw (IOException)x.getCause();
    }
}

 创建了一个PipeImpl对象， AccessController.doPrivileged调用后紧接着会执行initializer的run方法

 

 

PipeImpl.Initializer
-----
public Object run() throws IOException {
    ServerSocketChannel ssc = null;
    SocketChannel sc1 = null;
    SocketChannel sc2 = null;

    try {
        // loopback address
        InetAddress lb = InetAddress.getByName("127.0.0.1");
        assert(lb.isLoopbackAddress());

        // bind ServerSocketChannel to a port on the loopback address
        ssc = ServerSocketChannel.open();
        ssc.socket().bind(new InetSocketAddress(lb, 0));

        // Establish connection (assumes connections are eagerly
        // accepted)
        InetSocketAddress sa
            = new InetSocketAddress(lb, ssc.socket().getLocalPort());
        sc1 = SocketChannel.open(sa);

        ByteBuffer bb = ByteBuffer.allocate(8);
        long secret = rnd.nextLong();
        bb.putLong(secret).flip();
        sc1.write(bb);

        // Get a connection and verify it is legitimate
        for (;;) {
            sc2 = ssc.accept();
            bb.clear();
            sc2.read(bb);
            bb.rewind();
            if (bb.getLong() == secret)
                break;
            sc2.close();
        }

        // Create source and sink channels
        source = new SourceChannelImpl(sp, sc1);
        sink = new SinkChannelImpl(sp, sc2);
    } catch (IOException e) {
        try {
            if (sc1 != null)
                sc1.close();
            if (sc2 != null)
                sc2.close();
        } catch (IOException e2) { }
        IOException x = new IOException("Unable to establish"
                                        + " loopback connection");
        x.initCause(e);
        throw x;
    } finally {
        try {
            if (ssc != null)
                ssc.close();
        } catch (IOException e2) { }
    }
    return null;
}

 这里即为上图中最下面那部分创建pipe的过程，windows下的实现是创建两个本地的socketChannel,然后连接（链接的过程通过写一个随机long做两个socket的链接校验），两个socketChannel分别实现了管道的source与sink端。
source端由前面提到的WindowsSelectorImpl放到了pollWrapper中（pollWrapper.addWakeupSocket(wakeupSourceFd, 0)）

 

 

PollArrayWrapper.java
----
private AllocatedNativeObject pollArray; // The fd array

// Adds Windows wakeup socket at a given index.
void addWakeupSocket(int fdVal, int index) {
    putDescriptor(index, fdVal);
    putEventOps(index, POLLIN);
}

// Access methods for fd structures
void putDescriptor(int i, int fd) {
    pollArray.putInt(SIZE_POLLFD * i + FD_OFFSET, fd);
}

void putEventOps(int i, int event) {
    pollArray.putShort(SIZE_POLLFD * i + EVENT_OFFSET, (short)event);
}

 这里将source的POLLIN事件标识为感兴趣的，当sink端有数据写入时，source对应的文件描述符wakeupSourceFd就会处于就绪状态

 

 

AllocatedNativeObject.java
----
class AllocatedNativeObject extends NativeObject

AllocatedNativeObject(int size, boolean pageAligned) {
    super(size, pageAligned);
}

NativeObject.java
----
protected NativeObject(int size, boolean pageAligned) {
    if (!pageAligned) {
        this.allocationAddress = unsafe.allocateMemory(size);
        this.address = this.allocationAddress;
    } else {
        int ps = pageSize();
        long a = unsafe.allocateMemory(size + ps);
        this.allocationAddress = a;
        this.address = a + ps - (a & (ps - 1));
    }
}

 从以上可以看到pollArray是通过unsafe.allocateMemory(size + ps)分配的一块系统内存

到这里完成了Selector.open()，主要完成建立Pipe，并把pipe的wakeupSourceFd放入pollArray中,这个pollArray是Selector的枢纽。这里是以Windows的实现来看，在windows下通过两个链接的socketChannel实现了Pipe，linux下则是直接使用系统的pipe。

 

2. serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

 

AbstractSelectableChannel.java --> register() --> SelectorImpl.java
----
protected final SelectionKey register(AbstractSelectableChannel ch,int ops,Object attachment)
{
    if (!(ch instanceof SelChImpl))
        throw new IllegalSelectorException();
    SelectionKeyImpl k = new SelectionKeyImpl((SelChImpl)ch, this);
    k.attach(attachment);
    synchronized (publicKeys) {
        implRegister(k);
    }
    k.interestOps(ops);
    return k;
}

 关键是implRegister(k);

 

 

WindowsSelectorImpl.java
----
protected void implRegister(SelectionKeyImpl ski) {
    growIfNeeded();
    channelArray[totalChannels] = ski;
    ski.setIndex(totalChannels);
    fdMap.put(ski);
    keys.add(ski);
    pollWrapper.addEntry(totalChannels, ski);
    totalChannels++;
}

PollArrayWrapper.java
----
void addEntry(int index, SelectionKeyImpl ski) {
    putDescriptor(index, ski.channel.getFDVal());
}

 这里把socketChannel的文件描述符放到pollArray中。

 

 

3. selector.select();

 

SelectorImpl.java
----
public int select(long timeout) throws IOException
{
    if (timeout < 0)
        throw new IllegalArgumentException("Negative timeout");
    return lockAndDoSelect((timeout == 0) ? -1 : timeout);
}

private int lockAndDoSelect(long timeout) throws IOException {
    synchronized (this) {
        if (!isOpen())
            throw new ClosedSelectorException();
        synchronized (publicKeys) {
            synchronized (publicSelectedKeys) {
                return doSelect(timeout);
            }
        }
    }
}

 其中的doSelector又回到我们的Windows实现：

 

 

WindowsSelectorImpl.java
----
protected int doSelect(long timeout) throws IOException {
    if (channelArray == null)
        throw new ClosedSelectorException();
    this.timeout = timeout; // set selector timeout
    processDeregisterQueue();
    if (interruptTriggered) {
        resetWakeupSocket();
        return 0;
    }
    // Calculate number of helper threads needed for poll. If necessary
    // threads are created here and start waiting on startLock
    adjustThreadsCount();
    finishLock.reset(); // reset finishLock
    // Wakeup helper threads, waiting on startLock, so they start polling.
    // Redundant threads will exit here after wakeup.
    startLock.startThreads();
    // do polling in the main thread. Main thread is responsible for
    // first MAX_SELECTABLE_FDS entries in pollArray.
    try {
        begin();
        try {
            subSelector.poll();
        } catch (IOException e) {
            finishLock.setException(e); // Save this exception
        }
        // Main thread is out of poll(). Wakeup others and wait for them
        if (threads.size() > 0)
            finishLock.waitForHelperThreads();
      } finally {
          end();
      }
    // Done with poll(). Set wakeupSocket to nonsignaled  for the next run.
    finishLock.checkForException();
    processDeregisterQueue();
    int updated = updateSelectedKeys();
    // Done with poll(). Set wakeupSocket to nonsignaled  for the next run.
    resetWakeupSocket();
    return updated;
}

private int poll() throws IOException{ // poll for the main thread
    return poll0(pollWrapper.pollArrayAddress,
                 Math.min(totalChannels, MAX_SELECTABLE_FDS),
                 readFds, writeFds, exceptFds, timeout);
}

 private native int poll0(long pollAddress, int numfds, int[] readFds, int[] writeFds, int[] exceptFds, long timeout);
其他的都是一些准备工作，关键是subSelector.poll()，最后调用了native的poll0，并把pollWrapper.pollArrayAddress作为参数传给poll0，那么poll0对pollArray做了什么：

 

 

WindowsSelectorImpl.c
----
Java_sun_nio_ch_WindowsSelectorImpl_00024SubSelector_poll0(JNIEnv *env, jobject this,
                                   jlong pollAddress, jint numfds,
                                   jintArray returnReadFds, jintArray returnWriteFds,
                                   jintArray returnExceptFds, jlong timeout)
{

    // 代码.... 此处省略一万字

    /* Call select */
    if ((result = select(0 , &readfds, &writefds, &exceptfds, tv)) == SOCKET_ERROR) {

        // 代码.... 此处省略一万字

        for (i = 0; i < numfds; i++) {
            // 代码.... 此处省略一万字
        }
    }
}

 C代码已经忘得差不多了，但这里可以看到实现思路是调用c的select方法，这里的select对应于内核中的sys_select调用，sys_select首先将第二三四个参数指向的fd_set拷贝到内核，然后对每个被SET的描述符调用进行poll，并记录在临时结果中（fdset），如果有事件发生，select会将临时结果写到用户空间并返回；当轮询一遍后没有任何事件发生时，如果指定了超时时间，则select会睡眠到超时，睡眠结束后再进行一次轮询，并将临时结果写到用户空间，然后返回。
这里的select就是轮询pollArray中的FD，看有没有事件发生，如果有事件发生收集所有发生事件的FD，退出阻塞。
关于select系统调用参考了《select、poll、epoll的比较》这篇文章，同时看到nio的select在不同平台上的实现不同，在linux上通过epoll可以不用轮询，在第一次调用后，事件信息就会与对应的epoll描述符关联起来，待的描述符上注册回调函数，当事件发生时，回调函数负责把发生的事件存储在就绪事件链表中，最后写到用户空间。

到这里已经比较清楚了，退出阻塞的方式有：regist在selector上的socketChannel处于就绪状态（放在pollArray中的socketChannel的FD就绪） 或者 第1节中放在pollArray中的wakeupSourceFd就绪。前者（socketChannel）就绪唤醒应证了文章开始的阻塞->事件驱动->唤醒的过程，后者（wakeupSourceFd）就是下面要看的主动wakeup。

 

 

4. selector.wakeup()

 

WindowsSelectorImpl.java
----
public Selector wakeup() {
    synchronized (interruptLock) {
        if (!interruptTriggered) {
            setWakeupSocket();
            interruptTriggered = true;
        }
    }
    return this;
}

// Sets Windows wakeup socket to a signaled state.
private void setWakeupSocket() {
    setWakeupSocket0(wakeupSinkFd);
}

private native void setWakeupSocket0(int wakeupSinkFd);

 native实现摘要：

 

WindowsSelectorImpl.c
----
Java_sun_nio_ch_WindowsSelectorImpl_setWakeupSocket0(JNIEnv *env, jclass this,
                                                jint scoutFd)
{
    /* Write one byte into the pipe */
    send(scoutFd, (char*)&POLLIN, 1, 0);
}

 这里完成了向最开始建立的pipe的sink端写入了一个字节，source文件描述符就会处于就绪状态，poll方法会返回，从而导致select方法返回。（原来自己建立一个socket链着自己另外一个socket就是为了干这事）

到此通过源码分析理清了Selector阻塞唤醒原理

 

参考学习文章：

http://www.cnblogs.com/xuxm2007/archive/2011/08/15/2139809.html

http://blog.chinaunix.net/uid-20543672-id-3267385.html

http://goon.iteye.com/blog/1775421

http://blog.csdn.net/aesop_wubo/article/details/9117655