NIO2.0引入了新的異步通道的概念,並提供了異步文件通道和異步套接字通道的實現。異步通道提供兩種方式獲取獲取操作結果。
- 通過java.util.concurrent.Future類來表示異步操作的結果;
- 在執行異步操作的時候傳入一個java.nio.channels。
CompletionHandler接口的實現類作為操作完成的回調。
NIO2.0的異步套接字通道是真正的異步非阻塞I/O,它對應UNIX網絡編程中的事件驅動I/O(AIO),它不需要通過多路復用器(Selector)對注冊的通道進行輪詢操作即可實現異步讀寫,從而簡化了NIO的編程模型。
服務端代碼示例:
import java.io.IOException; public class TimeServer { public static void main(String[] args) throws IOException { int port = 8080; if (args != null && args.length > 0) { try { port = Integer.valueOf(args[0]); } catch (NumberFormatException e) { // 采用默認值 } } //首先創建異步的時間服務器處理類,然后啟動線程將AsyncTimeServerHandler啟動 AsyncTimeServerHandler timeServer = new AsyncTimeServerHandler(port); new Thread(timeServer, "AIO-AsyncTimeServerHandler-001").start(); } } import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.AsynchronousServerSocketChannel; import java.nio.channels.AsynchronousSocketChannel; import java.nio.channels.CompletionHandler; import java.util.concurrent.CountDownLatch; public class AsyncTimeServerHandler implements Runnable { CountDownLatch latch; AsynchronousServerSocketChannel asynchronousServerSocketChannel; public AsyncTimeServerHandler(int port) { //在構造方法中,我們首先創建一個異步的服務端通道AsynchronousServerSocketChannel, //然后調用它的bind方法綁定監聽端口,如果端口合法且沒被占用,綁定成功,打印啟動成功提示到控制台。 try { asynchronousServerSocketChannel = AsynchronousServerSocketChannel.open(); asynchronousServerSocketChannel.bind(new InetSocketAddress(port)); System.out.println("The time server is start in port : " + port); } catch (IOException e) { e.printStackTrace(); } } @Override public void run() { //在線程的run方法中,初始化CountDownLatch對象, //它的作用是在完成一組正在執行的操作之前,允許當前的線程一直阻塞。 //在本例程中,我們讓線程在此阻塞,防止服務端執行完成退出。 //在實際項目應用中,不需要啟動獨立的線程來處理AsynchronousServerSocketChannel,這里僅僅是個demo演示。 latch = new CountDownLatch(1); doAccept(); try { latch.await(); } catch (InterruptedException e) { e.printStackTrace(); } } //用於接收客戶端的連接,由於是異步操作, //我們可以傳遞一個CompletionHandler<AsynchronousSocketChannel,? super A>類型的handler實例接收accept操作成功的通知消息, //在本例程中我們通過AcceptCompletionHandler實例作為handler來接收通知消息, public void doAccept() { asynchronousServerSocketChannel.accept(this, new CompletionHandler<AsynchronousSocketChannel, AsyncTimeServerHandler>() { @Override public void completed(AsynchronousSocketChannel result, AsyncTimeServerHandler attachment) { //我們從attachment獲取成員變量AsynchronousServerSocketChannel,然后繼續調用它的accept方法。 //在此可能會心存疑惑:既然已經接收客戶端成功了,為什么還要再次調用accept方法呢? //原因是這樣的:當我們調用AsynchronousServerSocketChannel的accept方法后, //如果有新的客戶端連接接入,系統將回調我們傳入的CompletionHandler實例的completed方法, //表示新的客戶端已經接入成功,因為一個AsynchronousServerSocket Channel可以接收成千上萬個客戶端, //所以我們需要繼續調用它的accept方法,接收其他的客戶端連接,最終形成一個循環。 //每當接收一個客戶讀連接成功之后,再異步接收新的客戶端連接。 attachment.asynchronousServerSocketChannel.accept(attachment, this); //鏈路建立成功之后,服務端需要接收客戶端的請求消息, //創建新的ByteBuffer,預分配1M的緩沖區。 ByteBuffer buffer = ByteBuffer.allocate(1024); //通過調用AsynchronousSocketChannel的read方法進行異步讀操作。 //下面我們看看異步read方法的參數。 //ByteBuffer dst:接收緩沖區,用於從異步Channel中讀取數據包; //A attachment:異步Channel攜帶的附件,通知回調的時候作為入參使用; //CompletionHandler<Integer,? super A>:接收通知回調的業務handler,本例程中為ReadCompletionHandler。 result.read(buffer, buffer, new ReadCompletionHandler(result)); } @Override public void failed(Throwable exc, AsyncTimeServerHandler attachment) { exc.printStackTrace(); attachment.latch.countDown(); } }); } } import java.io.IOException; import java.io.UnsupportedEncodingException; import java.nio.ByteBuffer; import java.nio.channels.AsynchronousSocketChannel; import java.nio.channels.CompletionHandler; public class ReadCompletionHandler implements CompletionHandler<Integer, ByteBuffer> { private AsynchronousSocketChannel channel; public ReadCompletionHandler(AsynchronousSocketChannel channel) { //將AsynchronousSocketChannel通過參數傳遞到ReadCompletion Handler中當作成員變量來使用 //主要用於讀取半包消息和發送應答。本例程不對半包讀寫進行具體說明 if (this.channel == null) this.channel = channel; } @Override public void completed(Integer result, ByteBuffer attachment) { //讀取到消息后的處理,首先對attachment進行flip操作,為后續從緩沖區讀取數據做准備。 attachment.flip(); //根據緩沖區的可讀字節數創建byte數組 byte[] body = new byte[attachment.remaining()]; attachment.get(body); try { //通過new String方法創建請求消息,對請求消息進行判斷, //如果是"QUERY TIME ORDER"則獲取當前系統服務器的時間, String req = new String(body, "UTF-8"); System.out.println("The time server receive order : " + req); String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(req) ? new java.util.Date( System.currentTimeMillis()).toString() : "BAD ORDER"; //調用doWrite方法發送給客戶端。 doWrite(currentTime); } catch (UnsupportedEncodingException e) { e.printStackTrace(); } } private void doWrite(String currentTime) { if (currentTime != null && currentTime.trim().length() > 0) { //首先對當前時間進行合法性校驗,如果合法,調用字符串的解碼方法將應答消息編碼成字節數組, //然后將它復制到發送緩沖區writeBuffer中, byte[] bytes = (currentTime).getBytes(); ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length); writeBuffer.put(bytes); writeBuffer.flip(); //最后調用AsynchronousSocketChannel的異步write方法。 //正如前面介紹的異步read方法一樣,它也有三個與read方法相同的參數, //在本例程中我們直接實現write方法的異步回調接口CompletionHandler。 channel.write(writeBuffer, writeBuffer, new CompletionHandler<Integer, ByteBuffer>() { @Override public void completed(Integer result, ByteBuffer buffer) { //對發送的writeBuffer進行判斷,如果還有剩余的字節可寫,說明沒有發送完成,需要繼續發送,直到發送成功。 if (buffer.hasRemaining()) channel.write(buffer, buffer, this); } @Override public void failed(Throwable exc, ByteBuffer attachment) { //關注下failed方法,它的實現很簡單,就是當發生異常的時候,對異常Throwable進行判斷, //如果是I/O異常,就關閉鏈路,釋放資源, //如果是其他異常,按照業務自己的邏輯進行處理,如果沒有發送完成,繼續發送. //本例程作為簡單demo,沒有對異常進行分類判斷,只要發生了讀寫異常,就關閉鏈路,釋放資源。 try { channel.close(); } catch (IOException e) { // ingnore on close } } }); } } @Override public void failed(Throwable exc, ByteBuffer attachment) { try { this.channel.close(); } catch (IOException e) { e.printStackTrace(); } } }
客戶端代碼示例:
public class TimeClient { public static void main(String[] args) { int port = 8080; //通過一個獨立的I/O線程創建異步時間服務器客戶端handler, //在實際項目中,我們不需要獨立的線程創建異步連接對象,因為底層都是通過JDK的系統回調實現的. new Thread(new AsyncTimeClientHandler("127.0.0.1", port), "AIO-AsyncTimeClientHandler-001").start(); } } import java.io.IOException; import java.io.UnsupportedEncodingException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.AsynchronousSocketChannel; import java.nio.channels.CompletionHandler; import java.util.concurrent.CountDownLatch; public class AsyncTimeClientHandler implements CompletionHandler<Void, AsyncTimeClientHandler>, Runnable { private AsynchronousSocketChannel client; private String host; private int port; private CountDownLatch latch; //首先通過AsynchronousSocketChannel的open方法創建一個新的AsynchronousSocketChannel對象。 public AsyncTimeClientHandler(String host, int port) { this.host = host; this.port = port; try { client = AsynchronousSocketChannel.open(); } catch (IOException e) { e.printStackTrace(); } } @Override public void run() { //創建CountDownLatch進行等待,防止異步操作沒有執行完成線程就退出。 latch = new CountDownLatch(1); //通過connect方法發起異步操作,它有兩個參數, //A attachment:AsynchronousSocketChannel的附件,用於回調通知時作為入參被傳遞,調用者可以自定義; //CompletionHandler<Void,? super A> handler:異步操作回調通知接口,由調用者實現。 client.connect(new InetSocketAddress(host, port), this, this); try { latch.await(); } catch (InterruptedException e1) { e1.printStackTrace(); } try { client.close(); } catch (IOException e) { e.printStackTrace(); } } //異步連接成功之后的方法回調——completed方法 @Override public void completed(Void result, AsyncTimeClientHandler attachment) { //創建請求消息體,對其進行編碼,然后復制到發送緩沖區writeBuffer中, //調用Asynchronous SocketChannel的write方法進行異步寫。 //與服務端類似,我們可以實現CompletionHandler <Integer, ByteBuffer>接口用於寫操作完成后的回調。 byte[] req = "QUERY TIME ORDER".getBytes(); ByteBuffer writeBuffer = ByteBuffer.allocate(req.length); writeBuffer.put(req); writeBuffer.flip(); client.write(writeBuffer, writeBuffer, new CompletionHandler<Integer, ByteBuffer>() { @Override public void completed(Integer result, ByteBuffer buffer) { //如果發送緩沖區中仍有尚未發送的字節,將繼續異步發送,如果已經發送完成,則執行異步讀取操作。 if (buffer.hasRemaining()) { client.write(buffer, buffer, this); } else { //客戶端異步讀取時間服務器服務端應答消息的處理邏輯 ByteBuffer readBuffer = ByteBuffer.allocate(1024); //調用AsynchronousSocketChannel的read方法異步讀取服務端的響應消息。 //由於read操作是異步的,所以我們通過內部匿名類實現CompletionHandler<Integer,ByteBuffer>接口, //當讀取完成被JDK回調時,構造應答消息。
client.read(readBuffer,readBuffer, new CompletionHandler<Integer, ByteBuffer>() { @Override public void completed(Integer result,ByteBuffer buffer) { //從CompletionHandler的ByteBuffer中讀取應答消息,然后打印結果。 buffer.flip(); byte[] bytes = new byte[buffer.remaining()]; buffer.get(bytes); String body; try { body = new String(bytes,"UTF-8"); System.out.println("Now is : " + body); latch.countDown(); } catch (UnsupportedEncodingException e) { e.printStackTrace(); } } @Override public void failed(Throwable exc, ByteBuffer attachment) { //當讀取發生異常時,關閉鏈路, //同時調用CountDownLatch的countDown方法讓AsyncTimeClientHandler線程執行完畢,客戶端退出執行。 try { client.close(); latch.countDown(); } catch (IOException e) { // ingnore on close } } }); } } @Override public void failed(Throwable exc, ByteBuffer attachment) { try { client.close(); latch.countDown(); } catch (IOException e) { // ingnore on close } } }); } @Override public void failed(Throwable exc, AsyncTimeClientHandler attachment) { exc.printStackTrace(); try { client.close(); latch.countDown(); } catch (IOException e) { e.printStackTrace(); } } }
需要指出的是,正如之前的NIO例程,我們並沒有完整的處理網絡的半包讀寫,在對例程進行功能測試的時候沒有問題,但是,如果對代碼稍加改造,進行壓力或者性能測試,就會發現輸出結果存在問題。
通過打印線程堆棧的方式看下JDK回調異步Channel CompletionHandler的調用情況:
從“Thread-2”線程堆棧中可以發現,JDK底層通過線程池ThreadPoolExecutor來執行回調通知,異步回調通知類由sun.nio.ch.AsynchronousChannelGroupImpl實現,它經過層層調用,最終回調com.phei.netty.aio.AsyncTimeClientHandler$1.completed方法,完成回調通知。
由此我們也可以得出結論:異步SocketChannel是被動執行對象,我們不需要像NIO編程那樣創建一個獨立的I/O線程來處理讀寫操作。對於AsynchronousServerSocketChannel和AsynchronousSocketChannel,它們都由JDK底層的線程池負責回調並驅動讀寫操作。
正因為如此,基於NIO2.0新的異步非阻塞Channel進行編程比NIO編程更為簡單。