BIO/NIO/AIO的具體實現

阿新 • • 發佈：2018-12-18

1、BIO程式設計

1.1、傳統的BIO程式設計

網路程式設計的基本模型是C/S模型，即兩個程序間的通訊。

服務端提供IP和監聽埠，客戶端通過連線操作想服務端監聽的地址發起連線請求，通過三次握手連線，如果連線成功建立，雙方就可以通過套接字進行通訊。

傳統的同步阻塞模型開發中，ServerSocket負責繫結IP地址，啟動監聽埠；Socket負責發起連線操作。連線成功後，雙方通過輸入和輸出流進行同步阻塞式通訊。

簡單的描述一下BIO的服務端通訊模型：採用BIO通訊模型的服務端，通常由一個獨立的Acceptor執行緒負責監聽客戶端的連線，它接收到客戶端連線請求之後為每個客戶端建立一個新的執行緒進行鏈路處理沒處理完成後，通過輸出流返回應答給客戶端，執行緒銷燬。即典型的一請求一應答通宵模型。

傳統BIO通訊模型圖：

這裡寫圖片描述

該模型最大的問題就是缺乏彈性伸縮能力，當客戶端併發訪問量增加後，服務端的執行緒個數和客戶端併發訪問數呈1:1的正比關係，Java中的執行緒也是比較寶貴的系統資源，執行緒數量快速膨脹後，系統的效能將急劇下降，隨著訪問量的繼續增大，系統最終就死-掉-了。

同步阻塞式I/O建立的Server原始碼：

package com.anxpp.io.calculator.bio;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
/**
* BIO服務端原始碼
* @author yangtao__anxpp.com
* @version 1.0
*/
public final class ServerNormal {
//預設的埠號
private static int DEFAULT_PORT = 12345;
//單例的ServerSocket
private static ServerSocket server;
//根據傳入引數設定監聽埠，如果沒有引數呼叫以下方法並使用預設值
public static void start() throws IOException{
//使用預設值
start(DEFAULT_PORT);
}
//這個方法不會被大量併發訪問，不太需要考慮效率，直接進行方法同步就行了
public synchronized static void start(int port) throws IOException{
if(server != null) return;
try{
//通過建構函式建立ServerSocket
//如果埠合法且空閒，服務端就監聽成功
server = new ServerSocket(port);
System.out.println("伺服器已啟動，埠號：" + port);
//通過無線迴圈監聽客戶端連線
//如果沒有客戶端接入，將阻塞在accept操作上。
while(true){
Socket socket = server.accept();
//當有新的客戶端接入時，會執行下面的程式碼
//然後建立一個新的執行緒處理這條Socket鏈路
new Thread(new ServerHandler(socket)).start();
}
}finally{
//一些必要的清理工作
if(server != null){
System.out.println("伺服器已關閉。");
server.close();
server = null;
}
}
}
}

客戶端訊息處理執行緒ServerHandler原始碼：

package com.anxpp.io.calculator.bio;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;
import com.anxpp.io.utils.Calculator;
/**
* 客戶端執行緒
* @author yangtao__anxpp.com
* 用於處理一個客戶端的Socket鏈路
*/
public class ServerHandler implements Runnable{
private Socket socket;
public ServerHandler(Socket socket) {
this.socket = socket;
}
@Override
public void run() {
BufferedReader in = null;
PrintWriter out = null;
try{
in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
out = new PrintWriter(socket.getOutputStream(),true);
String expression;
String result;
while(true){
//通過BufferedReader讀取一行
//如果已經讀到輸入流尾部，返回null,退出迴圈
//如果得到非空值，就嘗試計算結果並返回
if((expression = in.readLine())==null) break;
System.out.println("伺服器收到訊息：" + expression);
try{
result = Calculator.cal(expression).toString();
}catch(Exception e){
result = "計算錯誤：" + e.getMessage();
}
out.println(result);
}
}catch(Exception e){
e.printStackTrace();
}finally{
//一些必要的清理工作
if(in != null){
try {
in.close();
} catch (IOException e) {
e.printStackTrace();
}
in = null;
}
if(out != null){
out.close();
out = null;
}
if(socket != null){
try {
socket.close();
} catch (IOException e) {
e.printStackTrace();
}
socket = null;
}
}
}
}

同步阻塞式I/O建立的Client原始碼：

package com.anxpp.io.calculator.bio;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;
/**
* 阻塞式I/O建立的客戶端
* @author yangtao__anxpp.com
* @version 1.0
*/
public class Client {
//預設的埠號
private static int DEFAULT_SERVER_PORT = 12345;
private static String DEFAULT_SERVER_IP = "127.0.0.1";
public static void send(String expression){
send(DEFAULT_SERVER_PORT,expression);
}
public static void send(int port,String expression){
System.out.println("算術表示式為：" + expression);
Socket socket = null;
BufferedReader in = null;
PrintWriter out = null;
try{
socket = new Socket(DEFAULT_SERVER_IP,port);
in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
out = new PrintWriter(socket.getOutputStream(),true);
out.println(expression);
System.out.println("___結果為：" + in.readLine());
}catch(Exception e){
e.printStackTrace();
}finally{
//一下必要的清理工作
if(in != null){
try {
in.close();
} catch (IOException e) {
e.printStackTrace();
}
in = null;
}
if(out != null){
out.close();
out = null;
}
if(socket != null){
try {
socket.close();
} catch (IOException e) {
e.printStackTrace();
}
socket = null;
}
}
}
}

測試程式碼，為了方便在控制檯看輸出結果，放到同一個程式（jvm）中執行：

package com.anxpp.io.calculator.bio;
import java.io.IOException;
import java.util.Random;
/**
* 測試方法
* @author yangtao__anxpp.com
* @version 1.0
*/
public class Test {
//測試主方法
public static void main(String[] args) throws InterruptedException {
//執行伺服器
new Thread(new Runnable() {
@Override
public void run() {
try {
ServerBetter.start();
} catch (IOException e) {
e.printStackTrace();
}
}
}).start();
//避免客戶端先於伺服器啟動前執行程式碼
Thread.sleep(100);
//執行客戶端
char operators[] = {'+','-','*','/'};
Random random = new Random(System.currentTimeMillis());
new Thread(new Runnable() {
@SuppressWarnings("static-access")
@Override
public void run() {
while(true){
//隨機產生算術表示式
String expression = random.nextInt(10)+""+operators[random.nextInt(4)]+(random.nextInt(10)+1);
Client.send(expression);
try {
Thread.currentThread().sleep(random.nextInt(1000));
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}).start();
}
}

其中一次的執行結果：

伺服器已啟動，埠號：12345
算術表示式為：4-2
伺服器收到訊息：4-2
___結果為：2
算術表示式為：5-10
伺服器收到訊息：5-10
___結果為：-5
算術表示式為：0-9
伺服器收到訊息：0-9
___結果為：-9
算術表示式為：0+6
伺服器收到訊息：0+6
___結果為：6
算術表示式為：1/6
伺服器收到訊息：1/6
___結果為：0.16666666666666666

從以上程式碼，很容易看出，BIO主要的問題在於每當有一個新的客戶端請求接入時，服務端必須建立一個新的執行緒來處理這條鏈路，在需要滿足高效能、高併發的場景是沒法應用的（大量建立新的執行緒會嚴重影響伺服器效能，甚至罷工）。

1.2、偽非同步I/O程式設計

為了改進這種一連線一執行緒的模型，我們可以使用執行緒池來管理這些執行緒（需要了解更多請參考前面提供的文章），實現1個或多個執行緒處理N個客戶端的模型（但是底層還是使用的同步阻塞I/O），通常被稱為“偽非同步I/O模型“。

偽非同步I/O模型圖：

實現很簡單，我們只需要將新建執行緒的地方，交給執行緒池管理即可，只需要改動剛剛的Server程式碼即可：

package com.anxpp.io.calculator.bio;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* BIO服務端原始碼__偽非同步I/O
* @author yangtao__anxpp.com
* @version 1.0
*/
public final class ServerBetter {
//預設的埠號
private static int DEFAULT_PORT = 12345;
//單例的ServerSocket
private static ServerSocket server;
//執行緒池懶漢式的單例
private static ExecutorService executorService = Executors.newFixedThreadPool(60);
//根據傳入引數設定監聽埠，如果沒有引數呼叫以下方法並使用預設值
public static void start() throws IOException{
//使用預設值
start(DEFAULT_PORT);
}
//這個方法不會被大量併發訪問，不太需要考慮效率，直接進行方法同步就行了
public synchronized static void start(int port) throws IOException{
if(server != null) return;
try{
//通過建構函式建立ServerSocket
//如果埠合法且空閒，服務端就監聽成功
server = new ServerSocket(port);
System.out.println("伺服器已啟動，埠號：" + port);
//通過無線迴圈監聽客戶端連線
//如果沒有客戶端接入，將阻塞在accept操作上。
while(true){
Socket socket = server.accept();
//當有新的客戶端接入時，會執行下面的程式碼
//然後建立一個新的執行緒處理這條Socket鏈路
executorService.execute(new ServerHandler(socket));
}
}finally{
//一些必要的清理工作
if(server != null){
System.out.println("伺服器已關閉。");
server.close();
server = null;
}
}
}
}

測試執行結果是一樣的。

我們知道，如果使用CachedThreadPool執行緒池（不限制執行緒數量，如果不清楚請參考文首提供的文章），其實除了能自動幫我們管理執行緒（複用），看起來也就像是1:1的客戶端：執行緒數模型，而使用FixedThreadPool我們就有效的控制了執行緒的最大數量，保證了系統有限的資源的控制，實現了N:M的偽非同步I/O模型。

但是，正因為限制了執行緒數量，如果發生大量併發請求，超過最大數量的執行緒就只能等待，直到執行緒池中的有空閒的執行緒可以被複用。而對Socket的輸入流就行讀取時，會一直阻塞，直到發生：

有資料可讀
可用資料以及讀取完畢
發生空指標或I/O異常

所以在讀取資料較慢時（比如資料量大、網路傳輸慢等），大量併發的情況下，其他接入的訊息，只能一直等待，這就是最大的弊端。

而後面即將介紹的NIO，就能解決這個難題。

2、NIO 程式設計

JDK 1.4中的java.nio.*包中引入新的Java I/O庫，其目的是提高速度。實際上，“舊”的I/O包已經使用NIO重新實現過，即使我們不顯式的使用NIO程式設計，也能從中受益。速度的提高在檔案I/O和網路I/O中都可能會發生，但本文只討論後者。

2.1、簡介

NIO我們一般認為是New I/O（也是官方的叫法），因為它是相對於老的I/O類庫新增的（其實在JDK 1.4中就已經被引入了，但這個名詞還會繼續用很久，即使它們在現在看來已經是“舊”的了，所以也提示我們在命名時，需要好好考慮），做了很大的改變。但民間跟多人稱之為Non-block I/O，即非阻塞I/O，因為這樣叫，更能體現它的特點。而下文中的NIO，不是指整個新的I/O庫，而是非阻塞I/O。

NIO提供了與傳統BIO模型中的Socket和ServerSocket相對應的SocketChannel和ServerSocketChannel兩種不同的套接字通道實現。

新增的著兩種通道都支援阻塞和非阻塞兩種模式。

阻塞模式使用就像傳統中的支援一樣，比較簡單，但是效能和可靠性都不好；非阻塞模式正好與之相反。

對於低負載、低併發的應用程式，可以使用同步阻塞I/O來提升開發速率和更好的維護性；對於高負載、高併發的（網路）應用，應使用NIO的非阻塞模式來開發。

這裡寫圖片描述

下面會先對基礎知識進行介紹。

2.2、緩衝區 Buffer

Buffer是一個物件，包含一些要寫入或者讀出的資料。在NIO庫中，所有資料都是用緩衝區處理的。在讀取資料時，它是直接讀到緩衝區中的；在寫入資料時，也是寫入到緩衝區中。任何時候訪問NIO中的資料，都是通過緩衝區進行操作。

緩衝區實際上是一個數組，並提供了對資料結構化訪問以及維護讀寫位置等資訊。

具體的快取區有這些：ByteBuffe、CharBuffer、 ShortBuffer、IntBuffer、LongBuffer、FloatBuffer、DoubleBuffer。他們實現了相同的介面：Buffer。

Buffer有兩種工作模式：寫模式和讀模式。在讀模式下，應用程式只能從Buffer中讀取資料，不能進行寫操作。但是在寫模式下，應用程式是可以進行讀操作的，這就表示可能會出現髒讀的情況。所以一旦您決定要從Buffer中讀取資料，一定要將Buffer的狀態改為讀模式。

這裡寫圖片描述

2.3、通道 Channel

我們對資料的讀取和寫入要通過Channel，它就像水管一樣，是一個通道。通道不同於流的地方就是通道是雙向的，可以用於讀、寫和同時讀寫操作。

底層的作業系統的通道一般都是全雙工的，所以全雙工的Channel比流能更好的對映底層作業系統的API。

Channel主要分兩大類：

SelectableChannel：使用者網路讀寫
FileChannel：用於檔案操作

後面程式碼會涉及的ServerSocketChannel和SocketChannel都是SelectableChannel的子類。所有被Selector（選擇器）註冊的通道，只能是繼承了SelectableChannel類的子類。ServerSocketChannel同時支援UDP協議和TCP協議，SocketChannel支援TCP協議，DatagramChannel支援UDP報文。

這裡寫圖片描述

2.4、多路複用器 Selector

Selector是Java NIO 程式設計的基礎。

Selector提供選擇已經就緒的任務的能力：Selector會不斷輪詢註冊在其上的Channel，如果某個Channel上面發生讀或者寫事件，這個Channel就處於就緒狀態，會被Selector輪詢出來，然後通過SelectionKey可以獲取就緒Channel的集合，進行後續的I/O操作。

一個Selector可以同時輪詢多個Channel，因為JDK使用了epoll()代替傳統的select實現，所以沒有最大連線控制代碼1024/2048的限制。所以，只需要一個執行緒負責Selector的輪詢，就可以接入成千上萬的客戶端。

應用程式將向Selector物件註冊需要它關注的Channel，以及具體的某一個Channel會對哪些IO事件感興趣。Selector中也會維護一個“已經註冊的Channel”的容器。以下程式碼來自WindowsSelectorImpl實現類中，對已經註冊的Channel的管理容器：

// Initial capacity of the poll array
private final int INIT_CAP = 8;
// Maximum number of sockets for select().
// Should be INIT_CAP times a power of 2
private final static int MAX_SELECTABLE_FDS = 1024;
// The list of SelectableChannels serviced by this Selector. Every mod
// MAX_SELECTABLE_FDS entry is bogus, to align this array with the poll
// array, where the corresponding entry is occupied by the wakeupSocket
private SelectionKeyImpl[] channelArray = new SelectionKeyImpl[INIT_CAP];

多路複用IO技術是需要作業系統進行支援的，其特點就是作業系統可以同時掃描同一個埠上不同網路連線的時間。所以作為上層的JVM，必須要為不同作業系統的多路複用IO實現編寫不同的程式碼。Windows環境下對應的實現類是sun.nio.ch.WindowsSelectorImpl：

2.5、NIO服務端

程式碼比傳統的Socket程式設計看起來要複雜不少。

直接貼程式碼吧，以註釋的形式給出程式碼說明。

NIO建立的Server原始碼：

package com.anxpp.io.calculator.nio;
public class Server {
private static int DEFAULT_PORT = 12345;
private static ServerHandle serverHandle;
public static void start(){
start(DEFAULT_PORT);
}
public static synchronized void start(int port){
if(serverHandle!=null)
serverHandle.stop();
serverHandle = new ServerHandle(port);
new Thread(serverHandle,"Server").start();
}
public static void main(String[] args){
start();
}
}

ServerHandle：

package com.anxpp.io.calculator.nio;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import com.anxpp.io.utils.Calculator;
/**
* NIO服務端
* @author yangtao__anxpp.com
* @version 1.0
*/
public class ServerHandle implements Runnable{
private Selector selector;
private ServerSocketChannel serverChannel;
private volatile boolean started;
/**
* 構造方法
* @param port 指定要監聽的埠號
*/
public ServerHandle(int port) {
try{
//建立選擇器
selector = Selector.open();
//開啟監聽通道
serverChannel = ServerSocketChannel.open();
//如果為 true，則此通道將被置於阻塞模式；如果為 false，則此通道將被置於非阻塞模式
serverChannel.configureBlocking(false);//開啟非阻塞模式
//繫結埠 backlog設為1024
serverChannel.socket().bind(new InetSocketAddress(port),1024);
//監聽客戶端連線請求
serverChannel.register(selector, SelectionKey.OP_ACCEPT);
//標記伺服器已開啟
started = true;
System.out.println("伺服器已啟動，埠號：" + port);
}catch(IOException e){
e.printStackTrace();
System.exit(1);
}
}
public void stop(){
started = false;
}
@Override
public void run() {
//迴圈遍歷selector
while(started){
try{
//無論是否有讀寫事件發生，selector每隔1s被喚醒一次
selector.select(1000);
//阻塞,只有當至少一個註冊的事件發生的時候才會繼續.
// selector.select();
Set<SelectionKey> keys = selector.selectedKeys();
Iterator<SelectionKey> it = keys.iterator();
SelectionKey key = null;
while(it.hasNext()){
key = it.next();
it.remove();
try{
handleInput(key);
}catch(Exception e){
if(key != null){
key.cancel();
if(key.channel() != null){
key.channel().close();
}
}
}
}
}catch(Throwable t){
t.printStackTrace();
}
}
//selector關閉後會自動釋放裡面管理的資源
if(selector != null)
try{
selector.close();
}catch (Exception e) {
e.printStackTrace();
}
}
private void handleInput(SelectionKey key) throws IOException{
if(key.isValid()){
//處理新接入的請求訊息
if(key.isAcceptable()){
ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
//通過ServerSocketChannel的accept建立SocketChannel例項
//完成該操作意味著完成TCP三次握手，TCP物理鏈路正式建立
SocketChannel sc = ssc.accept();
//設定為非阻塞的
sc.configureBlocking(false);
//註冊為讀
sc.register(selector, SelectionKey.OP_READ);
}
//讀訊息
if(key.isReadable()){
SocketChannel sc = (SocketChannel) key.channel();
//建立ByteBuffer，並開闢一個1M的緩衝區
ByteBuffer buffer = ByteBuffer.allocate(1024);
//讀取請求碼流，返回讀取到的位元組數
int readBytes = sc.read(buffer);
//讀取到位元組，對位元組進行編解碼
if(readBytes>0){
//將緩衝區當前的limit設定為position=0，用於後續對緩衝區的讀取操作
buffer.flip();
//根據緩衝區可讀位元組數建立位元組陣列
byte[] bytes = new byte[buffer.remaining()];
//將緩衝區可讀位元組陣列複製到新建的陣列中
buffer.get(bytes);
String expression = new String(bytes,"UTF-8");
System.out.println("伺服器收到訊息：" + expression);
//處理資料
String result = null;
try{
result = Calculator.cal(expression).toString();
}catch(Exception e){
result = "計算錯誤：" + e.getMessage();
}
//傳送應答訊息
doWrite(sc,result);
}
//沒有讀取到位元組忽略
// else if(readBytes==0);
//鏈路已經關閉，釋放資源
else if(readBytes<0){
key.cancel();
sc.close();
}
}
}
}
//非同步傳送應答訊息
private void doWrite(SocketChannel channel,String response) throws IOException{
//將訊息編碼為位元組陣列
byte[] bytes = response.getBytes();
//根據陣列容量建立ByteBuffer
ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
//將位元組陣列複製到緩衝區
writeBuffer.put(bytes);
//flip操作
writeBuffer.flip();
//傳送緩衝區的位元組陣列
channel.write(writeBuffer);
//****此處不含處理“寫半包”的程式碼
}
}

可以看到，建立NIO服務端的主要步驟如下：

開啟ServerSocketChannel，監聽客戶端連線

繫結監聽埠，設定連線為非阻塞模式

建立Reactor執行緒，建立多路複用器並啟動執行緒

將ServerSocketChannel註冊到Reactor執行緒中的Selector上，監聽ACCEPT事件

Selector輪詢準備就緒的key

Selector監聽到新的客戶端接入，處理新的接入請求，完成TCP三次握手，簡歷物理鏈路

設定客戶端鏈路為非阻塞模式

將新接入的客戶端連線註冊到Reactor執行緒的Selector上，監聽讀操作，讀取客戶端傳送的網路訊息

非同步讀取客戶端訊息到緩衝區

對Buffer編解碼，處理半包訊息，將解碼成功的訊息封裝成Task

將應答訊息編碼為Buffer，呼叫SocketChannel的write將訊息非同步傳送給客戶端

因為應答訊息的傳送，SocketChannel也是非同步非阻塞的，所以不能保證一次能吧需要傳送的資料傳送完，此時就會出現寫半包的問題。我們需要註冊寫操作，不斷輪詢Selector將沒有傳送完的訊息傳送完畢，然後通過Buffer的hasRemain()方法判斷訊息是否傳送完成。

2.6、NIO客戶端

還是直接上程式碼吧，過程也不需要太多解釋了，跟服務端程式碼有點類似。

Client：

package com.anxpp.io.calculator.nio;
public class Client {
private static String DEFAULT_HOST = "127.0.0.1";
private static int DEFAULT_PORT = 12345;
private static ClientHandle clientHandle;
public static void start(){
start(DEFAULT_HOST,DEFAULT_PORT);
}
public static synchronized void start(String ip,int port){
if(clientHandle!=null)
clientHandle.stop();
clientHandle = new ClientHandle(ip,port);
new Thread(clientHandle,"Server").start();
}
//向伺服器傳送訊息
public static boolean sendMsg(String msg) throws Exception{
if(msg.equals("q")) return false;
clientHandle.sendMsg(msg);
return true;
}
public static void main(String[] args){
start();
}
}

ClientHandle：

package com.anxpp.io.calculator.nio;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
/**
* NIO客戶端
* @author yangtao__anxpp.com
* @version 1.0
*/
public class ClientHandle implements Runnable{
private String host;
private int port;
private Selector selector;
private SocketChannel socketChannel;
private volatile boolean started;
public ClientHandle(String ip,int port) {
this.host = ip;
this.port = port;
try{
//建立選擇器
selector = Selector.open();
//開啟監聽通道
socketChannel = SocketChannel.open();
//如果為 true，則此通道將被置於阻塞模式；如果為 false，則此通道將被置於非阻塞模式
socketChannel.configureBlocking(false);//開啟非阻塞模式
started = true;
}catch(IOException e){
e.printStackTrace();
System.exit(1);
}
}
public void stop(){
started = false;
}
@Override
public void run() {
try{
doConnect();
}catch(IOException e){
e.printStackTrace();
System.exit(1);
}
//迴圈遍歷selector
while(started){
try{
//無論是否有讀寫事件發生，selector每隔1s被喚醒一次
selector.select(1000);
//阻塞,只有當至少一個註冊的事件發生的時候才會繼續.
// selector.select();
Set<SelectionKey> keys = selector.selectedKeys();
Iterator<SelectionKey> it = keys.iterator();
SelectionKey key = null;
while(it.hasNext()){
key = it.next();
it.remove();
try{
handleInput(key);
}catch(Exception e){
if(key != null){
key.cancel();
if(key.channel() != null){
key.channel().close();
}
}
}
}
}catch(Exception e){
e.printStackTrace();
System.exit(1);
}
}
//selector關閉後會自動釋放裡面管理的資源
if(selector != null)
try{
selector.close();
}catch (Exception e) {
e.printStackTrace();
}
}
private void handleInput(SelectionKey key) throws IOException{
if(key.isValid()){
SocketChannel sc = (SocketChannel) key.channel();
if(key.isConnectable()){
if(sc.finishConnect());
else System.exit(1);
}
//讀訊息
if(key.isReadable()){
//建立ByteBuffer，並開闢一個1M的緩衝區
ByteBuffer buffer = ByteBuffer.allocate(1024);
//讀取請求碼流，返回讀取到的位元組數
int readBytes = sc.read(buffer);
//讀取到位元組，對位元組進行編解碼
if(readBytes>0){
//將緩衝區當前的limit設定為position=0，用於後續對緩衝區的讀取操作
buffer.flip();
//根據緩衝區可讀位元組數建立位元組陣列
byte[] bytes = new byte[buffer.remaining()];
//將緩衝區可讀位元組陣列複製到新建的陣列中
buffer.get(bytes);
String result = new String(bytes,"UTF-8");
System.out.println("客戶端收到訊息：" + result);
}
//沒有讀取到位元組忽略
// else if(readBytes==0);
//鏈路已經關閉，釋放資源
else if(readBytes<0){
key.cancel();
sc.close();
}
}
}
}
//非同步傳送訊息
private void doWrite(SocketChannel channel,String request) throws IOException{
//將訊息編碼為位元組陣列
byte[] bytes = request.getBytes();
//根據陣列容量建立ByteBuffer
ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
//將位元組陣列複製到緩衝區
writeBuffer.put(bytes);
//flip操作
writeBuffer.flip();
//傳送緩衝區的位元組陣列
channel.write(writeBuffer);
//****此處不含處理“寫半包”的程式碼
}
private void doConnect() throws IOException{
if(socketChannel.connect(new InetSocketAddress(host,port)));
else socketChannel.register(selector, SelectionKey.OP_CONNECT);
}
public void sendMsg(String msg) throws Exception{
socketChannel.register(selector, SelectionKey.OP_READ);
doWrite(socketChannel, msg);
}
}

2.7、演示結果

首先執行伺服器，順便也執行一個客戶端：

package com.anxpp.io.calculator.nio;
import java.util.Scanner;
/**
* 測試方法
* @author yangtao__anxpp.com
* @version 1.0
*/
public class Test {
//測試主方法
@SuppressWarnings("resource")
public static void main(String[] args) throws Exception{
//執行伺服器
Server.start();
//避免客戶端先於伺服器啟動前執行程式碼
Thread.sleep(100);
//執行客戶端
Client.start();
while(Client.sendMsg(new Scanner(System.in).nextLine()));
}
}

我們也可以單獨執行客戶端，效果都是一樣的。

一次測試的結果：

伺服器已啟動，埠號：12345
1+2+3+4+5+6
伺服器收到訊息：1+2+3+4+5+6
客戶端收到訊息：21
1*2/3-4+5*6/7-8
伺服器收到訊息：1*2/3-4+5*6/7-8
客戶端收到訊息：-7.0476190476190474

3、AIO程式設計

NIO 2.0引入了新的非同步通道的概念，並提供了非同步檔案通道和非同步套接字通道的實現。

非同步的套接字通道時真正的非同步非阻塞I/O，對應於UNIX網路程式設計中的事件驅動I/O（AIO）。他不需要過多的Selector對註冊的通道進行輪詢即可實現非同步讀寫，從而簡化了NIO的程式設計模型。

非同步IO則是採用“訂閱-通知”模式：即應用程式向作業系統註冊IO監聽，然後繼續做自己的事情。當作業系統發生IO事件，並且準備好資料後，在主動通知應用程式，觸發相應的函式：

這裡寫圖片描述

和同步IO一樣，非同步IO也是由作業系統進行支援的。微軟的windows系統提供了一種非同步IO技術：IOCP（I/O Completion Port，I/O完成埠）（Windows下負責實現套接字通道的具體類是“sun.nio.ch.WindowsAsynchronousSocketChannelImpl”），Linux下由於沒有這種非同步IO技術，所以使用的是epoll（上文介紹過的一種多路複用IO技術的實現）對非同步IO進行模擬。

在JAVA NIO框架中，我們說到了一個重要概念“selector”（選擇器）。它負責代替應用查詢中所有已註冊的通道到作業系統中進行IO事件輪詢、管理當前註冊的通道集合，定位發生事件的通道等操操作；但是在JAVA AIO框架中，由於應用程式不是“輪詢”方式，而是訂閱-通知方式，所以不再需要“selector”（選擇器）了，改由channel通道直接到作業系統註冊監聽。JAVA NIO和JAVA AIO框架，除了因為作業系統的實現不一樣而去掉了Selector外，其他的重要概念都是存在的，例如上文中提到的Channel的概念，還有演示程式碼中使用的Buffer快取方式。

JAVA AIO框架中，只實現了兩種網路IO通道“AsynchronousServerSocketChannel”（伺服器監聽通道）、“AsynchronousSocketChannel”（socket套接字通道）。但是無論哪種通道他們都有獨立的fileDescriptor（檔案識別符號）、attachment（附件，附件可以使任意物件，類似“通道上下文”），並被獨立的SocketChannelReadHandle類例項引用。

直接上程式碼吧。

3.1、Server端程式碼

Server：

package com.anxpp.io.calculator.aio.server;
/**
* AIO服務端
* @author yangtao__anxpp.com
* @version 1.0
*/
public class Server {
private static int DEFAULT_PORT = 12345;
private static AsyncServerHandler serverHandle;
public volatile static long clientCount = 0;
public static void start(){
start(DEFAULT_PORT);
}
public static synchronized void start(int port){
if(serverHandle!=null)
return;
serverHandle = new AsyncServerHandler(port);
new Thread(serverHandle,"Server").start();
}
public static void main(String[] args){
Server.start();
}
}

AsyncServerHandler：

package com.anxpp.io.calculator.aio.server;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.util.concurrent.CountDownLatch;
public class AsyncServerHandler implements Runnable {
public CountDownLatch latch;
public AsynchronousServerSocketChannel channel;
public AsyncServerHandler(int port) {
try {
//建立服務端通道
channel = AsynchronousServerSocketChannel.open();
//繫結埠
channel.bind(new InetSocketAddress(port));
System.out.println("伺服器已啟動，埠號：" + port);
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void run() {
//CountDownLatch初始化
//它的作用：在完成一組正在執行的操作之前，允許當前的現場一直阻塞
//此處，讓現場在此阻塞，防止服務端執行完成後退出
//也可以使用while(true)+sleep
//生成環境就不需要擔心這個問題，以為服務端是不會退出的
latch = new CountDownLatch(1);
//用於接收客戶端的連線
channel.accept(this,new AcceptHandler());
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

AcceptHandler：

package com.anxpp.io.calculator.aio.server;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
//作為handler接收客戶端連線
public class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel, AsyncServerHandler> {
@Override
public void completed(AsynchronousSocketChannel channel,AsyncServerHandler serverHandler) {
//繼續接受其他客戶端的請求
Server.clientCount++;
System.out.println("連線的客戶端數：" + Server.clientCount);
serverHandler.channel.accept(serverHandler, this);
//建立新的Buffer
ByteBuffer buffer = ByteBuffer.allocate(1024);
//非同步讀第三個引數為接收訊息回撥的業務Handler
channel.read(buffer, buffer, new ReadHandler(channel));
}
@Override
public void failed(Throwable exc, AsyncServerHandler serverHandler) {
exc.printStackTrace();
serverHandler.latch.countDown();
}
}

ReadHandler：

package com.anxpp.io.calculator.aio.server;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import com.anxpp.io.utils.Calculator;
public class ReadHandler implements CompletionHandler<Integer, ByteBuffer> {
//用於讀取半包訊息和傳送應答
private AsynchronousSocketChannel channel;
public ReadHandler(AsynchronousSocketChannel channel) {
this.channel = channel;
}
//讀取到訊息後的處理
@Override
public void completed(Integer result, ByteBuffer attachment) {
//flip操作
attachment.flip();
//根據
byte[] message = new byte[attachment.remaining()];
attachment.get(message);
try {
String expression = new String(message, "UTF-8");
System.out.println("伺服器收到訊息: " + expression);
String calrResult = null;
try{
calrResult = Calculator.cal(expression).toString();
}catch(Exception e){
calrResult = "計算錯誤：" + e.getMessage();
}
//向客戶端傳送訊息
doWrite(calrResult);
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
//傳送訊息
private void doWrite(String result) {
byte[] bytes = result.getBytes();
ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
writeBuffer.put(bytes);
writeBuffer.flip();
//非同步寫資料引數與前面的read一樣
channel.write(writeBuffer, writeBuffer,new CompletionHandler<Integer, ByteBuffer>() {
@Override
public void completed(Integer result, ByteBuffer buffer) {
//如果沒有傳送完，就繼續傳送直到完成
if (buffer.hasRemaining())
channel.write(buffer, buffer, this);
else{
//建立新的Buffer
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
//非同步讀第三個引數為接收訊息回撥的業務Handler
channel.read(readBuffer, readBuffer, new ReadHandler(channel));
}
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
try {
channel.close();
} catch (IOException e) {
}
}
});
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
try {
this.channel.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}

OK，這樣就已經完成了，其實說起來也簡單，雖然程式碼感覺很多，但是API比NIO的使用起來真的簡單多了，主要就是監聽、讀、寫等各種CompletionHandler。此處本應有一個WriteHandler的，確實，我們在ReadHandler中，以一個匿名內部類實現了它。

下面看客戶端程式碼。

3.2、Client端程式碼

Client：

package com.anxpp.io.calculator.aio.client;
import java.util.Scanner;
public class Client {
private static String DEFAULT_HOST = "127.0.0.1";
private static int DEFAULT_PORT = 12345;
private static AsyncClientHandler clientHandle;
public static void start(){
st

 
 
              
           
              
              
            
            相關推薦
			   
            
            
            
 

    

    
    BIO/NIO/AIO的具體實現
      
                1、BIO程式設計

    1.1、傳統的BIO程式設計



    網路程式設計的基本模型是C/S模型，即兩個程序間的通訊。

    服務端提供IP和監聽埠，客戶端通過連線操作想服務端監聽的地址發起連線請求，通過三次握手連線，如果連線成功建立，雙方就可以通過套接字進行 

  
 

    

    
    JAVA BIO,NIO,AIO詳解（附程式碼實現）
       
 
  
  
 這幾天在看面試的東西，可能是自己比較笨，花了快兩天的時間才理清楚。特此記錄一下。  首先我們要理解的一個很重要概念是，客戶端連線和傳送資料是分開的，連線不代表立馬會傳輸資料。  說說BIO,NIO,AIO到底是什麼東西  BIO：同步堵塞  NIO：非同步堵塞  AIO：非同步非堵塞   

  
 

    

    
    BIO--NIO-AIO（IOCP在Java7中的實現）
      
                
背景
   在 Java 的早期，JVM 在解釋位元組碼時往往很少或沒有執行時優化。這就意味著，Java 程式往往拖得很長，其執行速率大大低於本地編譯程式碼，因而對作業系統I/O 子系統的要求並不太高。 如今在執行時優化方面，JVM 已然前進了一大步。現在 JVM 執行位元 

  
 

    

    
    java並發之bio nio aio
      image   進行   nbsp   ID   str   系統   同步非阻塞   src   lin   　　最近在進行tomcat優化，發現tomcat connector並發支持bio nio apr，發現想要理解tomcat並發離不開java io的理解。所有本文先探討java對io的支持。jav 

  
 

    

    
    Netty序章之BIO NIO AIO演變
      升級版   port   amr   gist   文件操作   -i   cti   升級   channel   Netty序章之BIO NIO AIO演變
Netty是一個提供異步事件驅動的網絡應用框架，用以快速開發高性能、高可靠的網絡服務器和客戶端程序。Netty簡化了網絡程序的開發，是很多框架和公司 

  
 

    

    
    JAVA IO : BIO NIO AIO
       
  
  
 
 
  JAVA IO : BIO NIO AIO
  
   同步非同步、阻塞非阻塞概念
   
    同步與非同步
    阻塞與非阻塞
   
   IO VS NIO VS AIO
   
    面向流與面向緩衝
    阻塞與非阻塞IO
   
   BIO、NIO、AIO的 

  
 

    

    
    【J2SE】JAVAIO 流BIO,NIO,AIO
       
 
 javaIO程式設計. 
    隨著java版本的不斷升級與迭代，java的IO模型開始得到改變。從原始的BIO,到1.4以後釋出的NIO,再到對NIO進行的改進AIO分別對IO模型做了優化，BIO是同步，阻塞的IO. 
 NIO是同步，非阻塞的IO,AIO是非同步非 

  
 

    

    
    java bio nio aio
      
                流程圖：



BIO

BIO 全稱Block-IO 是一種阻塞同步的通訊模式。我們常說的Stock IO 一般指的是BIO。是一個比較傳統的通訊方式，模式簡單，使用方便。但併發處理能力低，通訊耗時，依賴網速。
BIO 設計原理：
伺服器通過一個Acceptor執行緒負責監 

  
 

    

    
    JAVA 004 網絡編程 BIO NIO AIO
      基於   以及   技術分享   同步   linu   讀取   介紹   inux   應用程序   目錄（圖片來自於網絡）
　　多路復用Linux環境下底層機制
　　多路復用模式Reacotor和Proactor
　　BIO,NIO,AIO的簡單介紹
 
多路復用Linux環境下底層機制　

 
多路復 

  
 

    

    
    bio,nio,aio的理解
      
                先上幾個圖

BIO：執行緒發起IO請求，不管核心是否準備好IO操作，從發起請求起，執行緒一直阻塞，直到操作完成。如下圖：











NIO(reactor模型)：執行緒發起IO請求，立即返回；核心在做好IO操作的準備之後，通過呼叫註冊的回撥函式通知執行緒做IO操 

  
 

    

    
    伺服器最大TCP連線數及調優彙總 BIO,NIO,AIO的理解
      啟動執行緒數： 
啟動執行緒數=【任務執行時間/（任務執行時間-IO等待時間）】*CPU核心數 
最佳啟動執行緒數和CPU核心數量成正比，和IO阻塞時間成反比。如果任務都是CPU計算型任務，那麼執行緒數最多不超過CPU核心數，因為啟動再多執行緒，CPU也來不及排程；相反如果是任務需要等待磁碟操作，網路響應，那 

  
 

    

    
    JAVA中的BIO,NIO,AIO
       
 
 在瞭解BIO,NIO,AIO之前先了解一下IO的幾個概念: 
 　　1.同步 
 　　　　　　使用者程序觸發IO操作並等待或者輪詢的去檢視IO操作是否就緒， 例如自己親自出馬持銀行卡到銀行取錢 
 　　2.非同步 
 　　　　　　使用者觸發IO操作以後,可以幹別的事，IO操作完成以後再通知當前執行緒 

  
 

    

    
    非同步和非阻塞一樣嗎? (內容涉及BIO,NIO,AIO,Netty)
      
							
							
							寫在開頭：

本文不是純講技術，會涉及概念理解和語義分析的要點。正如博主標題說的，高效的學習方法比擁有多少知識重要。高效的學習方法途徑之一就是閱讀理解，理解能力越好，掌握越快，知識體系更清晰。




  如果剛入門Netty，看過Netty執行緒模型，又看過J 

  
 

    

    
    IO: BIO ? NIO ? AIO?
       
 
 
 IO的方式通常分為幾種，同步阻塞的BIO、同步非阻塞的NIO、非同步非阻塞的AIO。 
 一、BIO 
      在JDK1.4出來之前，我們建立網路連線的時候採用BIO模式，需要先在服務端啟動一個ServerSocket，然後在客戶端啟動So 

  
 

    

    
    java的BIO,NIO,AIO的區別和用法
      
                
java網路io程式設計，從傳統的BIO（同步阻塞）到NIO（同步非阻塞）再到AIO（非同步非阻塞）.

場景：客戶端想服務端傳送請求，服務端會為每個客戶端建立一個執行緒來響應，問題來了，如果客戶端出現了延時等異常，多是網路異常，這樣的話，服務端的為它建立的執行緒，一直處於 

  
 

    

    
    JAVA 中BIO,NIO,AIO的理解
      
                【轉自】http://qindongliang.iteye.com/blog/2018539 在高效能的IO體系設計中，有幾個名詞概念常常會使我們感到迷惑不解。具體如下： 序號	問題 1	什麼是同步？ 2	什麼是非同步？ 3	什麼是阻塞？ 4	什麼是非阻塞？ 5	什麼是同步阻 

  
 

    

    
    BIO,NIO,AIO及netty的介紹
      
                
同步阻塞io（BIO）、偽非同步io、非阻塞io(NIO)、非同步io(AIO)的概念及區別？
BIO：同步阻塞式IO，伺服器端與客戶端通過三次握手後建立連線，連線成功，雙方通過I/O進行同步阻塞式通訊。
弊端：1，讀和寫操作是同步阻塞的，任何一端出現網路效能問題，都會影響 

  
 

    

    
    BIO,NIO,AIO整理（全稱，介紹，區別）
      
                先來個例子理解一下概念，以銀行取款為例：同步 ： 自己親自出馬持銀行卡到銀行取錢（使用同步IO時，Java自己處理IO讀寫）。非同步 ： 委託一小弟拿銀行卡到銀行取錢，然後給你（使用非同步IO時，Java將IO讀寫委託給OS處理，需要將資料緩衝區地址和大小傳給OS(銀行卡和密 

  
 

    

    
    BIO,NIO,AIO
      模型   並行   系統   dwr   con   通道   獲得   ant   ring    一：事件分離器
        在IO讀寫時，把 IO請求 與 讀寫操作 分離調配進行，需要用到事件分離器。根據處理機制的不同，事件分離器又分為：同步的Reactor和異步的Proactor。
        

  
 

    

    
    Java BIO, NIO, AIO 簡單理解
      
                
1， 同步 VS. 非同步
比較的是當某種事件發生時呼叫處理函式的方式，
同步：應用程式主動呼叫處理函式，
非同步：作業系統呼叫應用程式的處理函式（也叫回調函式），對應用程式而言是被動過程。

2， 阻塞 VS. 非阻塞
阻塞：應用程式發起IO請求，IO請求（假設是網路上的