從 I/O 模型到 Netty（二）

Java NIO

在上一篇文章中對於I/O模型已經講的比較清楚了，在I/O密集型應用中使用Reactor模式能夠明顯提升系統的性能（咱們這裏談到的性能很大程度上指的是吞吐量），可是在具體的開發過程當中模式仍是要落地成真實的代碼，使用傳統的I/O庫確定是不行的，在Java中須要使用java.nio包下的庫。

雖然是講NIO的實現，但本文將不會把全部Java NIO中的主要API所有過一遍，而是經過例子理清NIO到底能夠作什麼事情。

本文中提到的JDK源代碼均可以在%JAVA_HOME%/jre/lib/rt.jar中看到。

Java NIO最初在Java4中被引入，可是到今天仍是有很大部分的開發者歷來沒使用過NIO的API，由於基礎I/O已經能知足了咱們平常的開發需求。但若是要開發I/O密集型應用的場景下，NIO能夠明顯的提高程序的性能，另外NIO與基礎I/O有本質思想上的區別。
本文主要講Java中的NIO，內容包含：

1. Oracle官方對NIO的說法
2. Java中NIO的歷史進程
3. NIO和NIO.2的區別在哪裏
4. NIO中的主要類的介紹
5. 使用NIO的API構建一個Socket服務器

Oracle官方對NIO的說法

首先看看Oracle的官方文檔中是怎麼說的：

Java中對於I/O的支持主要包括java.io 和java.nio兩個包的內容，它們共同提供了以下特性：

1. 經過數據流和序列化從文件系統中讀取和寫數據。
2. 提供Charsets，解碼器和編碼器，用於在字節和Unicode字符之間的翻譯。
3. 訪問文件、文件的屬性、文件系統。
4. 提供異步的或者非阻塞多路複用I/O的API，用於構建可擴展的服務器程序。

這裏並無提到網絡I/O的東西，在Java1.4之前，網絡I/O的API都是被放在java.net包下，在NIO中才被一塊兒放入了java.nio包下。

Java中NIO的歷史進程

1. 最開始Java中使用I/O來訪問文件系統只有經過java.io.File類來作，其中包含了一些對文件和目錄基本的操做。對於開發中常碰到的I/O需求通常都能覆蓋到，因此這也是平常開發工做中最常使用的I/O API。官方文檔中稱之爲基礎I/O（Basic I/O）。
   基礎I/O是基於各類流的概念的，其基本模型就是上一篇中講到的阻塞I/O。

2. 爲了進一步豐富I/O操做的API，也是爲了提高在I/O密集型應用中的性能，基於Reactor模式，在Java1.4中引入了java.nio包，其中重點包含幾個類：

   • java.nio.Buffer，用來存儲各類緩衝數據的容器。
   • java.nio.channels.Channel，用於鏈接程序和I/O設備的數據通道。
   • java.nio.channels.Selector，多路複用選擇器，在上一篇中講到過。
   • java.nio.charset.Charset，用來編解碼。

3. 在Java7中引入了NIO.2，引入了一系列新的API（主要在新加入的包Java.nio.file），對於訪問文件系統提供了更多的API實現，更加豐富的文件屬性類，增長了一些異步I/O的API。同時，還添加了不少實用方法。

   例如：之前簡單的拷貝一個文件就必需要寫一大堆的代碼，如今實用java.nio.file.Files.copy(Path, Path, CopyOption...)就能夠很輕鬆的作到了

NIO和NIO.2的區別在哪裏

在上一節中已經簡單介紹了這兩個概念的不一樣，這裏再簡單羅列一下。NIO中引入的一個重要概念就是Reactor模式，而NIO.2對NIO自己不是一次升級，而是一次擴充，NIO.2中新增了不少實用方法（utilities），以支持更多的功能需求，並非說可以提高多少的性能。主要增長了以下兩點：

1. 新的訪問文件的API。
   

   

   訪問文件從簡單到複雜的方法

   
   在Java.nio.file包和其子包中新增了大量的與訪問文件相關的類，其中比較重要的有如下幾個，更完整的更新能夠在Oracle的官網文檔中查看。
   • java.nio.file.Path，它能夠用來取代早期的java.io.File用來訪問文件。
   • java.nio.file.Files，其中包含了大量的對文件操做的API。

2. 異步I/O的API
   在NIO原來的API的基礎上，增長了對Proactor模式的支持，能夠在包java.nio.channels中看到新加入的java.nio.channels.AsynchronousChannel和java.nio.channels.CompletionHandler<V, A>。使用這些類能夠實現異步編程，如代碼1中所示：

   //代碼1
    //定義一個處理文件內容的函數式接口
    @FunctionalInterface
    static interface ProcessBuffer{
        void process(int result, ByteBuffer bb);
    }
    //遞歸地讀取文件的所有內容
    static void readFileThrough(AsynchronousFileChannel ch, ProcessBuffer runn, int position) {
   
        ByteBuffer bb = ByteBuffer.allocate(512);
        ch.read(bb, position, null, new CompletionHandler<Integer, Object>() {
   
            @Override
            public void completed(Integer result, Object attachment) {
                System.out.println("成功了");
                bb.flip();
                runn.process(result, bb);
                bb.clear();
                if (result == bb.capacity())
                    readFileThrough(ch, runn, position + result);
            }
   
            @Override
            public void failed(Throwable exc, Object attachment) {
                System.err.println("失敗了！！！");
            }
        });
    }複製代碼

NIO中的主要類的介紹

NIO的基本思想是要構建一個Reactor模式的實現，具體落實到API，在Java中主要有如下幾個類：

1. java.nio.Buffer

這是一個容器類，用來存儲「基礎數據類型」，全部從Channel中讀取出來的數據都要使用Buffer的子類來做爲存儲單元，能夠把它想象成一個帶着不少屬性的數組（和ArrayList很相似，其實它的實現機制也差很少就是這樣）。

第一次看到介紹Buffer是在一本書上，書上畫了好多方框和指向這些方框的屬性值，看着就頭暈。其實很簡單，Buffer就是一個數組。

在讀寫交換時，必不可少的要批量地去讀取並寫入到目標對象，這個道理是不變的。在基礎I/O中若是咱們要把一個輸入流寫入一個輸出流，可能會這麼作：

//代碼2
public static void copy(File src, File dest) throws IOException {
    FileInputStream in = new FileInputStream(src);
    FileOutputStream out = new FileOutputStream(dest);
    byte[] buffer = new byte[1024];
    int bytes = 0;
    while ((bytes = in.read(buffer)) > -1){
        out.write(buffer, 0, bytes);
    }
    out.close();
    in.close();
}複製代碼

以上代碼中使用了一個真實的數組用來作讀寫切換，從而達到批量（緩衝）讀寫的目標。
而在NIO中（如代碼1），讀寫切換也一樣是使用了一個數組進行暫存（緩衝），只不過在這個數組之上，封裝了一些屬性（java.nio.Buffer源碼中的一些屬性如代碼3所示）和操做。

//代碼3 - Buffer類中定義的一些屬性
// Invariants: mark <= position <= limit <= capacity
private int mark = -1;
private int position = 0;
private int limit;
private int capacity;複製代碼

關於Buffer類詳細的繼承關係和其主要方法，能夠參考下圖：

Buffer的繼承關係

2. java.nio.channels.Channel

Channel能夠看作是代碼2中InputStream和OutStream的合體，在實際使用中，咱們每每針對同一個I/O設備同時存在讀和寫的操做，在基礎I/O中咱們就須要針對同一個目標對象生成一個輸入流和輸出流的對象，但是在NIO中就能夠只創建一個Channel對象了。

Channel抽象的概念是對於某個I/O設備的「鏈接」，可使用這個鏈接進行一些I/O操做，java.nio.channels.Channel自己是一個接口，只有兩個方法，可是在Java的的環境中，每每最簡單的接口最煩人，由於它的實現類老是會異常的多。

//代碼4 - 去除了全部註釋的Channel類
package java.nio.channels;

import java.io.IOException;
import java.io.Closeable;

public interface Channel extends Closeable {

    public boolean isOpen();

    public void close() throws IOException;

}複製代碼

固然，這是享受多態帶來的好處的同時必須承受的。詳細的Channel繼承和實現關係以下：

Channel的繼承和實現關係

3. java.nio.channels.Selector

若是你是使用NIO來作網絡I/O，Selector是JavaNIO中最重要的類，正如它的註釋裏第一句說的，Selector是SelectableChannel的「多路複用器」。

SelectableChannel的實現類

多路複用，這是在上一篇介紹過的概念，在不一樣的操做系統也有不一樣的底層實現。用戶也能夠本身實現本身的Selector（經過類java.nio.channels.spi.SelectorProvider）

//代碼5 - provider構造方法
public static SelectorProvider provider() {
    synchronized (lock) {
        if (provider != null)
            return provider;
        return AccessController.doPrivileged(
            new PrivilegedAction<SelectorProvider>() {
                public SelectorProvider run() {
                        if (loadProviderFromProperty())
                            //若是設置了屬性java.nio.channels.spi.SelectorProvider，則會載入響應的類
                            return provider;
                        if (loadProviderAsService())
                            return provider;
                        provider = sun.nio.ch.DefaultSelectorProvider.create();
                        return provider;
                    }
                });
    }
}複製代碼

若是你不實現本身的SelectorProvidor，在代碼5中能夠看到JDK會使用類sun.nio.ch.DefaultSelectorProvider來建立，這裏會根據你的操做系統的類別不一樣而選擇不一樣的實現類。openJDK中也有相應的實現，有興趣的能夠去GrepCode查看一下，Mac OS下是使用KQueueSelectorProvider。

Selector的使用比較簡單，同時要配合SelectionKey使用，它們的繼承結構圖也比較簡單，以下：

Selector繼承關係

4. 其餘

其餘一些類如Charset我的感受屬於實用性很強的類，可是在NIO與基礎I/O的比較中就顯得不那麼重要了。

使用NIO的API構建一個Socket服務器

Java1.4引入的NIO中已經能夠實現Reactor模式，在NIO.2中又引入了AIO的API，因此本節將分別使用兩種模式來實現一個Socket服務器，這裏重點介紹Java中NIO API的使用，至於NIO和基礎I/O的性能對比，網上有不少，這裏就再也不作比較了。

首先定義一些基礎類，將從Socket中獲取的數據解析成TestRequest對象，而後再找到響應的Handler。看代碼：

我這裏爲了偷懶，將不少基礎類和方法定義在了一個類中，這種方法其實十分不可取。

//代碼6 
/** * 執行計算工做的線程池 */
private static ExecutorService workers = Executors.newFixedThreadPool(10);

/** * 解析出來的請求對象 * @author lk * */
public static class TestRequest{

    /** * 根據解析到的method來獲取響應的Handler */
    String method;
    String args;
    public static TestRequest parseFromString(String req) {
        System.out.println("收到請求：" + req);
        TestRequest request = new TestRequest();
        request.method = req.substring(0, 512);
        request.args = req.substring(512, req.length());
        return request;
    }
}


/** * 具體的邏輯須要實現此接口 * @author lk * */
public static interface SockerServerHandler {
    ByteBuffer handle(TestRequest req);
}複製代碼

主要的邏輯其實就是使用ServerSocketChannel的實例監聽本地端口，而且設置其爲非阻塞（默認爲阻塞模式）。代碼7中的parse()函數是一個典型的「使用Buffer讀取Channel中數據」的方法，這裏爲了簡(tou)單(lan)，默認只讀取1024個字節，因此並無實際去循環讀取。

//代碼7
private static void useNIO() {
    Selector dispatcher = null;
    ServerSocketChannel serverChannel = null;
    try {
        dispatcher = Selector.open();
        serverChannel = ServerSocketChannel.open();
        serverChannel.configureBlocking(false);

        serverChannel.socket().setReuseAddress(true);
        serverChannel.socket().bind(LOCAL_8080);

        //ServerSocketChannel只支持這一種key，由於server端的socket只能去accept
        serverChannel.register(dispatcher, SelectionKey.OP_ACCEPT);

        while (dispatcher.select() > 0) {
            operate(dispatcher);
        }

    } catch (Exception e) {
        e.printStackTrace();
    }
}
/** * 在分發器上循環獲取鏈接事件 * @param dispatcher * @throws IOException */
private static void operate(Selector dispatcher) throws IOException {
    //Set<SelectionKey> keys = dispatcher.keys();
    Set<SelectionKey> keys = dispatcher.selectedKeys();
    Iterator<SelectionKey> ki = keys.iterator();
    while(ki.hasNext()) {
        SelectionKey key = ki.next();
        ki.remove();
        if (key.isAcceptable()) {
            ServerSocketChannel channel = (ServerSocketChannel) key.channel();
            //針對此socket的IO就是BIO了
            final SocketChannel socket = channel.accept();
            workers.submit(() -> {
                try {

                    TestRequest request = TestRequest.parseFromString(parse(socket));
                    SockerServerHandler handler = (SockerServerHandler) Class.forName(getClassNameForMethod(request.method)).newInstance();

                    socket.write(handler.handle(request));

                } catch (Exception e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }    
            });
        }
    }
}

private static String parse(SocketChannel socket) throws IOException {
    String req = null;

    try {
        ByteBuffer buffer = ByteBuffer.allocateDirect(1024);
        byte[] bytes;
        int count = 0;
        if ((count = socket.read(buffer)) >= 0) {
            buffer.flip();
            bytes = new byte[count];
            buffer.get(bytes);
            req = new String(bytes, Charset.forName("utf-8"));
            buffer.clear();
        }

    } finally {
        socket.socket().shutdownInput();
    }
    return req;
}複製代碼

Java的程序有個通病，寫出來的程序又臭又長，一樣是使用JavaNIO的API實現一個非阻塞的Socket服務器，使用NIO.2中AIO（異步I/O）的API就很簡單了，可是卻陷入了回調地獄（固然能夠經過別的方式避免回調，可是其本質仍是同樣的）。和上邊介紹的Reactor模式相比，簡直就是拿核武器比步槍，有點降維攻擊的意味了。Reactor中那麼複雜的概念和邏輯所實現的功能，使用AIO的API很輕鬆就搞定了，並且概念比較少，邏輯更清晰。

//代碼8
private static void useAIO() {
    AsynchronousServerSocketChannel server;
    try {
        server = AsynchronousServerSocketChannel.open();
        server.bind(LOCAL_8080);
        while (true) {
            Future<AsynchronousSocketChannel> socketF = server.accept();
            try {
                final AsynchronousSocketChannel socket  = socketF.get();
                workers.submit(() -> {
                    try {
                        ByteBuffer buffer = ByteBuffer.allocateDirect(1024);

                        socket.read(buffer, null, new CompletionHandler<Integer, Object>() {

                            @Override
                            public void completed(Integer count, Object attachment) {
                                byte[] bytes;
                                if (count >= 0) {
                                    buffer.flip();
                                    bytes = new byte[count];
                                    buffer.get(bytes);
                                    String req = new String(bytes, Charset.forName("utf-8"));
                                    TestRequest request = TestRequest.parseFromString(req);
                                    try {
                                        SockerServerHandler handler = (SockerServerHandler) Class.forName(getClassNameForMethod(request.method)).newInstance();
                                        ByteBuffer bb = handler.handle(request);
                                        socket.write(bb, null, null);
                                    } catch (InstantiationException | IllegalAccessException
                                            | ClassNotFoundException e) {
                                        // TODO Auto-generated catch block
                                        e.printStackTrace();
                                    }
                                    buffer.clear();
                                }
                            }

                            @Override
                            public void failed(Throwable exc, Object attachment) {
                                // TODO Auto-generated method stub

                            }
                        });


                    } catch (Exception e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    } finally {

                    }        
                });
            } catch (InterruptedException | ExecutionException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
                break;
            }
        }
    } catch (IOException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
}複製代碼

最後是測試用的客戶端程序，NIO在客戶端一樣也能夠發揮很重要的做用，這裏就先略過了，代碼9中客戶端測試使用的是基礎I/O：

//代碼9
private volatile static int succ = 0;

public static void main(String[] args) throws UnknownHostException, IOException {
    CountDownLatch latch = new CountDownLatch(100);
    for (int i = 0; i < 100; i++) {
        new Thread( () -> {
            Socket soc;
            try {
                soc = new Socket("localhost", 8080);

                if (soc.isConnected()) {
                    OutputStream out = soc.getOutputStream();
                    byte[] req = "hello".getBytes("utf-8");

                    out.write(Arrays.copyOf(req, 1024));
                    InputStream in = soc.getInputStream();
                    byte[] resp = new byte[1024];
                    in.read(resp, 0, 1024);
                    String result = new String(resp, "utf-8");
                    if (result.equals("haha")) {
                        succ++;
                    }
                    System.out.println(Thread.currentThread().getName() + "收到回覆:" + result);
                    out.flush();
                    out.close();
                    in.close();
                    soc.close();
                }
                try {
                    System.out.println(Thread.currentThread().getName() + "去睡覺等待。。。");
                    latch.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();

        latch.countDown();
    }
    Runnable hook = () -> {
        System.out.println("成功個數：" + succ);
    };
    Runtime.getRuntime().addShutdownHook(new Thread(hook));
}複製代碼

總結

本來只是想寫一篇Netty在RPC框架中的使用，寫着寫着就寫多了。本文從Java中引入NIO的歷史講起，梳理了Java對NIO支持的具體的API，最後經過一個典型的Socket服務器的例子具體的展現了Java中NIO相關API的使用，將Reactor模式和Proactor模式從理論落地到實際的代碼。

因爲做者比較懶，貼圖所有都是在網上找的（代碼大部分是本身寫的），如侵刪。下一篇將講到比較火的一個NIO框架Netty的實現與使用。