深入理解NIO（三）—— NIO原理及部分源碼的解析

深入理解NIO（三）—— NIO原理及部分源碼的解析

歡迎回到淦™的源碼看爆系列

在看完前面兩個系列之后，相信大家對NIO也有了一定的理解，接下來我們就來深入源碼去解讀它，我這里的是OpenJDK-8u60版本，建議大家也下一份放ide里和我一起看會比較好理解。（這里主要介紹Selector，Buffer第一篇有提到一點，Channel也不過是些Buffer的操作方法而已，這里就不提及了，大家感興趣可以自己去看）

老哥行行好，轉載和我說一聲好嗎，我不介意轉載的，但是請把原文鏈接貼大點好嗎

 

open()

// 1. 創建Selector
Selector selector = Selector.open();

首先我們來分析open方法:

// Selector
public static Selector open() throws IOException {
    // 這里的靜態方法provider會使用DefaultSelectorProvider.create();方法根據系統選擇一個SelectorProvider
    // windows平台的話是WindowsSelectorProvider，
    // Linux平台是一個EPollSelectorProvider，這里主要分析Linux平台下的
    // 之后openSelector方法(一會看下面)會返回一個EPollSelectorImpl作為Selector的實現，我們一般提及的Selector就是它了
    return SelectorProvider.provider().openSelector();
}

// EPollSelectorProvider
public AbstractSelector openSelector() throws IOException {
    return new EPollSelectorImpl(this);
}

之后是EPollSelectorImpl的構造方法：

EPollSelectorImpl(SelectorProvider sp) throws IOException {
    super(sp);
    long pipeFds = IOUtil.makePipe(false);
    fd0 = (int) (pipeFds >>> 32);
    fd1 = (int) pipeFds;
    // 其他的我也看不太懂，我們直接進去這個EPollArrayWrapper的構造方法
    pollWrapper = new EPollArrayWrapper();
    pollWrapper.initInterrupt(fd0, fd1);
    fdToKey = new HashMap<>();
}

// EPollArrayWrapper
EPollArrayWrapper() throws IOException {
    // 直接看這里，這里調用了一個封裝出來的Linux的api：epoll_create，這個東西大概可以理解成一個selector，詳細的我們下一章再講解
    epfd = epollCreate();
}

 

所以其實Selector方法大抵上就是封裝了一個epoll_create() 方法，當然還調用了一下epoll_ctl()，把serverchannel給注冊進去。

 

register()

// 5. 將channel注冊到selector上，監聽連接事件
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

接下來我們分析把channel注冊到Selector上的register方法

// SelectableChannel
public final SelectionKey register(Selector sel, int ops)
        throws ClosedChannelException
{
    return register(sel, ops, null);
}

// AbstractSelectableChannel
public final SelectionKey register(Selector sel, int ops,
                                   Object att)
        throws ClosedChannelException
{
    // 212行，剩下的刪掉了
    k = ((AbstractSelector)sel).register(this, ops, att);
                
}

// SelectorImpl
protected final SelectionKey register(AbstractSelectableChannel ch,
                                      int ops,
                                      Object attachment)
{
    if (!(ch instanceof SelChImpl))
        throw new IllegalSelectorException();
    //生成SelectorKey來存儲到hashmap中，一共之后獲取
    SelectionKeyImpl k = new SelectionKeyImpl((SelChImpl)ch, this);
    //attach用戶想要存儲的對象
    k.attach(attachment);
    //調用子類的implRegister方法，接下來進去這里
    synchronized (publicKeys) {
        implRegister(k);
    }
    //設置關注的option
    k.interestOps(ops);
    return k;
}

 

protected void implRegister(SelectionKeyImpl ski) {
        if (closed)
            throw new ClosedSelectorException();
        SelChImpl ch = ski.channel;
        //獲取Channel所對應的fd,因為在linux下socket會被當作一個文件，也會有fd
        int fd = Integer.valueOf(ch.getFDVal());
        fdToKey.put(fd, ski);
        //調用pollWrapper的add方法,將channel的fd添加到監控列表中
        pollWrapper.add(fd);
        //保存到HashSet中，keys是SelectorImpl的成員變量
        keys.add(ski);
}

 

調用register方法並沒有涉及到EpollArrayWrapper中的native方法epollCtl的調用,這是因為他們將這個方法的調用推遲到Select方法中去了.

 

select()

// 獲取可用channel數量
int readyChannels = selector.select();

接下來我們來分析select()方法

// SelectorImpl
public int select(long timeout)
        throws IOException
{
    . . . .
    return lockAndDoSelect((timeout == 0) ? -1 : timeout);
}

// SelectorImpl
private int lockAndDoSelect(long timeout) throws IOException {
    . . . .
    return doSelect(timeout);
}

// EPollSelectorImpl
protected int doSelect(long timeout) throws IOException {
    .....
    try {
        ....
        //調用了poll方法,底層調用了native的epollCtl和epollWait方法
        pollWrapper.poll(timeout);
    } finally {
        ....
    }
    ....
    //更新selectedKeys,為之后的selectedKeys函數做准備
    int numKeysUpdated = updateSelectedKeys();
    ....
    return numKeysUpdated;
}

// EPollArrayWrapper
int poll(long timeout) throws IOException {
    // 這里面的實現就是調用epoll_ctl()方法注冊先前在register方法中保存的Channel的fd和感興趣的事件類型
    updateRegistrations();
    // 這里是調用epollWait方法等待感興趣事件的生成,導致線程阻塞
    updated = epollWait(pollArrayAddress, NUM_EPOLLEVENTS, timeout, epfd);
    . . . .
}

 

上面提到的epollCtl和epollWait方法在下一章我們會詳細講，這里先不講。

總之我們可以知道Selector其實就是封裝了Linux提供的api而已，也就是epollCreate、epollCtl和epollWait方法。

 

selectedKeys()

// 獲取可用channel的集合
Set<SelectionKey> selectionKeys = selector.selectedKeys();

接下來我們來看看selectedKeys()方法：

// SelectorImpl
//是通過Util.ungrowableSet生成的,不能添加,只能減少
private Set<SelectionKey> publicSelectedKeys;
public Set<SelectionKey> selectedKeys() {
    ....
    return publicSelectedKeys;
}

很奇怪啊,怎麽直接就返回publicSelectedKeys了,難道在select函數的執行過程中有修改過這個變量嗎?publicSelectedKeys這個對象其實是selectedKeys變量的一份副本,你可以在SelectorImpl的構造函數中找到它們倆的關系,我們再回頭看一下select中updateSelectedKeys方法：

private int updateSelectedKeys() {
    //更新了的keys的個數,或在說是產生的事件的個數
    int entries = pollWrapper.updated; 
    int numKeysUpdated = 0;
    for (int i=0; i<entries; i++) {
        //對應的channel的fd
        int nextFD = pollWrapper.getDescriptor(i);
        //通過fd找到對應的SelectionKey
        SelectionKeyImpl ski = fdToKey.get(Integer.valueOf(nextFD));
        if (ski != null) {
            int rOps = pollWrapper.getEventOps(i);
            //更新selectedKey變量,並通知響應的channel來做響應的處理
            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;
}

 

不知道大家有沒有留意到，如果我們不先調用select()，直接selectedKeys()是不會獲得任何Channel的，因為里面沒有更新publicSelectedKeys的方法

還有一點是，publicSelectedKeys是selectedKeys的引用，所以我們獲得的是它的引用，而不是每次返回一個新對象，這個引用里面的Channel我們處理完后要記得remove掉，不然下次還是會返回給你的。

順便一提這里publicSelectedKeys是采用 publicSelectedKeys = Util.ungrowableSet(selectedKeys); 的方式創建出來的，這個方法創建出來的set如方法名ungrowableSet，是不能調用add方法的，只能remove

 

 

為什么Netty自己又從新實現了一邊native相關的NIO底層方法? 聽聽Netty的創始人是怎麽說的吧鏈接。

因為Java的版本使用的epoll的LT模式，而Netty則希望使用ET模式（詳情看第四篇的兩種觸發模式），而且Java版本沒有將epoll的部分配置項暴露出來，比如說TCP_CORK和SO_REUSEPORT。

 

下一篇：epoll的實現原理

 

 

---

 

參考資料：

https://segmentfault.com/a/1190000017798684?utm_source=tag-newest