NIO

提到IO，這是Java提供的一套類庫(kù)，用于支持應(yīng)用程序與內(nèi)存、文件、網(wǎng)絡(luò)間進(jìn)行數(shù)據(jù)交互，實(shí)現(xiàn)數(shù)據(jù)寫入與輸出。JDK自從1.4版本后，提供了另一套類庫(kù)NIO，我們平時(shí)習(xí)慣稱呼為NEW IO或NON-blocking IO。

那么這套新的IO庫(kù)與之前的有何區(qū)別？為什么需要提供這樣一套IO庫(kù)呢？

IO與NIO

Java NIO相比與傳統(tǒng)的IO，除了提供標(biāo)準(zhǔn)IO的加強(qiáng)功能之外，最為核心的是對(duì)基于Socket的網(wǎng)絡(luò)編程提供了一套非阻塞編程模式。

• 流與緩沖
  • Java IO Java的IO很好的詮釋了Stream這個(gè)概念，該單詞本身的含義表示‘河流’，承載數(shù)據(jù)的流，平時(shí)我們說(shuō)的面向流的操作主要是在流的端點(diǎn)，實(shí)現(xiàn)對(duì)數(shù)據(jù)讀與寫。通過(guò)Stream相關(guān)的API可以看到， 不管是輸入還是輸出流，我們能做的僅僅是將數(shù)據(jù)讀取或?qū)懭氲搅髦小?/li>
  • Java NIO NIO是基于緩沖區(qū)來(lái)操作數(shù)據(jù)，主要是基于通道Channel從緩沖Buffer中進(jìn)行數(shù)據(jù)讀取或?qū)懭搿Ｆ渲蠦uffer的靈活性決定了NIO的可操作空間，同樣基于Buffer API可以看到， 其提供了對(duì)Buffer的基本讀寫功能外，還有提供了各種其他API來(lái)操作Buffer，相比Stream對(duì)數(shù)據(jù)的操作更加的靈活。
• 阻塞與非阻塞
  • Java IO 上面說(shuō)到IO的操作都是基于流的，往流中寫入數(shù)據(jù)時(shí)依賴于OutputStream#write，從流中讀取數(shù)據(jù)時(shí)通過(guò)InputStream#read，這些操作都是阻塞的。
  • Java NIO 支持非阻塞模式，但并非NIO就是非阻塞的，比如基于FileChannel操作文件時(shí)，仍然是阻塞的。我們說(shuō)的阻塞或非阻塞都是基于操作系統(tǒng)層面的read/write方法導(dǎo)致的，NIO的非阻塞 基于操作系統(tǒng)層面提供的多路復(fù)用IO模型實(shí)現(xiàn)，所以NIO的實(shí)現(xiàn)是依賴于操作系統(tǒng)的支持。

NIO相關(guān)概念

在NIO中，三個(gè)核心的對(duì)象Buffer、Channel、Selector

Buffer

我們經(jīng)常說(shuō)的面向緩沖區(qū)編程主要對(duì)該對(duì)象的操作，Buffer簡(jiǎn)單的看就是一個(gè)內(nèi)存塊，其內(nèi)部封裝了一個(gè)數(shù)組，同時(shí)該對(duì)象提供了大量API可以靈活對(duì)其操作，比如緩沖數(shù)據(jù)讀取與寫入、緩沖復(fù)制等。

其內(nèi)部結(jié)構(gòu)如下：

其內(nèi)部除了存儲(chǔ)數(shù)據(jù)的數(shù)組外，還維護(hù)了capacity、limit、position幾個(gè)屬性，用于標(biāo)記數(shù)組容量、存儲(chǔ)占用空間、下標(biāo)索引。Buffer存在讀寫兩種狀態(tài)，根據(jù)上圖可以看到其具體含義。

• capacity
  表示Buffer最大可緩沖中數(shù)據(jù)的容量。capacity一旦確定，則不可修改；寫入數(shù)據(jù)一旦達(dá)到容量，則不可繼續(xù)寫入；

• limit
  在寫模式時(shí)，limit=capacity，表示buf可寫入數(shù)據(jù)上限。在讀模式時(shí)，limit表示buf可讀數(shù)據(jù)上限。

• position
  表示Buffer數(shù)組下標(biāo)位置。初始化時(shí)，position=0；

  • 讀模式
  • 寫模式
  1. 當(dāng)緩沖區(qū)剛開始進(jìn)入讀模式時(shí)，position會(huì)被重置為0。
  2. 當(dāng)從緩沖區(qū)讀取時(shí)，也是從position位置開始讀。讀取數(shù)據(jù)后，position向前移動(dòng)到下一個(gè)可讀的位置。
  3. 在讀模式下，limit表示可讀數(shù)據(jù)的上限。position的最大值為最大可讀上限limit，當(dāng)position達(dá)到limit時(shí)表明緩沖區(qū)已經(jīng)無(wú)數(shù)據(jù)可讀。
  4. 在剛進(jìn)入寫模式時(shí)，position為0，表示當(dāng)前的寫入位置為從頭開始。
  5. 當(dāng)有數(shù)據(jù)寫入到緩沖區(qū)后，position會(huì)向后移動(dòng)寫入數(shù)量個(gè)位置。
  6. 初始的position值為0，最大可寫值為limit。當(dāng)position值達(dá)到limit時(shí)，緩沖區(qū)就已經(jīng)無(wú)空間可寫了。

• flip

用于將Buffer由寫狀態(tài)切換為讀狀態(tài)，limit = position; position = 0;

• compact、clear

用于將Buffer由讀狀態(tài)切換為寫狀態(tài)，compact：position=limit，limit=capacity; clear：position=0，limit=capacity。

• mark、reset

操作Buffer時(shí)，用于臨時(shí)存儲(chǔ)position（mark=position），當(dāng)有需要時(shí)，可以通過(guò)rest方法將臨時(shí)值取出并賦值到position（position=mark） 重新從標(biāo)記位置繼續(xù)操作Buffer。

Channel

直譯為通道，表示源端與目標(biāo)端的連接通道，主要負(fù)責(zé)將數(shù)據(jù)讀寫到Buffer。

• 通道可以同時(shí)進(jìn)行讀寫，而流只能讀或者只能寫
• 通道可以實(shí)現(xiàn)異步讀寫數(shù)據(jù)
• 通道可以從緩沖讀數(shù)據(jù)，也可以寫數(shù)據(jù)到緩沖

常用的Channel包括FileChannel、DatagramChannel、ServerSocketChannel和SocketChannel。

• FileChannel 用于文件的數(shù)據(jù)讀寫
• DatagramChannel 用于支持UDP協(xié)議的數(shù)據(jù)讀寫
• ServerSocketChannel和SocketChannel 用于支持TCP協(xié)議的數(shù)據(jù)傳輸

Selector

選擇器是NIO技術(shù)中的核心組件，可以將通道注冊(cè)進(jìn)選擇器中，其主要作用就是使用一個(gè)線程來(lái)對(duì)多個(gè)通道中的已就緒通道進(jìn)行選擇， 然后可以對(duì)選擇的通道進(jìn)行數(shù)據(jù)處理，屬于一對(duì)多的關(guān)系。這種機(jī)制在NIO技術(shù)中心稱為“IO多路復(fù)用”。其優(yōu)勢(shì)是可以在一個(gè)線程中 對(duì)多個(gè)連接實(shí)現(xiàn)監(jiān)聽，從而節(jié)省系統(tǒng)資源與CPU開銷。

其中包括三個(gè)核心類：

• Selector 主操作類，通過(guò)靜態(tài)方法實(shí)例化，通過(guò)select()方法來(lái)監(jiān)聽已經(jīng)注冊(cè)的通道
• SelectionKey 注冊(cè)完通道之后返回的鍵，通過(guò)該類來(lái)描述各個(gè)通道的狀態(tài)
• SelectableChannel 連接通道，通過(guò)該類獲取Socket對(duì)象，將之注冊(cè)到Selector中

我們可以將Channel注冊(cè)到Selector上并定義感興趣的事件，當(dāng)Channel就緒時(shí)，可以監(jiān)聽這些事件：

• Connect 某個(gè)Channel成功連接到另一個(gè)服務(wù)器時(shí)稱為‘連接就緒’，對(duì)應(yīng)常量：SelectionKey.OP_CONNECT
• Accept一個(gè)Server Socket Channel準(zhǔn)備好接收新進(jìn)入的連接稱為‘接收就緒’，對(duì)應(yīng)常量：SelectionKey.OP_ACCEPT
• Read 一個(gè)有數(shù)據(jù)可讀的通道可以說(shuō)是‘讀就緒’，對(duì)應(yīng)常量：SelectionKey.OP_READ
• Write 等待寫數(shù)據(jù)的通道可以說(shuō)是‘寫就緒’，對(duì)應(yīng)常量：SelectionKey.OP_WRITE

示例

1. 文件復(fù)制

傳統(tǒng)IO復(fù)制文件時(shí)需要依賴于InputStream、OutputStream來(lái)完成，基于NIO可以通過(guò)FileChannel：

// 文件復(fù)制
sourceChannel.transferTo(0, sourceChannel.size(), targetChannel);

// 其中獲取FileChannel的方法有以下三種：
FileChannel channel = new FileInputStream(file).getChannel();

FileChannel channel = new RandomAccessFile(file, "rw").getChannel();

FileChannel channel = FileChannel.open(file.toPath());

2. 基于UDP協(xié)議的數(shù)據(jù)傳輸

Server:

@Slf4j
public class Server {

    private Selector selector;
    private DatagramChannel datagramChannel;

    public Server(int port) {
        try {
            this.selector = Selector.open();

            this.datagramChannel = DatagramChannel.open();
            this.datagramChannel.configureBlocking(false);
            this.datagramChannel.bind(new InetSocketAddress(port));

            this.datagramChannel.register(this.selector, SelectionKey.OP_READ);

            log.info("++++++ DUP Server啟動(dòng)成功 ++++++");
        } catch (IOException e) {
            log.error("Server創(chuàng)建失敗：{}", e.getMessage());
        }
    }

    public void start() throws IOException {
        while (true){
            int select = selector.select();
            if(select >0 ){
                Iterator< SelectionKey > iterator = selector.selectedKeys().iterator();
                while (iterator.hasNext()){
                    SelectionKey key = iterator.next();
                    iterator.remove();

                    if(key.isReadable()){
                        DatagramChannel channel = (DatagramChannel) key.channel();
                        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                        channel.receive(byteBuffer);

                        byteBuffer.flip();
                        CharBuffer charBuffer = Charset.defaultCharset ().decode ( byteBuffer ) ;
                        log.info("Server接收消息：{}",  charBuffer);
                    }
                }
            }
        }
    }
}

Client:

@Slf4j
public class Client {

    private DatagramChannel datagramChannel;

    public Client(int port) {
        try {
            this.datagramChannel = DatagramChannel.open();
            this.datagramChannel.configureBlocking(true);
            this.datagramChannel.connect(new InetSocketAddress("127.0.0.1", port));
        } catch (IOException e) {
            log.error("Client創(chuàng)建失敗：{}", e.getMessage());
        }
    }

    public void invoke(String message) throws IOException {
        log.info("Client發(fā)送消息：{}", message);
        datagramChannel.write(Charset.defaultCharset().encode(message));
    }
}

Tests:

public class UDPTest {

    int port = 8095;

    @Test
    public void server() throws IOException {
        Server server = new Server(port);
        server.start();
    }

    @Test
    public void client() throws IOException {
        Client client = new Client(port);
        client.invoke(message);
        while (true){}
    }
}

3. 基于NIO的Socket示例

Server:

@Slf4j
public class Server {

    private ServerSocketChannel serverSocketChannel;
    private Selector selector;

    public Server(int port){
        try {
            this.selector = Selector.open();

            this.serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.configureBlocking(false);

            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            serverSocketChannel.bind(new InetSocketAddress(port));

            log.info("++++++ NIO Server啟動(dòng)成功 ++++++");
        } catch (IOException e) {
            log.error("創(chuàng)建ServerSocketChannel出錯(cuò)：{}", e.getMessage());
        }
    }

    public void start() throws IOException {
        while (true){
            selector.select(); // 阻塞
            Iterator< SelectionKey > keyIterator = selector.selectedKeys().iterator();
            while (keyIterator.hasNext()){
                SelectionKey selectionKey = keyIterator.next();
                keyIterator.remove(); //
                if(!selectionKey.isValid()){
                    continue;
                }
                if(selectionKey.isAcceptable()){
                    ServerSocketChannel ssc = (ServerSocketChannel) selectionKey.channel();
                    SocketChannel socketChannel = ssc.accept(); // 可以是阻塞或非阻塞，獲取的Channel一定是阻塞的
                    socketChannel.configureBlocking(false); // 這個(gè)有用？

                    socketChannel.register(selector, SelectionKey.OP_READ);
                }else if(selectionKey.isReadable()){
                    SocketChannel channel = (SocketChannel) selectionKey.channel();
                    ByteBuffer buffer = ByteBuffer.allocate(256);
                    int writeBytes = channel.read(buffer); //
                    if(writeBytes > 0){
                        buffer.flip();
                        byte[] bytes = new byte[buffer.remaining()];
                        buffer.get(bytes);
                        log.info(" >> > Server接收消息：{}", new String(bytes));
                    }
                    // 回復(fù)
                    channel.write(Charset.defaultCharset().encode("我是Server的回復(fù)內(nèi)容"));
                }
            }
        }
    }
}

Client:

@Slf4j
public class Client {

    private SocketChannel socketChannel;

    public Client(int port){
        try {
            this.socketChannel = SocketChannel.open();
            socketChannel.connect(new InetSocketAddress("127.0.0.1",port));
        } catch (IOException e) {
            log.error("創(chuàng)建SocketChannel出錯(cuò)：{}", e.getMessage());
        }
    }

    public void invoke(String message) throws IOException {
        log.info(" >> > Client發(fā)送消息：{}", message);
        this.socketChannel.write(Charset.defaultCharset().encode(message));
    }

}

NIO整體處理流程如下：

1. 通過(guò)Selector.open()獲取Selector
2. 通過(guò)ServerSocketChannel.open()獲取ServerSocketChannel
3. 設(shè)置ServerSocketChannel為非阻塞模式，ServerSocketChannel.configureBlocking(false)
4. 將Channel綁定到Selector上，并定義關(guān)注的操作類型， serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT)
5. 將ServerSocketChannel綁定Socket，并設(shè)定監(jiān)聽端口，ServerSocketChannel.bind(new InetSocketAddress(port))
6. 開始輪詢Selector
7. 阻塞Selector.select()，直到有準(zhǔn)備就緒的Channel
8. 輪詢Selector.selectedKeys()，獲取這些Channel
9. 基于SelectionKey，按需要可以對(duì)當(dāng)前Channel進(jìn)行Accept、Read、Write等操作
10. 比如當(dāng)接收客戶端鏈接時(shí)，需要將該Channel注冊(cè)到Selector；

零拷貝

首先我們要知道，程序在讀取系統(tǒng)文件時(shí)，是沒(méi)辦法直接讀取磁盤內(nèi)容，基于操作系統(tǒng)安全考慮，需要通過(guò)調(diào)用操作系統(tǒng)提供的系統(tǒng)API從內(nèi)核緩沖區(qū)將文件數(shù)據(jù)拷到用戶緩沖區(qū)后 才能讀取到文件信息。

在操作系統(tǒng)層面，如果為了完成網(wǎng)絡(luò)文件的傳輸，一般需要這樣做：

while( in.read(...)!=-1 ){
   out.write(...) 
}

拿到源文件的輸入流；拿到目標(biāo)文件的輸出流；從輸入流讀取數(shù)據(jù)；將數(shù)據(jù)寫入到輸出流；

整個(gè)過(guò)程經(jīng)歷了4次文件拷貝：

1. 讀取磁盤文件到操作系統(tǒng)內(nèi)核緩沖區(qū)
2. 將內(nèi)核緩沖區(qū)的數(shù)據(jù)，copy到應(yīng)用程序的buffer
3. 將應(yīng)用程序buffer中的數(shù)據(jù)，copy到socket網(wǎng)絡(luò)發(fā)送緩沖區(qū)
4. 將socket buffer的數(shù)據(jù)，copy到網(wǎng)卡，由網(wǎng)卡進(jìn)行網(wǎng)絡(luò)傳輸

經(jīng)歷了4次CPU切換：

1. 程序調(diào)用系統(tǒng)api將文件從磁盤讀取到內(nèi)核態(tài)緩沖區(qū)，用戶態(tài)切換內(nèi)核態(tài)
2. 將數(shù)據(jù)由內(nèi)核態(tài)緩沖區(qū)拷貝到用戶緩沖區(qū)，內(nèi)核態(tài)切換用戶態(tài)
3. 程序調(diào)用系統(tǒng)api將數(shù)據(jù)由用戶緩沖區(qū)拷貝到內(nèi)核緩沖區(qū)，用戶態(tài)切換內(nèi)核態(tài)
4. 將數(shù)據(jù)由內(nèi)核態(tài)緩沖區(qū)拷貝到網(wǎng)卡，內(nèi)核態(tài)切換用戶態(tài)

在高并發(fā)網(wǎng)絡(luò)通信環(huán)境中，通過(guò)傳統(tǒng)的方式由于多次的CPU切換與數(shù)據(jù)拷貝會(huì)消耗系統(tǒng)資源，因此為了提高網(wǎng)絡(luò)間文件傳輸?shù)男阅埽托枰獪p少‘用戶態(tài)與內(nèi)核態(tài)的上下文切換’和‘內(nèi)存拷貝’的次數(shù)。

零拷貝的“零”是指用戶態(tài)和內(nèi)核態(tài)間copy數(shù)據(jù)的次數(shù)為零

零拷貝依附于操作系統(tǒng)底層，基于虛擬內(nèi)存實(shí)現(xiàn)，將文件地址與虛擬地址件建立映射關(guān)系，

零拷貝技術(shù)可以減少數(shù)據(jù)拷貝和共享總線操作的次數(shù)，消除傳輸數(shù)據(jù)在存儲(chǔ)器之間不必要的中間拷貝次數(shù)，從而有效地提高數(shù)據(jù)傳輸效率；零拷貝技術(shù)減少了用戶進(jìn)程地址空間和內(nèi)核地址空間之間因?yàn)樯舷挛那袚Q而帶來(lái)的開銷

• MappedByteBuffer

RandomAccessFile randomAccessFile = new RandomAccessFile(file, "rw");
    FileChannel fileChannel = randomAccessFile.getChannel();
    MappedByteBuffer mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, fileChannel.size());

• DirectByteBuffer
  DirectByteBuffer繼承了MappedByteBuffer，主要是實(shí)現(xiàn)了byte獲得函數(shù)get等
• 零拷貝問(wèn)題

1. 直接內(nèi)存DirectMemory的大小默認(rèn)為-Xmx 的JVM堆的最大值，但是并不受其限制，而是由JVM參數(shù) MaxDirectMemorySize單獨(dú)控制。
2. 直接內(nèi)存不是分配在JVM堆中。并且直接內(nèi)存不受 GC(新生代的Minor GC)影響，只有當(dāng)執(zhí)行老年代的 Full GC時(shí)候才會(huì)順便回收直接內(nèi)存！而直接內(nèi)存是通過(guò)存儲(chǔ)在JVM堆中的DirectByteBuffer對(duì)象來(lái)引用的， 所以當(dāng)眾多的DirectByteBuffer對(duì)象從新生代被送入老年代后才觸發(fā)了 full gc。
3. MappedByteBuffer在處理大文件時(shí)的確性能很高，但也存在一些問(wèn)題，如內(nèi)存占用、文件關(guān)閉不確定，被其打開的文件只有在垃圾回收的才會(huì)被關(guān)閉，而且這個(gè)時(shí)間點(diǎn)是不確定的。

結(jié)束語(yǔ)

NIO的出現(xiàn)得益于操作系統(tǒng)的變革，由于網(wǎng)路編程對(duì)性能與資源使用上的要求更高，傳統(tǒng)的IO模型只能通過(guò)線程來(lái)提升系統(tǒng)吞吐率；為了滿足現(xiàn)代網(wǎng)絡(luò)通信的需求，在高級(jí)編程語(yǔ)言中的優(yōu)化 行為逐步遷移到操作系統(tǒng)底層，這樣通過(guò)底層邏輯優(yōu)化，不僅提供系統(tǒng)性能，最主要減少了系統(tǒng)資源的浪費(fèi)。