NIO2.0引入了新的异步通道的概念，并提供了异步文件通道和异步套接字通道的实现。异步通道提供以下两种方式获取操作结果。

　　1、通过java.util.concurrent.Future 类来表示异步操作的结果；

　　2、在执行异步操作的时候传入一个java.io.channels。

　　ComplementHandler接口的实现类作为操作完成的回调。

　　NIO2.0的异步套接字通道是真正的异步非阻塞I/O，它不需要通过多路复用器（Selector）对注册的通道进行轮询操作即可实现异步读写，从而简化了NIO编程模型。

　　改造后的代码

server端代码：

　　

 package com.example.biodemo;

 import java.io.*;

 import java.net.ServerSocket;

 import java.net.Socket;

 public class TimeServer {

     public static void main(String[] args) throws IOException {

         int port = 8092;

         if (args != null && args.length > 0) {

             try {

                 port = Integer.valueOf(args[0]);

             } catch (NumberFormatException e) {

                 port = 8092;

             }

         }

         AsyncTimeServerHandler timeServer = new AsyncTimeServerHandler(port);

         new Thread(timeServer,"AsychronousTimeServerHandler").start();

         // ===================改造代码为AIO将以下内容注释掉=================================

 //        创建多路复用线程类并初始化多路复用器，绑定端口等以及轮询注册功能

  /*       MultiplexerTimeServer timeServer = new MultiplexerTimeServer(port);

 //        启动多路复用类线程负责轮询多路复用器Selector，IO数据处理等操作

         new Thread(timeServer, "NIO-MultiplexerTimeSever-001").start();*/

 // ===================以下内容注释掉=================================

        /* ServerSocket server = null;

         try {

             server = new ServerSocket(port);

             System.out.println("the timeServer is start in port :" + port);

             Socket socket = null;

 //           引入线程池start

             TimeServerHandlerExecutePool singleExecutor = new TimeServerHandlerExecutePool(50,10000);

             while (true) {

                 socket = server.accept();

 //           替换BIO中new Thread(new TimeServerHandler(socket)).start();为下一行代码

                 singleExecutor.execute(new TimeServerHandler(socket));

 //           引入线程池end

             }

         } finally {

             if (server != null) {

                 System.out.println("the time server close");

                 server.close();

                 server = null;

             }

         }*/

     }

 }

异步时间服务处理器

 package com.example.biodemo;

 import java.io.IOException;

 import java.net.InetSocketAddress;

 import java.nio.channels.AsynchronousServerSocketChannel;

 import java.util.concurrent.CountDownLatch;

 public class AsyncTimeServerHandler implements Runnable{

      private int port;

 //     添加它作用是在完成一组正在执行的操作之前，允许当前线程一直阻塞

      CountDownLatch latch;

      AsynchronousServerSocketChannel asynchronousServerSocketChannel;

     public 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();

         }

     }

     private void doAccept() {

 //        AcceptCompletionHandler用于接收accept操作成功的通知消息

 //        accept(A attachment, CompletionHandler<AsynchronousSocketChannel,? super A> handler)：接受连接，并为连接绑定一个CompletionHandler处理Socket连接

         asynchronousServerSocketChannel.accept(this,new AcceptCompletionHandler());

     }

 }

AcceptCompletionHandle

 package com.example.biodemo;

 import java.nio.ByteBuffer;

 import java.nio.channels.AsynchronousServerSocketChannel;

 import java.nio.channels.AsynchronousSocketChannel;

 import java.nio.channels.CompletionHandler;

 public class AcceptCompletionHandler implements CompletionHandler<AsynchronousSocketChannel,AsyncTimeServerHandler>{

     @Override

     public void completed(AsynchronousSocketChannel result, AsyncTimeServerHandler attachment) {

         //再次让asynchronousServerSocketChannel对象调用accept方法是

         //调用AsynchronousServerSocketChannel的accept方法后，如果有新的客户端接入，

         // 系统将回调我们传入的CompletionHandler实例的completed方法，表示新客户端连接成功。

         // 因为AsynchronousServerSocketChannel可以接受成千上万个客户端，所以需要继续调用它的accept方法，

         // 接受其他客户端连接，最终形成一个环；每当一个客户端连接成功后，再异步接受新的客户端连接

         attachment.asynchronousServerSocketChannel.accept(attachment,this);

 //        据上，链路建立成功，服务端需要接受客户端新请求消息，

         ByteBuffer buffer = ByteBuffer.allocate(1024);

 //        调用read方法进行异步读操作

         result.read(buffer,buffer,new ReadCompletionHandler(result));

     }

     @Override

     public void failed(Throwable exc, AsyncTimeServerHandler attachment) {

         exc.printStackTrace();

         attachment.latch.countDown();

     }

 }

ReadCompletionHandler

 package com.example.biodemo;

 import java.io.IOException;

 import java.io.UnsupportedEncodingException;

 import java.nio.ByteBuffer;

 import java.nio.channels.AsynchronousSocketChannel;

 import java.nio.channels.CompletionHandler;

 import java.util.Date;

 public class ReadCompletionHandler implements CompletionHandler<Integer,ByteBuffer> {

     private AsynchronousSocketChannel channel;

     public ReadCompletionHandler(AsynchronousSocketChannel channel) {

         if(this.channel==null){

             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 recerve order : "+req);

             String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(req)?new Date(System.currentTimeMillis()).toString():"BAD ORDER";

             doWriter(currentTime);

         } catch (UnsupportedEncodingException e) {

             e.printStackTrace();

         }

     }

     private void doWriter(String currentTime) {

         if (currentTime != null && currentTime.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) {

                     }

                 }

             });

             }

     }

     @Override

     public void failed(Throwable exc, ByteBuffer attachment) {

         try {

             this.channel.close();

         } catch (IOException e) {

             e.printStackTrace();

         }

     }

 }

客户端的代码

 package com.example.biodemo;

 import java.io.*;

 import java.net.Socket;

 public class TimeClient {

     public static void main(String[] args) {

         int port = 8092;

         if (args != null && args.length > 0) {

             try {

                 port = Integer.valueOf(args[0]);

             } catch (NumberFormatException ne) {

                 port = 8092;

             }

         }

         new Thread(new AsyncTimeClientHandler("127.0.0.1",port)).start();

 //        new Thread(new TimeClientHandles("127.0.0.1",port),"TimeClient-001").start();

       /* 代码改造注释掉以下代码

        Socket socket = null;

         BufferedReader in = null;

         PrintWriter out = null;

         try {

             socket = new Socket("127.0.0.1", port);

             System.out.println(socket.getInputStream());

             in = new BufferedReader(new InputStreamReader(socket.getInputStream()));

             out = new PrintWriter(socket.getOutputStream(), true);

             out.println("QUERY TIME ORDER");

             System.out.println("send order 2 server succeed.");

             String resp = in.readLine();

             System.out.println("now is :" + resp);

         } catch (IOException e1) {

         } finally {

             if (out != null) {

                 out.close();

                 out = null;

             }

             if (in != null) {

                 try {

                     in.close();

                 } catch (IOException e2) {

                     e2.printStackTrace();

                 }

                 in = null;

                 if (socket != null) {

                     try {

                         socket.close();

                     } catch (IOException e3) {

                         e3.printStackTrace();

                     }

                 }

                 socket = null;

             }

         }*/

     }

 }

AsyncTimeClientHandler

 package com.example.biodemo;

 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 asynchronousSocketChannel;

     private String host;

     private int port;

     private CountDownLatch latch;

     public AsyncTimeClientHandler(String host, int port) {

         this.host = host == null ? "127.0.0.1" : host;

         this.port = port;

         try {

             asynchronousSocketChannel = AsynchronousSocketChannel.open();

         } catch (IOException e) {

             e.printStackTrace();

         }

     }

     @Override

     public void run() {

         latch = new CountDownLatch(1);

         asynchronousSocketChannel.connect(new InetSocketAddress(host, port), this, this);

         try {

             latch.await();

         } catch (InterruptedException e) {

             e.printStackTrace();

         }

         try {

             asynchronousSocketChannel.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();

         asynchronousSocketChannel.write(writeBuffer, writeBuffer, new CompletionHandler<Integer, ByteBuffer>() {

             @Override

             public void completed(Integer result, ByteBuffer attachment) {

                 if (attachment.hasRemaining()) {

                     asynchronousSocketChannel.write(attachment, attachment, this);

                 } else {

                     ByteBuffer readBuffer = ByteBuffer.allocate(1024);

                     asynchronousSocketChannel.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 = null;

                             try {

                                 body = new String(bytes, "utf-8");

                                 System.out.println(body);

                                 System.out.println("Now is :" + body);

                                 latch.countDown();

                             } catch (UnsupportedEncodingException e) {

                                 e.printStackTrace();

                             }

                         }

                         @Override

                         public void failed(Throwable exc, ByteBuffer attachment) {

                             try {

                                 asynchronousSocketChannel.close();

                                 latch.countDown();

                             } catch (IOException e) {

                                 e.printStackTrace();

                             }

                         }

                     });

                 }

             }

             @Override

             public void failed(Throwable exc, ByteBuffer attachment) {

                 try {

                     asynchronousSocketChannel.close();

                     latch.countDown();

                 } catch (IOException e) {

                     e.printStackTrace();

                 }

             }

         });

     }

     @Override

     public void failed(Throwable exc, AsyncTimeClientHandler attachment) {

         exc.printStackTrace();

         try {

             asynchronousSocketChannel.close();

             latch.countDown();

         } catch (IOException e) {

             e.printStackTrace();

         }

     }

 }

　　学到这里我们对NIO编程已经有了一个感性认识。

最新文章

  1. 数据结构：堆排序 （python版） 小顶堆实现从大到小排序 | 大顶堆实现从小到大排序
  2. IE6不支持min-heightt和min-width的解决办法
  3. 手把手教你如何把java代码，打包成jar文件以及转换为exe可执行文件
  4. MySQL\MariaDB 多线程复制初探
  5. 【C#进阶系列】10 属性
  6. 基于AppCan MAS系统，如何轻松实现移动应用数据服务？
  7. CPU 材料学才是最顶级的学科
  8. WIN7 64位配置Oracle SQL Developer工具
  9. c# 面相对象1-概括
  10. Mybatis学习（8）逆向工程
  11. css基本语法及页面引用
  12. Vue .Net 前后端分离框架搭建
  13. [转]ssh和SSH服务(包含隧道内容)
  14. SQL简单操作
  15. lucene简介——（一）
  16. 关于Linux DNS部分处理
  17. 代码整洁之道读书笔记（Ch4-Ch7）
  18. openwrt从0开始-目录
  19. ACE主动对象模式(1)
  20. ZOJ 3872 Beauty of Array【无重复连续子序列的贡献和/规律/DP】

热门文章

  1. 「NOIP2017」逛公园
  2. C++11并发编程3------线程传参
  3. Write-up-CH4INRULZ_v1.0.1
  4. 为常用的块类型创建typedef
  5. python生成器三元表达式
  6. Press Key关键字用法
  7. Centos7 VNC远程桌面服务安装配置
  8. [转]网络协议-redis协议
  9. LeetCode 1025. Divisor Game
  10. 【mysql】mysq8.0新特性