转载自：https://www.jianshu.com/p/0d497fe5484a

前言

Java NIO 由以下几个核心部分组成：
1、Buffer
2、Channel
3、Selector

Buffer和Channel在深入浅出NIO之Channel、Buffer一文中已经介绍过，本文主要讲解NIO的Selector实现原理。

之前进行socket编程时，accept方法会一直阻塞，直到有客户端请求的到来，并返回socket进行相应的处理。整个过程是流水线的，处理完一个请求，才能去获取并处理后面的请求，当然也可以把获取socket和处理socket的过程分开，一个线程负责accept，一个线程池负责处理请求。

但NIO提供了更好的解决方案，采用选择器（Selector）返回已经准备好的socket，并按顺序处理，基于通道（Channel）和缓冲区（Buffer）来进行数据的传输。

Selector

这里出来一个新概念，selector，具体是一个什么样的东西？

想想一个场景：在一个养鸡场，有这么一个人，每天的工作就是不停检查几个特殊的鸡笼，如果有鸡进来，有鸡出去，有鸡生蛋，有鸡生病等等，就把相应的情况记录下来，如果鸡场的负责人想知道情况，只需要询问那个人即可。

在这里，这个人就相当Selector，每个鸡笼相当于一个SocketChannel，每个线程通过一个Selector可以管理多个SocketChannel。

%title插图%num

为了实现Selector管理多个SocketChannel，必须将具体的SocketChannel对象注册到Selector，并声明需要监听的事件（这样Selector才知道需要记录什么数据），一共有4种事件：

1、connect：客户端连接服务端事件，对应值为SelectionKey.OP_CONNECT(8)
2、accept：服务端接收客户端连接事件，对应值为SelectionKey.OP_ACCEPT(16)
3、read：读事件，对应值为SelectionKey.OP_READ(1)
4、write：写事件，对应值为SelectionKey.OP_WRITE(4)

这个很好理解，每次请求到达服务器，都是从connect开始，connect成功后，服务端开始准备accept，准备就绪，开始读数据，并处理，*后写回数据返回。

所以，当SocketChannel有对应的事件发生时，Selector都可以观察到，并进行相应的处理。

服务端代码

为了更好的理解，先看一段服务端的示例代码

  1. ServerSocketChannel serverChannel = ServerSocketChannel.open();
  2. serverChannel.configureBlocking(false);
  3. serverChannel.socket().bind(new InetSocketAddress(port));
  4. Selector selector = Selector.open();
  5. serverChannel.register(selector, SelectionKey.OP_ACCEPT);
  6. while(true){
  7. int n = selector.select();
  8. if (n == 0) continue;
  9. Iterator ite = this.selector.selectedKeys().iterator();
  10. while(ite.hasNext()){
  11. SelectionKey key = (SelectionKey)ite.next();
  12. if (key.isAcceptable()){
  13. SocketChannel clntChan = ((ServerSocketChannel) key.channel()).accept();
  14. clntChan.configureBlocking(false);
  15. //将选择器注册到连接到的客户端信道，
  16. //并指定该信道key值的属性为OP_READ，
  17. //同时为该信道指定关联的附件
  18. clntChan.register(key.selector(), SelectionKey.OP_READ, ByteBuffer.allocate(bufSize));
  19. }
  20. if (key.isReadable()){
  21. handleRead(key);
  22. }
  23. if (key.isWritable() && key.isValid()){
  24. handleWrite(key);
  25. }
  26. if (key.isConnectable()){
  27. System.out.println(“isConnectable = true”);
  28. }
  29. ite.remove();
  30. }
  31. }

服务端操作过程

1、创建ServerSocketChannel实例，并绑定指定端口；
2、创建Selector实例；
3、将serverSocketChannel注册到selector，并指定事件OP_ACCEPT，*底层的socket通过channel和selector建立关联；
4、如果没有准备好的socket，select方法会被阻塞一段时间并返回0；
5、如果底层有socket已经准备好，selector的select方法会返回socket的个数，而且selectedKeys方法会返回socket对应的事件（connect、accept、read or write）；
6、根据事件类型，进行不同的处理逻辑；

在步骤3中，selector只注册了serverSocketChannel的OP_ACCEPT事件
1、如果有客户端A连接服务，执行select方法时，可以通过serverSocketChannel获取客户端A的socketChannel，并在selector上注册socketChannel的OP_READ事件。
2、如果客户端A发送数据，会触发read事件，这样下次轮询调用select方法时，就能通过socketChannel读取数据，同时在selector上注册该socketChannel的OP_WRITE事件，实现服务器往客户端写数据。

Selector实现原理

SocketChannel、ServerSocketChannel和Selector的实例初始化都通过SelectorProvider类实现，其中Selector是整个NIO Socket的核心实现。

  1. public static SelectorProvider provider() {
  2. synchronized (lock) {
  3. if (provider != null)
  4. return provider;
  5. return AccessController.doPrivileged(
  6. new PrivilegedAction<SelectorProvider>() {
  7. public SelectorProvider run() {
  8. if (loadProviderFromProperty())
  9. return provider;
  10. if (loadProviderAsService())
  11. return provider;
  12. provider = sun.nio.ch.DefaultSelectorProvider.create();
  13. return provider;
  14. }
  15. });
  16. }
  17. }

SelectorProvider在windows和linux下有不同的实现，provider方法会返回对应的实现。

这里不禁要问，Selector是如何做到同时管理多个socket？

下面我们看看Selector的具体实现，Selector初始化时，会实例化PollWrapper、SelectionKeyImpl数组和Pipe。

  1. WindowsSelectorImpl(SelectorProvider sp) throws IOException {
  2. super(sp);
  3. pollWrapper = new PollArrayWrapper(INIT_CAP);
  4. wakeupPipe = Pipe.open();
  5. wakeupSourceFd = ((SelChImpl)wakeupPipe.source()).getFDVal();
  7. // Disable the Nagle algorithm so that the wakeup is more immediate
  8. SinkChannelImpl sink = (SinkChannelImpl)wakeupPipe.sink();
  9. (sink.sc).socket().setTcpNoDelay(true);
  10. wakeupSinkFd = ((SelChImpl)sink).getFDVal();
  11. pollWrapper.addWakeupSocket(wakeupSourceFd, 0);
  12. }

pollWrapper用Unsafe类申请一块物理内存pollfd，存放socket句柄fdVal和events，其中pollfd共8位，0-3位保存socket句柄，4-7位保存events。

%title插图%num

%title插图%num

 

pollWrapper提供了fdVal和event数据的相应操作，如添加操作通过Unsafe的putInt和putShort实现。

  1. void putDescriptor(int i, int fd) {
  2. pollArray.putInt(SIZE_POLLFD * i + FD_OFFSET, fd);
  3. }
  4. void putEventOps(int i, int event) {
  5. pollArray.putShort(SIZE_POLLFD * i + EVENT_OFFSET, (short)event);
  6. }

先看看serverChannel.register(selector, SelectionKey.OP_ACCEPT)是如何实现的

  1. public final SelectionKey register(Selector sel, int ops, Object att)
  2. throws ClosedChannelException {
  3. synchronized (regLock) {
  4. SelectionKey k = findKey(sel);
  5. if (k != null) {
  6. k.interestOps(ops);
  7. k.attach(att);
  8. }
  9. if (k == null) {
  10. // New registration
  11. synchronized (keyLock) {
  12. if (!isOpen())
  13. throw new ClosedChannelException();
  14. k = ((AbstractSelector)sel).register(this, ops, att);
  15. addKey(k);
  16. }
  17. }
  18. return k;
  19. }
  20. }
  1. 如果该channel和selector已经注册过，则直接添加事件和附件。
  2. 否则通过selector实现注册过程。
  1. protected final SelectionKey register(AbstractSelectableChannel ch,
  2. int ops, Object attachment) {
  3. if (!(ch instanceof SelChImpl))
  4. throw new IllegalSelectorException();
  5. SelectionKeyImpl k = new SelectionKeyImpl((SelChImpl)ch, this);
  6. k.attach(attachment);
  7. synchronized (publicKeys) {
  8. implRegister(k);
  9. }
  10. k.interestOps(ops);
  11. return k;
  12. }
  14. protected void implRegister(SelectionKeyImpl ski) {
  15. synchronized (closeLock) {
  16. if (pollWrapper == null)
  17. throw new ClosedSelectorException();
  18. growIfNeeded();
  19. channelArray[totalChannels] = ski;
  20. ski.setIndex(totalChannels);
  21. fdMap.put(ski);
  22. keys.add(ski);
  23. pollWrapper.addEntry(totalChannels, ski);
  24. totalChannels++;
  25. }
  26. }

1、以当前channel和selector为参数，初始化SelectionKeyImpl 对象selectionKeyImpl ，并添加附件attachment。
2、如果当前channel的数量totalChannels等于SelectionKeyImpl数组大小，对SelectionKeyImpl数组和pollWrapper进行扩容操作。
3、如果totalChannels % MAX_SELECTABLE_FDS == 0，则多开一个线程处理selector。
4、pollWrapper.addEntry将把selectionKeyImpl中的socket句柄添加到对应的pollfd。
5、k.interestOps(ops)方法*终也会把event添加到对应的pollfd。

所以，不管serverSocketChannel，还是socketChannel，在selector注册的事件，*终都保存在pollArray中。

接着，再来看看selector中的select是如何实现一次获取多个有事件发生的channel的，底层由selector实现类的doSelect方法实现，如下：

  1. protected int doSelect(long timeout) throws IOException {
  2. if (channelArray == null)
  3. throw new ClosedSelectorException();
  4. this.timeout = timeout; // set selector timeout
  5. processDeregisterQueue();
  6. if (interruptTriggered) {
  7. resetWakeupSocket();
  8. return 0;
  9. }
  10. // Calculate number of helper threads needed for poll. If necessary
  11. // threads are created here and start waiting on startLock
  12. adjustThreadsCount();
  13. finishLock.reset(); // reset finishLock
  14. // Wakeup helper threads, waiting on startLock, so they start polling.
  15. // Redundant threads will exit here after wakeup.
  16. startLock.startThreads();
  17. // do polling in the main thread. Main thread is responsible for
  18. // first MAX_SELECTABLE_FDS entries in pollArray.
  19. try {
  20. begin();
  21. try {
  22. subSelector.poll();
  23. } catch (IOException e) {
  24. finishLock.setException(e); // Save this exception
  25. }
  26. // Main thread is out of poll(). Wakeup others and wait for them
  27. if (threads.size() > 0)
  28. finishLock.waitForHelperThreads();
  29. } finally {
  30. end();
  31. }
  32. // Done with poll(). Set wakeupSocket to nonsignaled for the next run.
  33. finishLock.checkForException();
  34. processDeregisterQueue();
  35. int updated = updateSelectedKeys();
  36. // Done with poll(). Set wakeupSocket to nonsignaled for the next run.
  37. resetWakeupSocket();
  38. return updated;
  39. }

其中 subSelector.poll() 是select的核心，由native函数poll0实现，readFds、writeFds 和exceptFds数组用来保存底层select的结果，数组的*个位置都是存放发生事件的socket的总数，其余位置存放发生事件的socket句柄fd。

  1. private final int[] readFds = new int [MAX_SELECTABLE_FDS + 1];
  2. private final int[] writeFds = new int [MAX_SELECTABLE_FDS + 1];
  3. private final int[] exceptFds = new int [MAX_SELECTABLE_FDS + 1];
  4. private int poll() throws IOException{ // poll for the main thread
  5. return poll0(pollWrapper.pollArrayAddress,
  6. Math.min(totalChannels, MAX_SELECTABLE_FDS),
  7. readFds, writeFds, exceptFds, timeout);
  8. }

执行 selector.select() ，poll0函数把指向socket句柄和事件的内存地址传给底层函数。
1、如果之前没有发生事件，程序就阻塞在select处，当然不会一直阻塞，因为epoll在timeout时间内如果没有事件，也会返回；
2、一旦有对应的事件发生，poll0方法就会返回；
3、processDeregisterQueue方法会清理那些已经cancelled的SelectionKey；
4、updateSelectedKeys方法统计有事件发生的SelectionKey数量，并把符合条件发生事件的SelectionKey添加到selectedKeys哈希表中，提供给后续使用。

在早期的JDK1.4和1.5 update10版本之前，Selector基于select/poll模型实现，是基于IO复用技术的非阻塞IO，不是异步IO。在JDK1.5 update10和linux core2.6以上版本，sun优化了Selctor的实现，底层使用epoll替换了select/poll。

read实现

通过遍历selector中的SelectionKeyImpl数组，获取发生事件的socketChannel对象，其中保存了对应的socket，实现如下

  1. public int read(ByteBuffer buf) throws IOException {
  2. if (buf == null)
  3. throw new NullPointerException();
  4. synchronized (readLock) {
  5. if (!ensureReadOpen())
  6. return –1;
  7. int n = 0;
  8. try {
  9. begin();
  10. synchronized (stateLock) {
  11. if (!isOpen()) {
  12. return 0;
  13. }
  14. readerThread = NativeThread.current();
  15. }
  16. for (;;) {
  17. n = IOUtil.read(fd, buf, –1, nd);
  18. if ((n == IOStatus.INTERRUPTED) && isOpen()) {
  19. // The system call was interrupted but the channel
  20. // is still open, so retry
  21. continue;
  22. }
  23. return IOStatus.normalize(n);
  24. }
  25. } finally {
  26. readerCleanup(); // Clear reader thread
  27. // The end method, which
  28. end(n > 0 || (n == IOStatus.UNAVAILABLE));
  30. // Extra case for socket channels: Asynchronous shutdown
  31. //
  32. synchronized (stateLock) {
  33. if ((n <= 0) && (!isInputOpen))
  34. return IOStatus.EOF;
  35. }
  36. assert IOStatus.check(n);
  37. }
  38. }
  39. }

*终通过Buffer的方式读取socket的数据。

wakeup实现

  1. public Selector wakeup() {
  2. synchronized (interruptLock) {
  3. if (!interruptTriggered) {
  4. setWakeupSocket();
  5. interruptTriggered = true;
  6. }
  7. }
  8. return this;
  9. }
  11. // Sets Windows wakeup socket to a signaled state.
  12. private void setWakeupSocket() {
  13. setWakeupSocket0(wakeupSinkFd);
  14. }
  15. private native void setWakeupSocket0(int wakeupSinkFd);

看来wakeupSinkFd这个变量是为wakeup方法使用的。
其中interruptTriggered为中断已触发标志，当pollWrapper.interrupt()之后，该标志即为true了；因为这个标志，连续两次wakeup，只会有一次效果。

epoll原理

epoll是Linux下的一种IO多路复用技术，可以非常高效的处理数以百万计的socket句柄。

三个epoll相关的系统调用：

  • int epoll_create(int size)
    epoll_create建立一个epoll对象。参数size是内核保证能够正确处理的*大句柄数，多于这个*大数时内核可不保证效果。
  • int epoll_ctl(int epfd, int op, int fd, struct epoll_event event)
    epoll_ctl可以操作epoll_create创建的epoll，如将socket句柄加入到epoll中让其监控，或把epoll正在监控的某个socket句柄移出epoll。
  • int epoll_wait(int epfd, struct epoll_event events,int maxevents, int timeout)
    epoll_wait在调用时，在给定的timeout时间内，所监控的句柄中有事件发生时，就返回用户态的进程。

epoll内部实现大概如下：

  1. epoll初始化时，会向内核注册一个文件系统，用于存储被监控的句柄文件，调用epoll_create时，会在这个文件系统中创建一个file节点。同时epoll会开辟自己的内核高速缓存区，以红黑树的结构保存句柄，以支持快速的查找、插入、删除。还会再建立一个list链表，用于存储准备就绪的事件。
  2. 当执行epoll_ctl时，除了把socket句柄放到epoll文件系统里file对象对应的红黑树上之外，还会给内核中断处理程序注册一个回调函数，告诉内核，如果这个句柄的中断到了，就把它放到准备就绪list链表里。所以，当一个socket上有数据到了，内核在把网卡上的数据copy到内核中后，就把socket插入到就绪链表里。
  3. 当epoll_wait调用时，仅仅观察就绪链表里有没有数据，如果有数据就返回，否则就sleep，超时时立刻返回。