Netty源碼分析--NIO（一）

      好久沒寫博客了，最近打算花些時間把Netty的源碼好好讀一讀，下面是本人在學習的過程中的一些筆記，不能確保自己思考的完全是正確的，如果有錯誤，歡迎大家指正。

      由於本人的語文功底爛的很，通篇使用大白話來講解0.0，有一些概念上的東西，博主可能不會明確的給出定義，建議使用過Netty的同學一起來研究。

      好了，我們一起來看下吧。

      Netty 是一款用於快速開發的高性能的網絡應用程序的Java框架。說到Netty, 我們先對幾種I/O模型進行一下比對：

       

       那么偽異步IO是啥呢？

       其實就是加入了線程池（ThreadPoolExecutor），對接入的客戶端的Socket封裝成task，實現了Runnable接口，然后投遞到線程池中處理，這樣就避免了BIO那種一個客戶端連接一個IO線程的情況，防止資源耗盡和宕機。但是這種方式底層的通信依然采用了同步阻塞模型，無法從根本上解決問題。

       那么AIO又是啥呢？

       NIO2.0 引入了新的一步通道的概念，並提供了異步文件通道和異步套接字的實現。它不需要通過多路復用器對注冊的通道進行輪詢操作即可實現異步讀寫，屬於真正意義上的異步非阻塞IO。

       1、通過java.util.concurrent.Future 類來異步獲取操作的結果。

       2、在執行異步操作的時候傳入一個CompletionHandler接口的實現類，作為操作完成的回調。

            接口有以下兩個方法。     

/**
 * Invoked when an operation has completed.
 *
 * @param   result
 *          The result of the I/O operation.
 * @param   attachment
 *          The object attached to the I/O operation when it was initiated.
 */
void completed(V result, A attachment);

/**
 * Invoked when an operation fails.
 *
 * @param   exc
 *          The exception to indicate why the I/O operation failed
 * @param   attachment
 *          The object attached to the I/O operation when it was initiated.
 */
void failed(Throwable exc, A attachment);

 

       好的，下面也稍微回顧一下NIO，以及NIO涉及的幾個關鍵組件：

•  緩沖區 Buffer 
•  通道 Channel
•  多路復用器 Selector

1. Buffer : 看什么都不如看官方文檔來的更准確，下面是官方Buffer javadoc內容，我們來看下： 

          

         里面講述了，buffer抽象類 是一個數據容器，除了內容，還有一些屬性，capacity、limit、position。

         capacity 是容器的容量，這個值一旦被創建，就無法修改。 limit 是 不應該被讀或寫的第一個元素的位置。 position 是指下一個將會被讀或寫的位置，這個值一定小於等於limit。

         另外javadoc中還提到了mark和reset， 其中mark其實就是打一個標記，把當前的position賦給mark。  那么 reset 的 描述是這樣的 把當前的position 改成之前mark的位置。

         

         ok，由上面的文檔可以得出下面的順序  0 <= mark <= position <= limit <= capacity

         其實Buffer中還有一個非常重要的方法必須要說一下，那就是 flip() ，看下javadoc

         

         這個其實就是把 當前的limit = position, position = 0, 當然如果之前有mark也會失效，設置成-1， 當你往buffer中寫了數據的時候，只有執行flip()方法, 才可以正確的讀取數據,  doc中還指出這個方法經常和compact()方法連着用。同樣，貼出javadoc：

        

       相當於什么呢，就相當於是清理掉已經讀取過得數據，比如 position = 5 , limit = 10，前5個數據經讀取過了，那么將新建一個buffer，將當前position到limit的數據拷貝到一個新的Buffer中，那么新的buffer的postion = limit-postion， limit = capacity, 好了，看源碼是這樣的，接下來就是驗證一下了：

ByteBuffer buffer = ByteBuffer.allocate(10);
buffer.put("helloworld".getBytes());
System.out.println(buffer.position() + ":" + buffer.limit());
buffer.flip();
System.out.println(buffer.position() + ":" + buffer.limit());
byte[] bytes = new byte[buffer.limit() + 1];
for(int i=0; i<6; i++) {
    bytes[i] = buffer.get();
}
System.out.println(new String(bytes));
System.out.println(buffer.position() + ":" + buffer.limit());
System.out.println(buffer);
buffer.compact();
System.out.println(buffer.position() + ":" + buffer.limit());
System.out.println(buffer);

測試結果如下：

10:10
0:10
hellow     
6:10
java.nio.HeapByteBuffer[pos=6 lim=10 cap=10]
4:10
java.nio.HeapByteBuffer[pos=4 lim=10 cap=10]

 好了，Buffer的源碼看到這里也算是差不多了。

2、Channel

      Channel是一個通道， 它就像自來水管一樣，網絡數據通過Channel讀取與寫入，通道與流的不同之處在於通道是雙向的，流只是在一個方向上移動（一個流必須是InputStream或者OutStream的子類），而通道可以用於讀、寫或者二者同時進行， 屬於全雙工。

     這里我們也來看下源碼吧，就看ServerSocketChannel

     提供了幾個比較重要的api：  

     public static ServerSocketChannel open() throws IOException； // 通過該方法創建一個Channel

     看下javadoc ， 明確說明了 新創建的channel是沒有任何綁定的，在進行accepted之前需要綁定一個地址。

    

     public final ServerSocketChannel bind(SocketAddress local)；// 綁定一個端口號

     public abstract SocketChannel accept() throws IOException; // 接收新的客戶端

     

 3、Selector 多路復用器 ，簡單來說呢，Selector 會不斷的輪詢注冊在其上的Channel, 如果某個Channel上面發生了讀寫等事件，這個Channel就會處理就緒狀態， 會被Selector輪詢出來，然后拿到SelectionKey Set集合,從而獲取到每一個就緒狀態的Channel，進行后續的I/O操作。

       由於JDK使用了epoll() 代替傳統的select實現，所以沒有最大句柄的1024/2048的限制， 只需要一個線程負責Selector的輪詢，就可以接入成千上萬的客戶端。NB

       

        channel將會通過一個SelectionKey注冊到一個selector上，一個selector 通過 open方法去創建。

       

        這一段着重指出，selectionKey集合只能通過 set 集合的 remove() 方法 或者 一個迭代器的 remove() 方法來移除。其余的方法都不可以修改 selected-key 。

        好了，看到這里，有些朋友可能似懂非懂，但是看下下面的單元測試一下子就懂了。

        這段代碼實現了Nio的服務器端，接收到客戶端消息后，然后通知所有的客戶端。

    private static final Map<String, SocketChannel> clientMap = new ConcurrentHashMap();

    public static void main(String[] args) {

        try {
            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();  // 創建一個Channel
            serverSocketChannel.configureBlocking(false); // 設置為非阻塞
            serverSocketChannel.bind(new InetSocketAddress(8899)); // 綁定端口

            Selector selector = Selector.open(); // 創建一個Selector
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT); // 將Channel注冊到Selector上，設置selectionKey 為 accept， 准備接收新的客戶端連接

            while (true) { // 死循環不斷輪詢，查看 是否有准備就緒的channel
                selector.select();  // 阻塞等到就緒的channel
                Set<SelectionKey> selectionKeys = selector.selectedKeys(); // 獲取到就緒的selectionKeys集合
                selectionKeys.forEach(value -> {
                    try {
                        if(value.isAcceptable()) { // 接收新的客戶端事件
                            ServerSocketChannel channel = (ServerSocketChannel)value.channel(); // 獲取channel
                            SocketChannel clientChannel = channel.accept(); // 獲取客戶端的 socketChannel
                            clientChannel.configureBlocking(false); // 設置為非阻塞
                            String clientId = UUID.randomUUID().toString();
                            System.out.println("客戶端接入" + clientId); 
                            clientMap.put(clientId, clientChannel);
                            clientChannel.register(selector, SelectionKey.OP_READ); // 這里重點說下， 當接收到新的客戶端后，接下來就是准備接收數據，所以這里就是注冊的是Read事件
                                                                                       // 並且這里注冊到selector上的是客戶端對應的SocketChannel, 而不是ServerSocketChannel，
                                                                                       // 因為ServerScoketChannel只負責接收新的客戶端
                        } else if(value.isReadable()) { // 接收到read事件
                            SocketChannel clientChannel = (SocketChannel)value.channel();  // 所以這里是SocketChannel
                            ByteBuffer buffer = ByteBuffer.allocate(1024); // 分配內存
                            int count = clientChannel.read(buffer); // 寫channel中的數據到Buffer中
                            if (count > 0) {
                                buffer.flip(); // 寫完之后，一定要執行flip。轉化成讀 
                                Charset charset = Charset.forName("utf-8");
                                String receiveMsg = String.valueOf(charset.decode(buffer).array());
                                System.out.println("receiveMsg = " +receiveMsg);
                                Iterator<Map.Entry<String, SocketChannel>> it = clientMap.entrySet().iterator();
                                String sendClient = null;
                                while (it.hasNext()) {
                                    Map.Entry<String, SocketChannel> next = it.next();
                                    if(next.getValue() == clientChannel) {
                                        sendClient = next.getKey();
                                        break;
                                    }
                                }
                                it = clientMap.entrySet().iterator();
                                ByteBuffer writeBuffer = ByteBuffer.allocate(1024);
                                while (it.hasNext()) {
                                    SocketChannel socketChannel = it.next().getValue();
                                    writeBuffer.clear();
                                    writeBuffer.put(("sendClient:" + sendClient + "發送了消息").getBytes());
                                    writeBuffer.flip();
                                    socketChannel.write(writeBuffer);
                                }
                            }
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                });
                selectionKeys.clear(); // 每次處理完這一批selectionKeys，一定要清空掉集合。
            }

        } catch (IOException e) {
            e.printStackTrace();
        } finally {
        }
    }

    ok， 上面是我自己的一些理解，如果有問題歡迎大家指正。下一篇，我們將開始學習Netty的源碼。