淺析Java NIO

時間 2019-11-18

標籤淺析 java nio 欄目 Java 简体版

原文原文鏈接

淺析Java NIO

前言

在說NIO以前，先來講說IO的讀寫原理。咱們都知道Java中的IO流能夠分爲網絡IO流和文件IO流，前者在網絡中使用，後者在操做文件時使用。但實際上兩種流區別並非太大，對於操做系統來講區別僅僅是和硬盤打交道仍是和網卡打交道。
其次，咱們直接操控的是Jvm虛擬機，虛擬機是運行在操做系統上的、用戶層面的進程，jvm虛擬機並不能直接操控底層硬件（這也是爲何Java不多用來作壞事的緣由之一），而是向系統進行發出調用申請。
所以，當Jvm運行到IO流的read方法後會向系統發出read系統調用，由系統使用硬盤驅動來從硬盤讀取數據（這只是個簡單的比喻，實際狀況是有點複雜的）。須要注意的是系統並不會直接把數據從硬盤複製到Jvm內存中，而是把數據先複製到一個「內核緩衝區」的地方。咱們使用字節流時都會new一個byte數組做爲緩衝區，這個緩衝區是用戶緩衝區，內核中也存在這樣一個緩衝區。因此一個常規的IO流讀取文件的過程是這樣的：硬盤 -> 內核緩衝區 -> 用戶緩衝區（Jvm內存，也就是byte數組），寫操做也是一樣的道理。
當內核沒有準備好數據的時候，整個用戶進程是阻塞的，直到系統吧數據從內核緩存移動到jvm內存中後整個進程纔會繼續運行下去。系統從本地文件讀取數據時可能會快一點，可是當從網卡讀取數據時因爲網絡延遲的存在，其效率會很是低，而且一個線程只能處理一個網絡請求。
若是有多個客戶端訪問時雖然能夠開多線程來處理，可是線程是一種「很是貴」的資源，不管線程是否工做，虛擬機會爲每一個線程至少保留128K~1M的空間，而且當線程多了以後，線程之間爭搶資源、CPU頻繁切換不一樣線程會致使整個系統都效率低下（切換線程須要保存當前線程上下文，浪費CPU性能）。html

何爲NIO & 爲何使用NIO

什麼是NIO：java

NIO的官方解釋是：NIO stands for non-blocking I/O（並不是某些人所說的 new IO），直譯就是非阻塞IO。須要說明的是Java中的NIO並不屬於非阻塞IO模型，而是IO複用模型，不過一樣實現了非阻塞狀態。linux

與普通IO的不一樣數據庫

普通的IO的核心是Stream ，NIO的核心是Buffer（ 緩存區）、Channel（通道）和Selector（選擇器）。編程

爲何使用NIOsegmentfault

須要明白的是NIO解決了網絡中一個線程只能處理單個鏈接的痛點。還能夠減小文件傳輸時CPU在存儲中反覆拷貝的反作用，即減小了系統內核和用戶應用程序地址空間這二者之間進行數據拷貝，這就是零拷貝（zero copy）技術）。設計模式

那麼什麼是Buffer（緩存區）、Channel（通道）和Selector（選擇器）呢？
Buffer這個比較好理解，就是一個用來存放數據的地方，即緩衝區。Channel則有點像流，不過Channel是雙向的，數據能夠從Buffer讀取到channel中，也能夠從channel中寫入到buffer。數組

Selector則是選擇器，用來對多個channel進行篩選，進而挑出能夠處理的channel，這樣就把多線程級別的處理降級爲單線程的輪詢，不用咱們手動維護線程池而交給selector來處理。須要注意的是調用selector的select()方法後若是沒有可用的channel，此時該線程是阻塞的。緩存

Buffer

Buffer（緩衝區）和使用普通IO流時建立的byte數組區別並不大，只不過封裝成了類並添加了一些特有的方法。
Buffer的翻譯是什麼？緩衝啊，緩衝區是幹什麼的？存取數據唄，怎麼存？put、get呀。所以Buffer的核心方法即是put()和get()。
同時，Buffer維護了四個變量來描述這個數據緩衝區的狀態：安全

容量Capacity：緩衝區可以容納的數據元素的最大數量。容量在緩衝區建立時被設定，而且永遠不能被改變。(不能被改變的緣由也很簡單，底層是數組嘛)
上界Limit：緩衝區裏的數據的總數（可以存或者讀的界限），表明了當前緩衝區中一共有多少數據。
位置Position：下一個要被讀或寫的元素的位置。Position會自動由相應的 get( )和 put( )函數更新。

直接用起來大概就是這個樣子：

//使用allocate()方法構建緩衝區，分配大小爲128字節
ByteBuffer byteBuffer = ByteBuffer.allocate(128);
//寫入數據
byteBuffer.put("Java".getBytes());
//切換模式
byteBuffer.flip();

while (byteBuffer.hasRemaining()){//Remaining : 剩餘
    System.out.println((char)byteBuffer.get());
}

看flip()的源碼就會發現也就這樣（flip : 翻動）：

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

想從新寫入數據能夠調用clear()方法：

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

沒錯，clear後數據還在，只不過position歸0不能讀了。想從新讀取能夠調用rewind()方法（rewind : 倒帶）：

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

那若是讀取到一半又想寫入了怎麼辦呢？能夠調用compact()方法，這個方法能夠將全部未讀的數據拷貝到Buffer起始處。而後將position設到最後一個未讀元素正後面。limit屬性依然像clear()方法同樣，設置成capacity。如今Buffer準備好寫數據了，可是不會覆蓋未讀的數據。(compact : 緊湊)

調用position()方法能夠得到當前position的位置。

可能有同窗發現了，上面我說這個類維護了四個變量來描述緩衝區，我卻只列出了三個，而且在源代碼中頻繁出現了mark這個關鍵字，沒錯，這就是第四個變量，用來當作做爲一個標記。能夠調用mark()方法來標記此時的position的位置，而後調用reset()方法將position回到此處，下面是源碼：

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

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

簡單粗暴😂

Buffer的類型：

ByteBuffer
MappedByteBuffer
CharBuffer
DoubleBuffer
FloatBuffer
IntBuffer
LongBuffer
ShortBuffer

想說咋沒有StringBuffer？😏

仔細想一想StringBuffer在哪一個包裏？

StringBuffer在java.lang包裏啊喂(#`O′)，StringBuilder、StringBuffer是可變字符串，StringBuilder更快可是線程不安全，StringBuffer慢一點可是線程安全。😆

Channel

Channel的做用正如它的翻譯：渠道，渠道的做用是什麼，運輸水呀，在程序中數據就像水同樣。（通常將Channel翻譯爲通道）

下面是JAVA NIO中的一些主要Channel的實現：

FileChannel（文件IO）
SocketChannel（TCP IO）
DatagramChannel（UDP IO）
ServerSocketChannel（服務端監聽TCP鏈接，並創建響應通道）

可見既包括了網絡流也包括了文件流。

FileChannel & AsynchronousFileChannel

FileChannel是操做文件的流，能夠由FileInputStream、FileOutputStream、RandomAccessFile三個東東的getChannel()方法來獲取通道。

得到通道以後可使用read(buffer)和write(buffer)來從通道讀取數據到緩衝區或者把數據從緩衝區寫入到數據。
而且，Channel不只僅能夠讀/寫一個緩衝區，還能夠讀寫緩衝區數組。

不過很遺憾，FileChannel是阻塞的，上面所說的非阻塞值得是網絡IO，文件IO只能運行在阻塞模式下。
而且因爲能夠自由操控FileChannel的position，在寫入文件時會可能產生「文件空洞」，這可能會破壞文件。在效率上，使用通道進行文件拷貝和使用普通IO流進行拷貝差異並不大，甚至使用通道還會更麻煩一點（由於多一步從流獲取通道的過程）。

我上面有說：NIO能夠減小數據傳輸時CPU反覆拷貝，這裏貼篇文章吧：《經過零拷貝實現有效數據傳輸》，這篇文章的原理就是直接操做內核緩存，通道對通道，不通過用戶緩衝區，所以能夠提升IO效率，具體實現本文再也不贅述。

文件IO的特性僅止於此了嗎？並非！
咱們可使用真正的異步IO（AIO）來進行文件的讀寫：AsynchronousFileChannel 類。要知道NIO是1.4版本加入的，而AsynchronousFileChannel 是1.7版本才加入的、真正的異步IO！
具體細節請看個人另外一篇文章：《NIO前奏之Path、Files、AsynchronousFileChannel》

通道間傳輸

通道除了和緩衝交換數據以外還能夠直接和通道交換數據。

transferFrom()和transferTo()：
使用起來大概是這個樣子：

接收數據的通道.transferFrom(開始位置，數據量，發送數據的通道);
發送數據的通道.transferTo(開始位置，數據量，接收數據的通道);

兩個方法使用起來差異不大。

網絡IO模型

在說Selector以前先來簡單認識一下常見的網絡IO模型。
通常來講網絡IO有5種模型：

同步阻塞IO（Blocking IO）
同步非阻塞IO（Non-blocking IO）
IO多路複用（IO Multiplexing）
信號驅動IO（signal-driven IO）
異步IO（Asynchronous IO）

其中前4種爲同步IO，只有第5個纔是異步IO。

因爲信號驅動模型使用很少，這裏再也不說明。

服務器端編程常常須要構造高性能的IO模型，常見的IO模型有四種：

（1）同步阻塞IO（Blocking IO）

首先，解釋一下這裏的阻塞與非阻塞：

阻塞IO，指的是須要內核IO操做完全完成後，才返回到用戶空間，執行用戶的操做。阻塞指的是用戶空間程序的執行狀態，用戶空間程序需等到IO操做完全完成。傳統的IO模型都是同步阻塞IO。在java中，默認建立的socket都是阻塞的。

其次，解釋一下同步與異步：

同步IO，是一種用戶空間與內核空間的調用發起方式。同步IO是指用戶空間線程是主動發起IO請求的一方，內核空間是被動接受方。異步IO則反過來，是指內核kernel是主動發起IO請求的一方，用戶線程是被動接受方。

（2）同步非阻塞IO（Non-blocking IO）

非阻塞IO，指的是用戶程序不須要等待內核IO操做完成後，內核當即返回給用戶一個狀態值，用戶空間無需等到內核的IO操做完全完成，能夠當即返回用戶空間，執行用戶的操做，處於非阻塞的狀態。

簡單的說：阻塞是指用戶空間（調用線程）一直在等待，並且別的事情什麼都不作；非阻塞是指用戶空間（調用線程）拿到狀態就返回，IO操做能夠幹就幹，不能夠幹，就去幹的事情。

非阻塞IO要求socket被設置爲NONBLOCK。

強調一下，這裏所說的NIO（同步非阻塞IO）模型，並不是Java的NIO（New IO）庫。

（3）IO多路複用（IO Multiplexing）

即經典的Reactor設計模式，有時也稱爲異步阻塞IO，Java中的Selector和Linux中的epoll都是這種模型。

（5）異步IO（Asynchronous IO）

異步IO，指的是用戶空間與內核空間的調用方式反過來。用戶空間線程是變成被動接受的，內核空間是主動調用者。

這一點，有點相似於Java中比較典型的模式是回調模式，用戶空間線程向內核空間註冊各類IO事件的回調函數，由內核去主動調用。

參考《10分鐘看懂， Java NIO 底層原理》，具體細節能夠看原博客，這裏再也不過多說明。

Selector

在知道了網絡模型以後，咱們就能理解Selector的做用了。

下面我將演示一次一個客戶端和服務端的一次通訊：

服務端

//服務端使用ServerSocketChannel，使用靜態方法open()
ServerSocketChannel ssc = ServerSocketChannel.open();
//設置爲非阻塞模式
ssc.configureBlocking(false);
//監聽端口，能夠不加IP地址，默認本地
ssc.socket().bind(new InetSocketAddress(8888));
//建立選擇器，一樣是靜態方法
Selector selector = Selector.open();
//將通道註冊到選擇器中
ssc.register(selector, SelectionKey.OP_ACCEPT);//這裏會返回一個選擇鍵
//建立一個1024大小的緩衝
ByteBuffer buffer = ByteBuffer.allocate(1024);

while (true) {
    int readyNum = selector.select();//這裏會阻塞
    if (readyNum == 0) {//正常狀況下這裏不可能爲真
        System.out.println("------------------");
        continue;
    }

    Set<SelectionKey> selectionKeys = selector.selectedKeys();//獲取選擇鍵集，這裏能獲取到的是已就緒的選擇鍵
    Iterator<SelectionKey> it = selectionKeys.iterator();//鍵集迭代器

    while (it.hasNext()) {
        SelectionKey key = it.next();//選擇鍵，一個鍵就對應着一個就緒的通道
        it.remove();//獲取以後須要移除，不然下次會繼續迭代該鍵，而後發生空指針異常

        if (key.isAcceptable()) {
            // 接受鏈接
            System.out.println("可接受...");
            //經過key來獲取通道
            SocketChannel accept = ((ServerSocketChannel) key.channel()).accept();//這裏須要進行強制轉換
            //設置爲非阻塞
            accept.configureBlocking(false);
            //註冊爲可讀
            accept.register(selector, SelectionKey.OP_READ);
        } else if (key.isReadable()) {
            // 通道可讀
            SocketChannel clientChannel = (SocketChannel) key.channel();
            System.out.println("有可讀通道...");
            buffer.clear();//這裏必定要清空緩衝，不然第二次訪問沒法讀取數據
            while (clientChannel.read(buffer) > 0) {
                buffer.flip();//改成讀模式
                byte[] bytes = new byte[buffer.limit()];
                buffer.get(bytes);
                System.out.println(new String(bytes));
                buffer.clear();
            }

            //這裏返回0表明還有數據可是沒有讀完，-1表明讀取完畢
            if (clientChannel.read(buffer) < 0) {
                //就收完數據以後註冊爲可寫
                key.interestOps(SelectionKey.OP_WRITE);
            }
        } else if (key.isWritable()) {
            // 通道可寫
            System.out.println("寫");
            SocketChannel channel = (SocketChannel) key.channel();
            buffer.clear();
            buffer.put("OVER".getBytes());
            buffer.flip();
            channel.write(buffer);
            //寫完以後記得關閉通道
            channel.close();//這裏也能夠繼續註冊爲可讀
        }
    }
}

一個Selector面對的是多個channel，一個channel也能夠註冊多個selector（可是不推薦這麼作），而描述selector和channel的關係的就是選擇鍵SelectionKey。
register()方法的第二個參數Selectionkey.OP_READ表明該選擇器對該通道的那個方面比較感興趣，總共有四種時間類型：

Accept接收事件，用SelectionKey.OP_ACCEPT表示。（常量1 << 4也就是16）
Connect鏈接事件，用SelectionKey.OP_CONNECT表示。（常量1 << 3也就是8）
Write可寫事件，用SelectionKey.OP_WRITE表示。（常量1 << 2也就是4）
Read可讀事件，用SelectionKey.OP_READ表示。（常量1 << 0也就是1）

若是一個通道不止對一種事件感興趣，能夠這麼表達：SelectionKey.OP_READ | SelectionKey.OP_WRITE（其實用+也能夠的）

須要注意的是：ServerSocketChannel只能註冊OP_ACCEPT，SocketChannel只能註冊OP_CONNECT、OP_WRITE、OP_READ。

經過選擇鍵的interestOps()方法能夠得到感興趣的集合的值，而後經過下面這種蹩腳的方式判斷該鍵是否對某個事件感興趣（沒啥用）：

int interestSet = selectionKey.interestOps();
//對可接受是否感興趣
boolean isInterestedInAccept  = (interestSet & SelectionKey.OP_ACCEPT) == SelectionKey.OP_ACCEPT;
//或者
boolean isInterestedInAccept  = (interestSet & SelectionKey.OP_ACCEPT) > 0;

具體原理是這些數都是2的冪，interestSet是2的冪的和。

實際咱們處理的仍是已經就緒的選擇鍵，並經過下面的方法判斷是否對某個事件感興趣：

key.isAcceptable();
key.isConnectable();
key.isReadable();
key.isWritable();

既然選擇鍵是選擇器和通道的關係，那麼選擇鍵固然能夠得到選擇器和通道：

SelectableChannel selectableChannel = key.channel();
Selector selector = key.selector();

須要注意的是key.channel()返回的是抽象父類，須要向下強制轉換來使用。

register()方法的第三個參數是一個附件，綁定到key上能夠用來傳遞數據：

SelectionKey key = channel.register(selector, SelectionKey.OP_READ, theObject);

該key能夠得到或者是設置這個附件：

//添加新的附件會致使舊的附件丟失，也能夠添加null來主動丟棄附件
selectionKey.attach(theObject);
//獲取附件
Object attachedObj = selectionKey.attachment();//固然使用的時候仍是須要向下轉型的

整個過程是一個死循環，當沒有註冊過的就緒通道時，循環會在Selector.open()這裏阻塞，直到有就緒通道，方法的返回值是已就緒的通道數量。
選擇器還有下面兩種方法：

int select(long timeout)設定最長阻塞時間（毫秒），時間到了以後會中止阻塞，若是沒有可用通道會返回 0
int selectNow()當即返回，不阻塞

Selector的selectedKeys()方法能夠返回已就緒的通道的選擇鍵集，而後對每一個選擇鍵進行不一樣的操做。

注意每次迭代的it.remove()調用。Selector不會本身從已選擇鍵集中移除SelectionKey實例。必須在處理完通道時本身移除。下次該通道變成就緒時，Selector會再次將其放入已選擇鍵集中。
而且若是一個就緒通道被選擇以後沒有任何操做，那麼下次改通道還會被選中。
上面兩種不要搞混了，前者是內層while循環，後者是外側while循環。

某個線程調用select()方法阻塞後，其它線程能夠調用該選擇器的wakeUp()方法來讓這次阻塞當即返回，若是在未阻塞的狀況下調用該方法的話則會取消下次select()阻塞。

選擇器再也不使用以後能夠調用close()方法來關閉該選擇器，這樣會使註冊過的SelectionKey失效，但不會使通道關閉。

至此服務端就介紹完畢了，下面是客戶端，客戶端可使用nio，也可使用普通的socket IO

NIO 客戶端

@SuppressWarnings("all")
public class Client {
    public static void main(String[] args) throws Exception {
        //客戶端建立的是SocketChannel通道，InetSocketAddress默認是本地地址
        SocketChannel channel = SocketChannel.open(new InetSocketAddress(8888));
        //設置爲非阻塞模式
        channel.configureBlocking(false);

        //也能夠用下面的方式建立通道
//        SocketChannel channel1 = SocketChannel.open();
//        channel1.configureBlocking(false);
//        channel1.connect(new InetSocketAddress(8888));
//        上面這種寫法有可能在鏈接上以前就返回了，因此須要使用channel1.finishConnect()來判斷是否鏈接上了。
        
        //建立緩衝
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        //建立選擇器
        Selector selector = Selector.open();

        //註冊可寫（客戶端先發送數據）
        //也能夠註冊可鏈接，而後在下面的循環裏添加一個可鏈接分支，在分支裏鏈接
        channel.register(selector, SelectionKey.OP_WRITE);

        while (true) {
            int numReady = selector.select();
            if (numReady == 0) {
                continue;
            }
            //鍵集迭代器
            Iterator<SelectionKey> it = selector.selectedKeys().iterator();
            while (it.hasNext()) {
                //獲取選擇鍵
                SelectionKey key = it.next();
                //迭代以後移該除選擇鍵
                it.remove();
                if (key.isWritable()) {
                    System.out.println("可寫");
                    SocketChannel channel1 = (SocketChannel) key.channel();
                    buffer.put("爲美好的世界獻上祝福".getBytes());
                    for (int i = 0; i < 3; i++) {
                        buffer.flip();
                        channel1.write(buffer);
                        System.out.println(i);
                        //模擬網絡延遲
                        Thread.sleep(1000);
                    }
                    //這裏須要通知服務端已經書寫完畢，不然服務端會一直嘗試讀取
                    channel.shutdownOutput();
                    //而後把該通道註冊爲可讀
                    key.interestOps(SelectionKey.OP_READ);
                } else if (key.isReadable()) {
                    System.out.println("可讀");
                    //讀以前須要清空緩衝區，不然有時候會讀不進去
                    buffer.clear();
                    SocketChannel readChannel = (SocketChannel) key.channel();
                    while (readChannel.read(buffer) > 0) {
                        buffer.flip();
                        byte[] bytes = new byte[buffer.limit()];
                        buffer.get(bytes);
                        System.out.println(new String(bytes));
                        buffer.clear();
                    }
                    //收到寫入完畢的信號以後關閉通道，並結束。
                    if (readChannel.read(buffer) == -1) {
                        readChannel.close();
                        return;
                    }
                }
            }
            //循環標記,沒什麼用
            System.out.println("=================");
        }
    }
}

和服務端差不太多，沒什麼好說的。

BIO 客戶端

Channel說白了就是對Socket的封裝，讓其能夠配合選擇器在單線程上處理多個鏈接，所以也可使用普通Socket來鏈接 NIO 服務器。

普通IO的客戶端就很簡單了：

public class Client2 {
    public static void main(String[] args) throws Exception {
        Socket socket = new Socket("127.0.0.1",8888);
        OutputStream os = socket.getOutputStream();
        os.write("爲美好的世界獻上祝福".getBytes());
        os.flush();
        socket.shutdownOutput();
        InputStream is = socket.getInputStream();
        byte[] bytes = new byte[1024];
        int len = is.read(bytes);
        System.out.println(new String(bytes,0,len));
        socket.shutdownInput();
        is.close();
        os.close();
        socket.close();
    }
}

至此，TCP相關的鏈接就說完了，下面是UDP的連接方法。

DatagramChannel

DatagramChannel是一個能收發 UDP 包的通道。

TCP與UDP效率比較：

TCP協議適用於對效率要求相對低，但對準確性要求相對高的場景下，或者是有一種鏈接概念的場景下；而UDP協議適用於對效率要求相對高，對準確性要求相對低的場景。

TCP與UDP應用場景：

TCP能夠用於網絡數據庫，分佈式高精度計算系統的數據傳輸；UDP能夠用於服務系統內部之間的數據傳輸，由於數據可能比較多，內部系統局域網內的丟包錯包率又很低，即使丟包，頂可能是操做無效，這種狀況下，UDP常常被使用。（大部分遊戲都是UDP）

由於 UDP 是無鏈接的網絡協議，因此不能像其它通道那樣讀取和寫入。它發送和接收的是數據包。

代碼仍是比較簡單的：

public class UDPServer {
    public static void main(String[] args) throws Exception{
        DatagramChannel channel = DatagramChannel.open();
        //監聽
        channel.socket().bind(new InetSocketAddress(8888));
        //設置非阻塞
        channel.configureBlocking(false);

        ByteBuffer buffer = ByteBuffer.allocate(1024);

        //建立選擇器
        Selector selector = Selector.open();
        //註冊可讀
        channel.register(selector, SelectionKey.OP_READ);

        while (selector.select()>0){
            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()){
                SelectionKey key = iterator.next();
                iterator.remove();

                if (key.isReadable()){
                    System.out.println("可讀");
                    DatagramChannel client = (DatagramChannel) key.channel();
                    buffer.clear();
                    client.receive(buffer);
                    buffer.flip();
                    System.out.println(new String(buffer.array(),0,buffer.limit()));
                    buffer.clear();
                    //這裏不能關閉通道，不然第二次訪問時接收不到數據
                    //client.close();
                }
            }
        }
    }
}

客戶端：

public class UDPClient {
    public static void main(String[] args) throws Exception{
        DatagramChannel channel = DatagramChannel.open();
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        buffer.put("爲美好的世界獻上祝福".getBytes());
        buffer.flip();
        channel.send(buffer, new InetSocketAddress("127.0.0.1",8888));//並不知道服務器是否接收到
        channel.close();
    }
}

由於UDP是無鏈接的，因此數據沒有保障，也就是說不打開服務端只打開客戶端也不會報錯。

不過DatagramChannel也能夠調用connect()方法，只不過這個不是真正的連接，只是至關於綁定了地址而已：

channel.connect(new InetSocketAddress("127.0.0.1", 8888));

這樣就能夠了調用其read和write方法了。

Pipe（管道）

Pipe管道能夠在兩個線程間進行單向數據傳輸，Pipe有一個 source 通道和一個 sink 通道。數據會被寫到 sink 通道，從 source 通道讀取：

代碼很簡單：

public class TestPipe {
    public static void main(String[] args) throws Exception {
        Pipe pipe = Pipe.open();
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        buffer.put("爲美好的世界獻上祝福".getBytes());
        buffer.flip();
        Pipe.SinkChannel sink = pipe.sink();
        while (buffer.hasRemaining()) {
            sink.write(buffer);
        }

        Pipe.SourceChannel source = pipe.source();
        buffer.clear();
        source.read(buffer);
        buffer.flip();
        System.out.println(new String(buffer.array(),0,buffer.limit()));
    }
}