【J2SE】java NIO 基礎(chǔ)學(xué)習(xí)

yexiaobai 發(fā)布于2019-08-16 15:49 / 906人閱讀

摘要：標(biāo)記，表示記錄當(dāng)前的位置。直接緩沖通過方法分配的緩沖區(qū)，此緩沖區(qū)建立在物理內(nèi)存中。直接在兩個(gè)空間中開辟內(nèi)存空間，創(chuàng)建映射文件，去除了在內(nèi)核地址空間和用戶地址空間中的操作，使得直接通過物理內(nèi)存?zhèn)鬏敂?shù)據(jù)。

NIO與IO的區(qū)別

IO	NIO
阻塞式	非阻塞式、選擇器selectors
面向流：?jiǎn)蜗蛄鲃?dòng)，直接將數(shù)據(jù)從一方流向另一方	面向緩存：將數(shù)據(jù)放到緩存區(qū)中進(jìn)行存取，經(jīng)通道進(jìn)行數(shù)據(jù)的傳輸

緩沖Buffer

根據(jù)數(shù)據(jù)類型的不同，提供了對(duì)應(yīng)的類型緩沖區(qū)（boolean類型除外），每一個(gè)Buffer類都是Buffer接口的一個(gè)實(shí)例。通過Buffer類.allocate()方法獲取緩沖區(qū)；對(duì)緩沖區(qū)的數(shù)據(jù)進(jìn)行操作可以使用put方法和get方法。

四個(gè)核心屬性

// Invariants: mark <= position <= limit <= capacity
private int mark = -1;
private int position = 0;
private int limit;
private int capacity;

capacity：容量，表示緩沖區(qū)中最大存儲(chǔ)容量，一旦聲明不可更改。

limit：界限，表示限制可對(duì)緩沖區(qū)操作數(shù)據(jù)的范圍，范圍外的數(shù)據(jù)不可被操作。

position：位置，表示當(dāng)前操作的數(shù)據(jù)位于緩沖區(qū)中的位置。

mark：標(biāo)記，表示記錄當(dāng)前position的位置。

常用方法（以ByteBuffer為例）

public static ByteBuffer allocateDirect(int capacity)：分配一個(gè)直接緩沖區(qū)public static ByteBuffer allocate(int capacity)：分配一個(gè)間接緩沖區(qū)

當(dāng)分配一個(gè)緩沖區(qū)時(shí)，capacity=capacity，mark=-1， position=0， limit=capacity，源碼分析如下：

public static ByteBuffer allocate(int capacity) {
    ...
    return new HeapByteBuffer(capacity, capacity);
}

// class HeapByteBuffer extends ByteBuffer
HeapByteBuffer(int cap, int lim) {
    // 調(diào)用ByteBuffer的構(gòu)造函數(shù)傳入默認(rèn)參數(shù)：mark=-1， position=0， limit=capacity
    super(-1, 0, lim, cap, new byte[cap], 0);
};

// public abstract class ByteBuffer extends Buffer
ByteBuffer(int mark, int pos, int lim, int cap, byte[] hb, int offset) {
    super(mark, pos, lim, cap);
    this.hb = hb;            // final byte[] hb; 
    this.offset = offset;    // final int offset;
}

Buffer(int mark, int pos, int lim, int cap) { 
    ...
    this.capacity = cap;
    limit(lim);        //    設(shè)置limit
    position(pos);    //    設(shè)置position
    if (mark >= 0) {
        ...
        this.mark = mark;
    }
}

public final ByteBuffer put(byte[] src)：將一個(gè)字節(jié)數(shù)組放入緩沖區(qū)。

每當(dāng)放置一個(gè)字節(jié)時(shí)，position將會(huì)+1，保證position的值就是下一個(gè)可插入數(shù)據(jù)的buffer單元位置。源碼分析如下：

public final ByteBuffer put(byte[] src) {
    return put(src, 0, src.length);    
}

// 由allocate方法調(diào)用分配緩沖區(qū)可知，返回的是Buffer的實(shí)現(xiàn)類HeapByteBuffer對(duì)象
public ByteBuffer put(byte[] src, int offset, int length) {
    checkBounds(offset, length, src.length);    // 檢查是否下標(biāo)越界
    if (length > remaining())                    // 檢查是否超出了可操作的數(shù)據(jù)范圍= limit-position
        throw new BufferOverflowException();
    System.arraycopy(src, offset, hb, ix(position()), length);
    position(position() + length);                // 重設(shè)position
    return this;
}

public ByteBuffer get(byte[] dst)：從緩沖區(qū)中讀取數(shù)據(jù)到 dst中。應(yīng)在 flip() 方法后調(diào)用。

獲取數(shù)據(jù)，是在緩沖區(qū)字節(jié)數(shù)組中的position位置處開始，讀取一次完畢后，并會(huì)記錄當(dāng)前讀取的位置，即position，以便于下一次調(diào)用get方法繼續(xù)讀取。

public ByteBuffer get(byte[] dst) {
    return get(dst, 0, dst.length);
}

// 調(diào)用HeapByteBuffer對(duì)象的get方法
public ByteBuffer get(byte[] dst, int offset, int length) {
    ...
    // 從緩沖區(qū)的字節(jié)數(shù)組final byte[] hb中，拷貝從 hb的 offset+position（注：offset=0） 處的長(zhǎng)度為length的數(shù)據(jù)到 dst中
    System.arraycopy(hb, ix(position()), dst, offset, length);
    position(position() + length);    // 設(shè)置position
    return this;
}

通過源碼分析可知，當(dāng)put操作后，position記錄的是下一個(gè)可用的buffer單元，而get會(huì)從position位置處開始獲取數(shù)據(jù)，這顯然是無(wú)法獲得的，因此需要重新設(shè)置 position，即 flip()方法。

public final Buffer flip() ：翻轉(zhuǎn)緩沖區(qū)，在一個(gè)通道讀取或PUT操作序列之后，調(diào)用此方法以準(zhǔn)備一個(gè)通道寫入或相對(duì)獲取操作的序列

將此通道的緩沖區(qū)的界限設(shè)置為當(dāng)前position，保證了有可操作的數(shù)據(jù)。

public final Buffer flip() {
    limit = position;
    position = 0;
    mark = -1;
    return this;
}

public final Buffer mark()：標(biāo)記當(dāng)前position

可用于在put操作轉(zhuǎn)get操作時(shí)標(biāo)記當(dāng)前的position位置，以便于調(diào)用reset方法從該位置繼續(xù)操作

public final Buffer mark() {
    mark = position;
    return this;
}

public final Buffer reset()：回到mark標(biāo)記的位置

public final Buffer reset() {
    int m = mark;
    if (m < 0)
        throw new InvalidMarkException();
    position = m;
    return this;
}

public final Buffer clear()：清除緩沖，重置初始化原始狀態(tài)

public final Buffer clear() {
    position = 0;
    limit = capacity;
    mark = -1;
    return this;
}

public final Buffer rewind()：倒回，用于重新讀取數(shù)據(jù)

public final Buffer rewind() {
    position = 0;
    mark = -1;
    return this;
}

直接緩沖區(qū)與間接緩沖區(qū)

間接緩沖：通過allocate方法分配的緩沖區(qū)。當(dāng)程序發(fā)起read請(qǐng)求獲取磁盤文件時(shí)，該文件首先被OS讀取到內(nèi)核地址空間中，并copy一份原始數(shù)據(jù)傳入JVM用戶地址空間，再傳給應(yīng)用程序。增加了一個(gè)copy操作，導(dǎo)致效率降低。

直接緩沖：通過allocateDirecr方法分配的緩沖區(qū)，此緩沖區(qū)建立在物理內(nèi)存中。直接在兩個(gè)空間中開辟內(nèi)存空間，創(chuàng)建映射文件，去除了在內(nèi)核地址空間和用戶地址空間中的copy操作，使得直接通過物理內(nèi)存?zhèn)鬏敂?shù)據(jù)。雖然有效提高了效率，但是分配和銷毀該緩沖區(qū)的成本高于間接緩沖，且對(duì)于緩沖區(qū)中的數(shù)據(jù)將交付給OS管理，程序員無(wú)法控制。

通道Channel

用于源節(jié)點(diǎn)與目標(biāo)節(jié)點(diǎn)之間的連接，負(fù)責(zé)對(duì)緩沖區(qū)中的數(shù)據(jù)提供傳輸服務(wù)。

常用類

? FileChannel：用于讀取、寫入、映射和操作文件的通道。

? SocketChannel：通過 TCP 讀寫網(wǎng)絡(luò)中的數(shù)據(jù)。

? ServerSocketChannerl：通過 UDP 讀寫網(wǎng)絡(luò)中的數(shù)據(jù)通道。

? DatagramChannel：通過 UDP 讀寫網(wǎng)絡(luò)中的數(shù)據(jù)通道。

? 本地IO：FileInputStream、FileOutputStream、RandomAccessFile

? 網(wǎng)絡(luò)IO：Socket、ServerSocket、DatagramSocket

獲取Channel方式（以FileChannel為例）

? 1. Files.newByteChannel工具類靜態(tài)方法

? 2. getChannel方法：通過對(duì)象動(dòng)態(tài)獲取，使用間接緩沖區(qū)。

FileInputStream fis = new FileInputStream(ORIGINAL_FILE);
FileOutputStream fos = new FileOutputStream(OUTPUT_FILE);

// 獲取通道
FileChannel inChannel = fis.getChannel();
FileChannel outChannel = fos.getChannel();

// 提供緩沖區(qū)（間接緩沖區(qū)）
ByteBuffer buffer = ByteBuffer.allocate(1024);
while (inChannel.read(buffer) != -1) {
    buffer.flip();
    outChannel.write(buffer);
    buffer.clear();
}

? 3. 靜態(tài)open方法：使用open獲取到的Channel通道，使用直接緩沖區(qū)。

FileChannel inChannel = 
    FileChannel.open(Paths.get(ORIGINAL_FILE), StandardOpenOption.READ);
FileChannel outChannel = 
    FileChannel.open(Paths.get(OUTPUT_FILE), StandardOpenOption.READ,
    StandardOpenOption.CREATE, StandardOpenOption.WRITE);
    
// 使用物理內(nèi)存 內(nèi)存映射文件 
MappedByteBuffer inBuffer = 
    inChannel.map(MapMode.READ_ONLY, 0, inChannel.size());
MappedByteBuffer outBuffer = 
    outChannel.map(MapMode.READ_WRITE, 0, inChannel.size());
    
byte[] dst = new byte[inBuffer.limit()];
inBuffer.get(dst);
outBuffer.put(dst);

// 使用DMA 直接存儲(chǔ)器存儲(chǔ)
inChannel.transferTo(0, inChannel.size(), outChannel);
outChannel.transferFrom(inChannel, 0, inChannel.size());

public static FileChannel open(Path path, OpenOption... options)：從path路徑中以某種方式獲取文件的Channel

StandardOpenOption	描述
CREATE	創(chuàng)建一個(gè)新的文件，如果存在，則覆蓋。
CREATE_NEW	創(chuàng)建一個(gè)新的文件，如果該文件已經(jīng)存在則失敗。
DELETE_ON_CLOSE	關(guān)閉時(shí)刪除。
DSYNC	要求將文件內(nèi)容的每次更新都與底層存儲(chǔ)設(shè)備同步寫入。
READ	讀方式
SPARSE	稀疏文件
SYNC	要求將文件內(nèi)容或元數(shù)據(jù)的每次更新都同步寫入底層存儲(chǔ)設(shè)備。
TRUNCATE_EXISTING	如果文件已經(jīng)存在，并且打開 wirte訪問，則其長(zhǎng)度將截?cái)酁?。
WRITE	寫方式
APPEND	如果文件以wirte訪問打開，則字節(jié)將被寫入文件的末尾而不是開頭。

public abstract MappedByteBuffer map(MapMode mode, long position, long size)：將通道的文件區(qū)域映射到內(nèi)從中。當(dāng)操作較大的文件時(shí)，將數(shù)據(jù)映射到物理內(nèi)存中才是值得的，因?yàn)橛成涞絻?nèi)存是需要開銷的。

FileChannel.MapMode	描述
PRIVATE	專用映射模式（寫入時(shí)拷貝）
READ_ONLY	只讀模式
READ_WRIT	讀寫模式

public abstract long transferFrom(ReadableByteChannel src, long position, long count)：從給定的可讀取通道src，傳輸?shù)奖就ǖ乐小Ｖ苯邮褂弥苯哟鎯?chǔ)器（DMA）對(duì)數(shù)據(jù)進(jìn)行存儲(chǔ)。public abstract long transferTo(long position, long count, WritableByteChannel target)：將本通道的文件傳輸?shù)娇蓪懭氲膖arget通道中。

分散（Scatter）與聚集（Gather）

? 分散讀取：將通道中的數(shù)據(jù)分散到多個(gè)緩沖區(qū)中。 public final long read(ByteBuffer[] dsts)

? 聚集寫入：將多個(gè)緩沖區(qū)中的數(shù)據(jù)聚集到一個(gè)Channel通道中。public final long write(ByteBuffer[] srcs)

字符集（Charset）

public final ByteBuffer encode(CharBuffer cb)：編碼public final CharBuffer decode(ByteBuffer bb)：解碼

網(wǎng)絡(luò)通信的阻塞與非阻塞

阻塞是相對(duì)網(wǎng)絡(luò)傳輸而言的。傳統(tǒng)的IO流都是阻塞的，在網(wǎng)絡(luò)通信中，由于 IO 阻塞，需要為每一個(gè)客戶端創(chuàng)建一個(gè)獨(dú)立的線程來(lái)進(jìn)行數(shù)據(jù)傳輸，性能大大降低；而NIO是非阻塞的，當(dāng)存在空閑線程時(shí)，可以轉(zhuǎn)去操作其他通道，因此不必非要?jiǎng)?chuàng)建一個(gè)獨(dú)立的線程來(lái)服務(wù)每一個(gè)客戶端請(qǐng)求。

選擇器（Selector）

SelectableChannle對(duì)象的多路復(fù)用器，可同時(shí)對(duì)多個(gè)SelectableChannle對(duì)象的 IO 狀態(tài)監(jiān)聽，每當(dāng)創(chuàng)建一個(gè)Channel時(shí)，就向Selector進(jìn)行注冊(cè)，交由Selector進(jìn)行管理，只有Channel準(zhǔn)備就緒時(shí)，Selector可會(huì)將任務(wù)分配給一個(gè)或多個(gè)線程去執(zhí)行。Selector可以同時(shí)管理多個(gè)Channel，是非阻塞 IO 的核心。

NIO 阻塞式

服務(wù)器Server不斷監(jiān)聽客戶端Client的請(qǐng)求，當(dāng)建立了一個(gè)Channel時(shí)，服務(wù)器進(jìn)行read操作，接收客戶端發(fā)送的數(shù)據(jù)，只有當(dāng)客戶端斷開連接close，或者執(zhí)行shutdownOutput操作時(shí)，服務(wù)器才知曉沒有數(shù)據(jù)了，否則會(huì)一直進(jìn)行read操作；當(dāng)客戶端在read操作獲取服務(wù)器的反饋時(shí)，若服務(wù)器沒有關(guān)閉連接或者shutdownInput時(shí)也會(huì)一直阻塞。示例代碼如下：

 static final String ORIGINAL_FILE = "F:/1.png";
 static final String OUTPUT_FILE = "F:/2.jpg";

public void server() throws Exception {
    // 打開TCP通道，綁定端口監(jiān)聽
    ServerSocketChannel serverChannel = ServerSocketChannel.open();
    serverChannel.bind(new InetSocketAddress(9988));    
    
    ByteBuffer buf = ByteBuffer.allocate(1024);
    // 獲取連接
    SocketChannel accept = null;
    while ((accept= serverChannel.accept()) != null) {
        FileChannel fileChannel = FileChannel.open(
            Paths.get(OUTPUT_FILE), StandardOpenOption.CREATE,
            StandardOpenOption.WRITE);
        
        // 讀取客戶端的請(qǐng)求數(shù)據(jù)
        while (accept.read(buf) != -1) {
            buf.flip();
            fileChannel.write(buf);
            buf.clear();
        }
        
        // 發(fā)送執(zhí)行結(jié)果
        buf.put("成功接收".getBytes());
        buf.flip();
        accept.write(buf);
        buf.clear();
        
        fileChannel.close();
        // 關(guān)閉連接，否則客戶端會(huì)一直等待讀取導(dǎo)致阻塞，可使用shutdownInput，但任務(wù)已結(jié)束，該close
        accept.close();
    }
    serverChannel.close();
}

public void client() throws Exception {
    // 打開一個(gè)socket通道
    SocketChannel clientChannel = SocketChannel.open(
        new InetSocketAddress("127.0.0.1", 9988));
    // 創(chuàng)建緩沖區(qū)和文件傳輸通道
    FileChannel fileChannel = FileChannel.open(Paths.get(ORIGINAL_FILE),
            StandardOpenOption.READ);
    ByteBuffer buf = ByteBuffer.allocate(1024);
    
    while ( fileChannel.read(buf) != -1) {
        buf.flip();
        clientChannel.write(buf);
        buf.clear();
    }
    
    // 關(guān)閉輸出（不關(guān)閉通道），告知服務(wù)器已經(jīng)發(fā)送完畢，去掉下面一行代碼服務(wù)區(qū)將一直讀取導(dǎo)致阻塞
    clientChannel.shutdownOutput();
    
    int len = 0;
    while ((len = clientChannel.read(buf)) != -1) {
        buf.flip();
        System.out.println(new String(buf.array(), 0, len));
        buf.clear();
    }
    
    fileChannel.close();
    clientChannel.close();
}

NIO 非阻塞式

通過在通道Channel中調(diào)用configureBlocking將blocking設(shè)置為false，讓Channel可以進(jìn)行異步 I/O 操作。

public void client() throws Exception {
    // 打開一個(gè)socket通道
    SocketChannel clientChannel = SocketChannel.open(
        new InetSocketAddress("127.0.0.1", 9988));
    ByteBuffer buf = ByteBuffer.allocate(1024);

    // 告知服務(wù)器，已經(jīng)發(fā)送完畢
    //    clientChannel.shutdownOutput();
    //    設(shè)置非阻塞
    clientChannel.configureBlocking(Boolean.FALSE);

    buf.put("哈哈".getBytes());
    buf.flip();
    clientChannel.write(buf);

    clientChannel.close();
}

public void server() throws Exception {
    // 打開TCP通道，綁定端口監(jiān)聽
    ServerSocketChannel serverChannel = ServerSocketChannel.open();
    serverChannel.configureBlocking(Boolean.FALSE);
    serverChannel.bind(new InetSocketAddress(9988));

    //    創(chuàng)建一個(gè)Selector用于管理Channel
    Selector selector = Selector.open();
    //    將服務(wù)器的Channel注冊(cè)到selector中，并添加 OP_ACCEPT 事件，讓selector監(jiān)聽通道的請(qǐng)求
    serverChannel.register(selector, SelectionKey.OP_ACCEPT);

    // 一直判斷是否有已經(jīng)準(zhǔn)備就緒的Channel
    while (selector.select() > 0) {
          // 存在一個(gè)已經(jīng)準(zhǔn)備就緒的Channel，獲取SelectionKey集合中獲取觸發(fā)該事件的所有key
        Iterator keys = selector.selectedKeys().iterator();
        while (keys.hasNext()) {
            SelectionKey sk = keys.next();
            SocketChannel accept = null;
            ByteBuffer buffer = null;
            // 針對(duì)不同的狀態(tài)進(jìn)行操作
            if (sk.isAcceptable()) {
                // 可被連接，設(shè)置非阻塞并注冊(cè)到selector中
                accept = serverChannel.accept();
                accept.configureBlocking(Boolean.FALSE);
                accept.register(selector, SelectionKey.OP_READ);
            } else if (sk.isReadable()) {
                // 可讀，獲取該選擇器上的 Channel進(jìn)行讀操作
                accept = (SocketChannel) sk.channel();
                buffer = ByteBuffer.allocate(1024);
                int len = 0;
                while ((len = accept.read(buffer)) != -1) {
                    buffer.flip();
                    System.out.println(new String(buffer.array(), 0, len));
                    buffer.clear();
                }
            }
        }
        // 移除本次操作的SelectionKey
        keys.remove();
    }
    serverChannel.close();
}

方法使用說(shuō)明

ServerSocketChannel對(duì)象只能注冊(cè)accept 事件。

設(shè)置configureBlocking為false，才能使套接字通道中進(jìn)行異步 I/O 操作。

調(diào)用selectedKeys方法，返回發(fā)生了SelectionKey對(duì)象的集合。

調(diào)用remove方法，用于從SelectionKey集合中移除已經(jīng)被處理的key，若不處理，那么它將繼續(xù)以當(dāng)前的激活事件狀態(tài)繼續(xù)存在。

Pipe管道

Channel都是雙向通道傳輸，而Pipe就是為了實(shí)現(xiàn)單向管道傳送的通道對(duì)，有一個(gè)source通道（Pipe.SourceChannel）和一個(gè)sink通道（Pipe.SinkChannel）。sink用于寫數(shù)據(jù)，source用于讀數(shù)據(jù)。直接使用Pipe.open()獲取Pipe對(duì)象，操作和FileChannel一樣。