深度解析Java NIO底层实现原理（selector是怎样select的）

Love The Way You Lie 2022-10-19 00:58 319阅读 0赞

> 原创不易，转载请注明出处

### 文章目录 ###

*   *   *  前言
         *  selector\#select源码剖析
         *  总结

--------------------

### 前言 ###

在前几篇《深度解析Java NIO底层实现原理》系列文章中，我们深度剖析了serverSocketChannel的open与bind底层原理，selector的open底层原理和serverSocketChannel注册到selector的底层原理，本文接着往下剖析，就会到`int select = selector.select();`这行代码了，接下来我们看看是怎样select的。

### selector\#select源码剖析 ###

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

可以看到是走的select(timeout)的方法，然后调用了lockAndDoSelect方法。

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

可以看到上了一堆锁，最终调用的是doSelect方法。

protected abstract int doSelect(long timeout) throws IOException;

可以看到这个doSelect是抽象方法，由子类实现，这里也就是EPollSelectorImpl这个子类

protected int doSelect(long timeout) throws IOException { 
            if (closed)
                throw new ClosedSelectorException();
            processDeregisterQueue();
            try { 
                begin();
                pollWrapper.poll(timeout);
            } finally { 
                end();
            }
            processDeregisterQueue();
    
            // 更新selectedKey
            int numKeysUpdated = updateSelectedKeys();
            if (pollWrapper.interrupted()) { 
                // Clear the wakeup pipe
                pollWrapper.putEventOps(pollWrapper.interruptedIndex(), 0);
                synchronized (interruptLock) { 
                    pollWrapper.clearInterrupted();
                    IOUtil.drain(fd0);
                    interruptTriggered = false;
                }
            }
            return numKeysUpdated;
        }

这里先调用processDeregisterQueue方法，处理了一下取消注册的那些fd。我们这里稍微看下是怎样处理取消注册的fd的

void processDeregisterQueue() throws IOException { 
        // Precondition: Synchronized on this, keys, and selectedKeys
        Set<SelectionKey> cks = cancelledKeys();
        synchronized (cks) { 
            if (!cks.isEmpty()) { 
                Iterator<SelectionKey> i = cks.iterator();
                while (i.hasNext()) { 
                    SelectionKeyImpl ski = (SelectionKeyImpl)i.next();
                    try { 
                        implDereg(ski);
                    } catch (SocketException se) { 
                        throw new IOException("Error deregistering key", se);
                    } finally { 
                        i.remove();
                    }
                }
            }
        }
    }

其实就是遍历cancelledKeys 这个集合里面的数据。然后调用implDereg方法进行处理  
implDereg也在EPollSelectorImpl类中

protected void implDereg(SelectionKeyImpl ski) throws IOException { 
            assert (ski.getIndex() >= 0);
            SelChImpl ch = ski.channel;
            int fd = ch.getFDVal();
    
            //删除 fd 与 selectionKey 的对应关系
            fdToKey.remove(Integer.valueOf(fd));
    
            // 调用epoll包装类移除fd
            pollWrapper.remove(fd);
            ski.setIndex(-1);
            keys.remove(ski);
            selectedKeys.remove(ski);
            deregister((AbstractSelectionKey)ski);
            SelectableChannel selch = ski.channel();
            if (!selch.isOpen() && !selch.isRegistered())
                ((SelChImpl)selch).kill();
        }

可以看到，就是删除一些数据结构里面维护的一些注册信息。  
我们这里再回到doSelect 方法中begin();这个方法我们就不看了，它其实就是设置了一些关于中断的东西。  
直接看下`pollWrapper.poll(timeout);`这行代码，**这个是重点**

int poll(long timeout) throws IOException { 
            // 处理一下更新过的fd，就是调用epoll的epoll_ctl
            updateRegistrations();
    
            // 进行epoll_wait
            updated = epollWait(pollArrayAddress, NUM_EPOLLEVENTS, timeout, epfd);
            for (int i=0; i<updated; i++) { 
                if (getDescriptor(i) == incomingInterruptFD) { 
                    interruptedIndex = i;
                    interrupted = true;
                    break;
                }
            }
            return updated;
        }

这个poll方法干了2件事情  
1是将变更过事件的fd，重新设置epoll中  
2是调用epoll\_wait函数进行select，取到发生事件的fd，epoll会将发生的事件放到pollArrayAddress这个数组中。

先看下是怎样将变更过事件的fd，重新设置epoll中，也就是updateRegistrations方法

private void updateRegistrations() { 
            synchronized (updateLock) { 
                int j = 0;
                while (j < updateCount) { 
                    int fd = updateDescriptors[j];
    
                    // 获取事件
                    short events = getUpdateEvents(fd);
                    boolean isRegistered = registered.get(fd);
                    int opcode = 0;
    
                    if (events != KILLED) { 
    
                        if (isRegistered) { 
                            // 如果已经注册过，如果事件不是0的话，就是修改操作，如果是0的话，就是删除操作
                            opcode = (events != 0) ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
                        } else { 
                            // 如果没有注册过，不是0的话，就是添加操作。否则就是啥也不干
                            opcode = (events != 0) ? EPOLL_CTL_ADD : 0;
                        }
                        // 如果有操作，就进行操作
                        if (opcode != 0) { 
                            // 调用epoll来
                            epollCtl(epfd, opcode, fd, events);
                            // 如果是添加操作。放入registered 这个set中
                            if (opcode == EPOLL_CTL_ADD) { 
                                registered.set(fd);
                                // 如果是删除操作，就从set中清除这个fd
                            } else if (opcode == EPOLL_CTL_DEL) { 
                                registered.clear(fd);
                            }
                        }
                    }
                    j++;
                }
    
                //重置updateCount
                updateCount = 0;
            }
        }

这个方法代码有点琐碎，但是不难理解，就是遍历updateDescriptors数组中所有变更过事件的fd，然后判断一下之前是否注册过，然后对应的操作调用 `epollCtl(epfd, opcode, fd, events);`方法重新往epoll中注册，这个epollCtl 方法其实就是调用的epoll\_ctl函数

JNIEXPORT void JNICALL
    Java_sun_nio_ch_EPollArrayWrapper_epollCtl(JNIEnv *env, jobject this, jint epfd,
                                               jint opcode, jint fd, jint events)
    { 
        struct epoll_event event;
        int res;
        event.events = events;
        event.data.fd = fd;
        // 调用epoll_ctl函数
        RESTARTABLE(epoll_ctl(epfd, (int)opcode, (int)fd, &event), res);
        if (res < 0 && errno != EBADF && errno != ENOENT && errno != EPERM) { 
            JNU_ThrowIOExceptionWithLastError(env, "epoll_ctl failed");
        }
    }

再来看看epollWait 方法是怎样实现的

private native int epollWait(long pollAddress, int numfds, long timeout,
                                     int epfd) throws IOException;

可以看到这个epollWait也是个native 方法，直接到EPollArrayWrapper.c 里面找到对应的c函数即可

JNIEXPORT jint JNICALL
    Java_sun_nio_ch_EPollArrayWrapper_epollWait(JNIEnv *env, jobject this,
                                                jlong address, jint numfds,
                                                jlong timeout, jint epfd)
    { 
        struct epoll_event *events = jlong_to_ptr(address);
        int res;
        if (timeout <= 0) {            /* Indefinite or no wait */
            RESTARTABLE(epoll_wait(epfd, events, numfds, timeout), res);
        } else {                       /* Bounded wait; bounded restarts */
            res = iepoll(epfd, events, numfds, timeout);
        }
        if (res < 0) { 
            JNU_ThrowIOExceptionWithLastError(env, "epoll_wait failed");
        }
        return res;
    }

可以看到是调用的iepoll 这个函数

static int
    iepoll(int epfd, struct epoll_event *events, int numfds, jlong timeout)
    { 
        jlong start, now;
        int remaining = timeout;
        struct timeval t;
        int diff;
    
        gettimeofday(&t, NULL);
        start = t.tv_sec * 1000 + t.tv_usec / 1000;
    
        for (;;) { 
        	// epoll_wait
            int res = epoll_wait(epfd, events, numfds, timeout);
            if (res < 0 && errno == EINTR) { 
                if (remaining >= 0) { 
                    gettimeofday(&t, NULL);
                    now = t.tv_sec * 1000 + t.tv_usec / 1000;
                    diff = now - start;
                    remaining -= diff;
                    if (diff < 0 || remaining <= 0) { 
                        return 0;
                    }
                    start = now;
                }
            } else { 
                return res;
            }
        }
    }

可以看到iepoll 函数里面就是调用的epoll\_wait函数。

好了，doSelect 方法中的pollWrapper.poll(timeout); 代码底层实现我们就知道了，最后看下updateSelectedKeys 方法的底层原理。

private int updateSelectedKeys() { 
            int entries = pollWrapper.updated;
            int numKeysUpdated = 0;
            for (int i=0; i<entries; i++) { 
                //获取事件对应的文件描述符
                int nextFD = pollWrapper.getDescriptor(i);
                // 从映射中获取对应fd的selectionKey
                SelectionKeyImpl ski = fdToKey.get(Integer.valueOf(nextFD));
                // ski is null in the case of an interrupt
                // ski在中断的情况下是空的
                if (ski != null) { 
                    // 获取对应的事件
                    int rOps = pollWrapper.getEventOps(i);
                    //已经存在的情况
                    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;
        }

updateSelectedKeys 这个方法干的事情就是，取出epoll\_wait发生的事件，然后找到事件对应fd对应的那个selectionKey，进行事件的转换，并设置准备好的事件到selectionKey，最后将selectionKey 添加到selectedKeys 集合中去。

### 总结 ###

到这里我们select方法底层源码就剖析完成了，其实就干了4件事：  
1是删除哪些取消注册维护的一些数据结构信息  
2是将变更过的fd，重新调用epoll\_ctl函数将感兴趣事件信息维护到epoll中  
3是调用epoll\_wait函数进行select，等待事件发生，返回事件发生个数  
4是将第3步发生的事件fd与事件信息维护到对应selectionkey中，并将selectionKey添加到selectionKeys集合中去。