Netty（一）引題

2019-11-14 14:59:24

字體：大中小

來源：轉載

供稿：網友

本文介紹java BIO（同步阻塞IO）,偽異步IO，NIO（非阻塞IO），AIO（異步IO）這四種IO的情況，并對不同IO模型作比較。

1.BIO

2.偽異步IO

3.NIO

4.AIO

5.四種IO比較

6.BIO/偽異步IO/NIO/AIO源碼下載

1.BIO

采用BIO通信模型的服務器,通常由一個獨立的Acceptor線程負責監聽客戶端的連接，它接收到客戶端連接請求后為每個客戶端創建一個新的線程進程鏈路連接處理，處理完后，通過輸出流返回應答給客戶端,線程銷毀。

該模型最大的問題性能問題，當客戶端并發訪問增加后，服務端線程增加，當線程數膨脹后，系統的性能下降，隨著并發量增大，系統會發生線程堆棧溢出、創建新線程失敗等問題，最終導致線程宕機或者僵死,不能對外提供服務。而且開線程有很大的開銷，影響服務器性能。

源碼在src/main/java/NIOInduction/BIO下，分為客戶端和服務端，簡單的網絡、線程的處理。

2.偽異步IO

為了解決同步阻塞IO面臨的一個鏈接需要一個線程處理情況，現在引入了“池”的概念，加入了線程池。

當有新的客戶端連接的時候，將客戶端的Socket封裝為Task（java的Runnable接口實現了）投遞到后端線程池中進行處理。由于線程池可以設置消息隊列的大小和最大線程數，因此它的資源是可控的,無論多少個客戶端并發訪問，都不會導致資源的耗盡和宕機。

偽異步IO通訊框架采用了線程池的實現，因此避免了為每個請求都創建一個獨立的線程造成的線程資源耗盡問題。但是由于它的底層的通信依然采用的同步阻塞模型,因此無法從根本上解決問題。

java輸出流InputStream：當對socket的輸入流進行讀操作時，它會一直阻塞下去，直到發生以下三種事件。

有數據可讀；
可用數據已經讀取完畢；
發生空指針或者IO異常。

這意味著當對方發數據請求或者應答消息緩慢（網絡傳輸慢）時，讀取寫入流一方的通訊線程將長時間阻塞，如果對方要100s才有消息發生完成，讀取的一方的IO線程也會將同步阻塞100s，在此時間里，其他接入消息只能在消息隊列中排隊。

java輸入流OutputStream：當調用OutputStream的write方法寫輸出流的時候，它將會唄阻塞，直到所有要發送的字節全部寫入完畢，或者發生異常。搞過TCP/ip的都曉得，當消息的接收方處理緩慢的時候，將不能及時從TCP緩沖區讀取數據，這將導致發送方的TCP window size不斷減小,直到為0，雙方處于keep-alive狀態，消息發送方就不能再將TCP緩沖區寫入數據，這時采用同步阻塞的IO，write操作將會無限期阻塞,直到tcp window size大于0或者發生IO異常。

源碼在src/main/java/NIOInduction/PseudoAsynchronousIO下，分為客戶端和服務端。客戶端和BIO的客戶端一樣，服務端加入了線程池ExecutorService，相關構造函數請讀者自行查閱。

3.NIO

NIO庫，是在JDK1.4中引入的，NIO彌補了同步阻塞IO的不足。在所有的數據，NIO都是用緩沖區處理掉的（Buffer），任何時候訪問NIO中的數據，都是通過緩沖區進行操作。緩沖區實際就是一個數組。Java NIO的基礎是多路復用器Selector，簡單來說，selector會不斷的輪詢注冊在其上的Channel（通道，全雙工的），如果某個Channel上有新的TCP連接接入、讀寫事件，這個Channel會處于就緒狀態,會被Selector輪詢出來，然后通過SelectionKey可以獲取就緒的select集合，進行后續的IO操作。

一個多路復用器可以同時輪詢多個Channel，而且由于jdk使用了epoll替代了select實現，所以沒有最大連接句柄的限制。（題外話，這里說的eopll、select是說的linux下的IO復用，和select、epoll一樣，清楚流程概念請直接看源碼）。

NIO服務端序列圖

1.打開ServerSocketChannel，用于監聽客戶端的連接，它是所有客戶端連接的父管道。

ServerSocketChannel accptorSvr = ServerSocketChannel.open();

2.綁定監聽端口，設置連接為非阻塞模式。

acceptorSvr.socket().bind(　　new InetSocketAddress(InetAddress.getByName("IP"),port));acceptorSvr.configureBlocking(false);

3.創建Reactor線程，創建多路復用器并啟動線程。

Selector selectot = Selector.open();new Thread(new RectorTask()).start();

4.將SelectSocketChannel注冊到Reactor線程的多路復用器selector上，監聽accept事件。

SelectionKey key = acceptorSvr.register(selector,SelectionKey.OP_ACCEPT,ioHandler);

5.多路復用器在線程run方法中無線循環里輪詢準備就緒的key。

int num = selector.select();Set selectkeys = selector.selectedKeys();Iterator it = selectkeys.iterator();while(it.hasNext){      SelectionKey key = (SelectionKey)it.next;      /*     deal with  IO event    */    }

6.多路復用監聽到有新的用戶接入，處理新的接入請求,完成TCP三次握手，建立物理連接。

SocketChannel sc = ssc.accept();

7.設置客戶端鏈路為非阻塞模式

sc.configureBlocking(false);sc.socket().setReuseAddress(true);...

8.將新接入的客戶端連接注冊到Reactor線程的多路復用器上，監聽讀操作,用來讀取客戶端發送的網絡消息。

SelectionKey key = sc.register(selector,SelectionKey.OP_READ,ioHangler);

9.異步讀取客戶端請求消息到緩沖區

int readNumber = channel.read(receivedBuffer);

10.對bytebuffer進行編解碼,如果有半包消息指針reset，繼續讀取后續的報文，將解碼成功的消息封裝成task，投遞到業務線程池中，進行業務邏輯處理。

Object message = null;        while (buffer.hasRemain()){            byteBuffer.mark();            Object message = decode(byteBuffer);            if(message==null){                byteBuffer.reset();                break;            }            messageList.add(message);        }        if(!byteBuffer.hasRemain()){            byteBuffer.clear();        }        else byteBuffer.compact();        if(messageList!=null & !messageList.isEmpty()) {            for(Object messageF:messageList)                handleTask(messageE);        }

11.將pojo對象encode成bytebuffer，調用SocketChannel的異步write接口,將消息異步發送到客戶端。

socketChannel.wite(buffer);

注意：如果發送區TCP緩沖區滿了，會導致寫半包，此時，需要注冊寫操作位，循環寫，直到整個包消息寫入TCP緩沖區。

NIO客戶端序列圖（大多數和服務端類似）

1.打開SocketChannel，綁定客戶端本地地址（可選，默認系統會隨機分配一個可用的本地地址）

SocketChannel clientChannel = SocketChannel.open();

2.設置SocketChannel為非阻塞模式,同時設置連接的TCP參數。

SocketChannel.configureBlocking(false);
socket.setReuseAddress(true);
socket.setReceiveBufferSize(BUFFER_SIZE);
socket.setSendBufferSize(BUFFER_SIZE);

3.異步連接服務器。

boolean connected = clientChannel.connect(new InetSocketAdress("ip",port));

4.判斷是否連接成功，如果成功，則直接注冊讀狀態位到多路復用器中，如果沒成功（異步連接,返回false，說明客戶端已經已經發送sync包，服務端沒有返回ack包,物理連接還沒建立——關于ack、sync包，請讀者自行查閱TCP/IP中的TCP的三次握手，四次分手的過程）

if(connect)
　　clientChannel.register(selector,SelectionKey.OP_READ,ioHandler);
else
　　clientChannel.register(selector,SelectionKey.OP_CONNECT,ioHandler);

5.向Reactor線程的多路復用器注冊OP_CONNECT狀態位，監聽服務器的TCP ACK應答。

clientChannel.register(selector,SelectionKey.OP_CONNECT,ioHandler);

6.創建Reactor線程，創建多路復用器并啟動線程。

Selector selectot = Selector.open();new Thread(new RectorTask()).start();

7.多路復用器在線程run方法中無線循環里輪詢準備就緒的key。

int num = selector.select();Set selectkeys = selector.selectedKeys();Iterator it = selectkeys.iterator();while(it.hasNext){      SelectionKey key = (SelectionKey)it.next;      /*     deal with  IO event    */    }

8.接收connect事件進行處理

if(key.isConnectable())
　　//handlerConnect();

9.判斷連接結果，如果連接成功，注冊讀事件到多路復用器

if(channel.finishConnect())
　　registerRead();

10.注冊讀事件到多路復用器

clientChannel.register(selector,SelectionKey.OP_READ,ioHandler);

11.異步讀取客戶端請求消息到緩沖區

int readNumber = channel.read(receivedBuffer);

12.對bytebuffer進行編解碼,如果有半包消息指針reset，繼續讀取后續的報文，將解碼成功的消息封裝成task，投遞到業務線程池中，進行業務邏輯處理。

Object message = null;        while (buffer.hasRemain()){            byteBuffer.mark();            Object message = decode(byteBuffer);            if(message==null){                byteBuffer.reset();                break;            }            messageList.add(message);        }        if(!byteBuffer.hasRemain()){            byteBuffer.clear();        }        else byteBuffer.compact();        if(messageList!=null & !messageList.isEmpty()) {            for(Object messageF:messageList)                handleTask(messageE);        }

13.將pojo對象encode成bytebuffer，調用SocketChannel的異步write接口,將消息異步發送到客戶端。