NIO 2.0引入了新的異步通道的概念,并提供了異步檔案通道和異步套接字通道的實作。異步通道提供一下兩種方式擷取操作結果。
- 通過java.util.concurrent.Future類來表示異步操作的結果;
-
在執行異步操作的時候傳入一個java.nio.channels。
CompletionHandler接口的實作類作為操作完成的回調
NIO 2.0的異步套接字通道是真正的異步非阻塞I/O,對應于UNIX網絡程式設計中的事件驅動I/O(AIO)。它不需要通過多路複用器(Selector)對注冊的通道進行輪詢操作即可實作異步讀寫,進而簡化了NIO的程式設計模型。
簡單的說,NIO1.0 和NIO2.0 相比最大的差別是NIO1.0通過多路複用器輪詢來擷取讀寫事件,然後執行具體的操作,NIO2.0采用和Future和回調,使用上更加簡潔。
下面是一個例子,用戶端向服務端發一個時間查詢消息,服務端傳回查詢的結果。
eg
服務端:
public class TimeServer {
public static void main(String[] args) {
AsyncTimeServerHandler asyncTimeServerHandler=new AsyncTimeServerHandler(8080);
new Thread(asyncTimeServerHandler,"AI0-SERVER").start();
}
}
public class AsyncTimeServerHandler implements Runnable {
private int port;
CountDownLatch latch;
AsynchronousServerSocketChannel asynchronousServerSocketChannel;
AsyncTimeServerHandler(int port) {
this.port = port;
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() {
latch = new CountDownLatch(1);
//接收用戶端請求
doAccept();
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public void doAccept() {
//注冊接收用戶端請求,接收到請求會回調acceptCompletionHandler類
asynchronousServerSocketChannel.accept(this, new AcceptCompletionHandler());
}
}
public class AcceptCompletionHandler implements CompletionHandler<AsynchronousSocketChannel, AsyncTimeServerHandler> {
@Override
public void completed(AsynchronousSocketChannel result, AsyncTimeServerHandler attachment) {
//接收到用戶端請求後,在注冊接收請求,用于接收其他用戶端的請求
attachment.asynchronousServerSocketChannel.accept(attachment, this);
ByteBuffer buffer = ByteBuffer.allocate(1024);
//監聽該用戶端連接配接上的讀請求,如果接收到消息,會調用readCompletionHandler
result.read(buffer, buffer, new ReadCompletionHandler(result));
}
@Override
public void failed(Throwable exc, AsyncTimeServerHandler attachment) {
//接收失敗的回調
exc.printStackTrace();
attachment.latch.countDown();
}
}
public class ReadCompletionHandler implements CompletionHandler<Integer,ByteBuffer>{
private AsynchronousSocketChannel channel;
public ReadCompletionHandler(AsynchronousSocketChannel channel){
this.channel=channel;
}
@Override
public void completed(Integer result, ByteBuffer attachment) {
//讀取用戶端請求
attachment.flip();
byte[] body=new byte[attachment.remaining()];
attachment.get(body);
try {
String req=new String(body,"UTF-8");
System.out.println("The time server receive order:"+req);
String currentTime="QUERY TIME ORDER".equalsIgnoreCase(req)?
new Date(System.currentTimeMillis()).toString():"BAD ORDER";
//回複用戶端消息
doWrite(currentTime);
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
try {
this.channel.close();
} catch (IOException e) {
e.printStackTrace();
}
}
private void doWrite(String currentTime){
if(currentTime!=null&¤tTime.trim().length()>0){
byte[] bytes=currentTime.getBytes();
ByteBuffer writeBuffer=ByteBuffer.allocate(bytes.length);
writeBuffer.put(bytes);
writeBuffer.flip();
//向用戶端回複消息
channel.write(writeBuffer, writeBuffer,
new CompletionHandler<Integer, ByteBuffer>() {
@Override
public void completed(Integer result, ByteBuffer attachment) {
// 如果沒有發送完成,繼續發送
if(attachment.hasRemaining()){
channel.write(attachment,attachment,this);
}
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
try {
channel.close();
} catch (IOException e) {
e.printStackTrace();
}
}
});
}
}
}
用戶端:
public class TimeClient {
public static void main(String[] args) {
AsyncTimeClientHandler asyncTimeClientHandler = new AsyncTimeClientHandler("127.0.0.1", 8080);
new Thread(asyncTimeClientHandler, "client-1").start();
}
}
public class AsyncTimeClientHandler implements CompletionHandler<Void, AsyncTimeClientHandler>, Runnable {
private AsynchronousSocketChannel client;
private String host;
private int port;
private CountDownLatch latch;
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() {
latch = new CountDownLatch(1);
//異步連接配接服務端,連接配接成功回調this
client.connect(new InetSocketAddress(host, port), this, this);
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
try {
client.close();
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void completed(Void result, AsyncTimeClientHandler attachment) {
//連接配接成功發送消息
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 attachment) {
if (attachment.hasRemaining()) {
client.write(attachment, attachment, this);
} else {
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
//發送完消息,監聽讀消息
client.read(
readBuffer,
readBuffer,
new CompletionHandler<Integer, ByteBuffer>() {
@Override
public void completed(Integer result, ByteBuffer attachment) {
attachment.flip();
byte[] bytes = new byte[attachment.remaining()];
attachment.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) {
try {
client.close();
latch.countDown();
} catch (IOException e) {
e.printStackTrace();
}
}
}
);
}
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
try {
client.close();
latch.countDown();
} catch (IOException e) {
e.printStackTrace();
}
}
});
}
@Override
public void failed(Throwable exc, AsyncTimeClientHandler attachment) {
try {
client.close();
latch.countDown();
} catch (IOException e) {
e.printStackTrace();
}
}
}