NIO源碼詳解

時間 2019-12-14

標籤 nio 源碼詳解欄目 Netty 简体版

原文原文鏈接

阻塞io和無阻塞io：
阻塞io是指jdk1.4以前版本面向流的io，服務端須要對每一個請求創建一堆線程等待請求，而客戶端發送請求後，先諮詢服務端是否有線程相應，若是沒有則會一直等待或者遭到拒絕請求，若是有的話，客戶端會線程會等待請求結束後才繼續執行。
當併發量大，然後端服務或客戶端處理數據慢時就會產生產生大量線程處於等待中，即上述的阻塞。java

無阻塞io是使用單線程或者只使用少許的多線程，每一個鏈接共用一個線程，當處於等待（沒有事件）的時候線程資源能夠釋放出來處理別的請求，經過事件驅動模型當有accept/read/write等事件發生後通知（喚醒）主線程分配資源來處理相關事件。java.nio.channels.Selector就是在該模型中事件的觀察者，能夠將多個SocketChannel的事件註冊到一個Selector上，當沒有事件發生時Selector處於阻塞狀態，當SocketChannel有accept/read/write等事件發生時喚醒Selector。

這個Selector是使用了單線程模型，主要用來描述事件驅動模型，要優化性能須要一個好的線程模型來使用，目前比較好的nio框架有Netty，apache的mina等。線程模型這塊後面再分享，這裏重點研究Selector的阻塞和喚醒原理。
退出阻塞的方式有：register在selector上的socketChannel處於就緒狀態（放在pollArray中的socketChannel的FD就緒）或者第1節中放在pollArray中的wakeupSourceFd就緒。前者（socketChannel）就緒喚醒應證了文章開始的阻塞->事件驅動->喚醒的過程，後者（wakeupSourceFd）就是下面要看的主動wakeup。apache

這裏建立了一個管道pipe，並對pipe的source端的POLLIN事件感興趣，addWakeupSocket方法將source的POLLIN事件標識爲感興趣的，當sink端有數據寫入時，source對應的文件描述描wakeupSourceFd就會處於就緒狀態。（事實上windows就是經過向管道中寫數據來喚醒阻塞的選擇器的）從以上代碼能夠看出：通道的打開其實是構造了一個SelectorImpl對象windows

subSelector.poll() 是select的核心，由native函數poll0實現，readFds、writeFds 和exceptFds數組用來保存底層select的結果，數組的第一個位置都是存放發生事件的socket的總數，其他位置存放發生事件的socket句柄fd

圖像詳解後端

1，Selector.open()數組

Pipe.open()打開一個管道拿到wakeupSourceFd和wakeupSinkFd兩個文件描述符；把喚醒端的文件描述符（wakeupSourceFd）放到pollWrapper裏；上圖中最下面那部分建立pipe的過程，windows下的實現是建立兩個本地的socketChannel,而後鏈接（連接的過程經過寫一個隨機long作兩個socket的連接校驗），兩個socketChannel分別實現了管道的source與sink端。
source端由前面提到的WindowsSelectorImpl放到了pollWrapper中（pollWrapper.addWakeupSocket(wakeupSourceFd, 0)）數據結構

pollWrapper用Unsafe類申請一塊物理內存，存放註冊時的socket句柄fdVal和event的數據結構pollfd，其中pollfd共8字節，0~3字節保存socket句柄，4~7字節保存event多線程

先了解一下Unsafe的基本操做併發

//分配var1字節大小的內存，返回起始地址偏移量 
 public native long allocateMemory(long var1);
//從新給var1起始地址的內存分配長度爲var3字節大小的內存，返回新的內存起始地址偏移量
 public native long reallocateMemory(long var1, long var3);
 //釋放起始地址爲var1的內存
 public native void freeMemory(long var1);

pollWrapper申請了一個64個字節的對外內存空間，address爲內存空間的開始地址app

PollArrayWrapper pollWrapper = new PollArrayWrapper(8);
PollArrayWrapper(int var1) { int var2 = var1 * SIZE_POLLFD; 
this.pollArray = new AllocatedNativeObject(var2, true); 
this.pollArrayAddress = this.pollArray.address(); this.size = var1; }

protected NativeObject(int var1, boolean var2) {
    if(!var2) {
        this.allocationAddress = unsafe.allocateMemory((long)var1);
        this.address = this.allocationAddress;
    } else {
        int var3 = pageSize();
        long var4 = unsafe.allocateMemory((long)(var1 + var3));
        this.allocationAddress = var4;
        this.address = var4 + (long)var3 - (var4 & (long)(var3 - 1));
    }

}

pollWrapper.addWakeupSocket(wakeupSourceFd, 0)，把喚醒端的文件描述符（wakeupSourceFd）放到pollWrapper裏，本次操做pollfd會使用6個字節框架

void addWakeupSocket(int var1, int var2) {
    this.putDescriptor(var2, var1);
    this.putEventOps(var2, Net.POLLIN);
}

void putDescriptor(int var1, int var2) {
    this.pollArray.putInt(SIZE_POLLFD * var1 + 0, var2);
}

void putEventOps(int var1, int var2) {
    this.pollArray.putShort(SIZE_POLLFD * var1 + 4, (short)var2);
}

實際上pollfd的結構

2，Channel.Register()

3，Selector.select()

4，SelectionKey.cancel()

上圖淺藍色部分

網上找的一些輔助理解圖片