2021年7月7日星期三

深入学习Netty（3）——传统AIO编程

前言

　　之前已经整理过了BIO、NIO两种I/O的相关博文，每一种I/O都有其特点，但相对开发而言，肯定是要又高效又简单的I/O编程才是真正需要的，在之前的NIO博文（深入学习Netty（2）——传统NIO编程）中就已经介绍过NIO编程的缺点（相比较而言的缺点：同步非阻塞，需要单独开启线程不断轮询），所以才会有真正的异步非阻塞I/O出现，这就是此篇博文需要介绍的AIO编程。

　　参考资料《Netty In Action》、《Netty权威指南》（有需要的小伙伴可以评论或者私信我）

　　博文中所有的代码都已上传到Github，欢迎Star、Fork

　　感兴趣可以先学习相关博文：

  • 深入学习Netty（1）——传统BIO编程
  • 深入学习Netty（2）——传统NIO编程

 

 

一、NIO 2.0与AIO编程

　　JDK 1.7升级了NIO类库，升级后的NIO类库称之为NIO 2.0，Java提供了异步文件I/O操作，同时提供了与UNIX网络编程事件驱动I/O对应的AIO。

　　NIO 2.0的异步套接字通道是真正的异步非阻塞I/O，对应有UNIX网络编程中的事件驱动I/O（AIO），相比较NIO，它不需要通过Selector对注册的通道进行轮询操作即可实现异步读写，简化了NIO的编程模型。

　　NIO 2.0提供了新的异步通道的概念，异步通道提供了以下两种方式获取操作结果：

  • 通过juc.Futrue类来表示异步操作的结果。
AsynchronousSocketChannel socketChannel = AsynchronousSocketChannel.open();InetSocketAddress inetSocketAddress = new InetSocketAddress("localhost", 8080);Future<Void> connect = socketChannel.connect(inetSocketAddress);while (!connect.isDone()) {
　　Thread.sleep(10);}

  • 在异步操作的时候传入java.nio.channels。实现CompletionHandler接口complete()的方法作为操作完成回调。
private class MyCompletionHandler implements CompletionHandler<Integer, ByteBuffer> { 　　@Override　　public void completed(Integer result, ByteBuffer attachment) {　　　　// TODO 回调后业务操作　　}　　@Override　　public void failed(Throwable t, ByteBuffer attachment) {　　　　t.printStackTrace();　　}

二、AIO服务端

（1）服务端AIO异步处理任务AsyncTimeServerHandler：

  • 创建异步服务通道并监听端口
  • 异步监听客户端连接
/** * 服务端AIO异步处理任务 * -创建异步服务通道监听端口 * -监听客户端连接 */public class AsyncTimeServerHandler implements Runnable{ private int port; CountDownLatch latch; AsynchronousServerSocketChannel asynchronousServerSocketChannel; public AsyncTimeServerHandler(int port) {  this.port = port;  try {   // 创建异步的服务通道asynchronousServerSocketChannel, 并bind监听端口   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() {  // countDownLatch没有count减一，所以导致一直阻塞  latch = new CountDownLatch(1);  doAccept();  try {   // 防止执行操作线程还未结束，服务端线程就退出，程序不退出的前提下，才能够让accept继续可以回调接受来自客户端的连接   // 实际开发过程中不需要单独开启线程去处理AsynchronousServerSocketChannel   latch.await();  } catch (InterruptedException e) {   e.printStackTrace();  } } /**  * 接收客户端的连接  * 参数CompletionHandler类型的handler实例来接收accept操作成功的通知消息  */ public void doAccept() {  asynchronousServerSocketChannel.accept(this, new AcceptCompletionHandler()); }}

（2）服务端连接异步回调处理器AcceptCompletionHandler：异步处理客户端连接完成后的操作

/** * 客户端连接异步处理器 * completed()方法完成回调logic * failed()方法完成失败回调logic */public class AcceptCompletionHandler implements CompletionHandler<AsynchronousSocketChannel, AsyncTimeServerHandler> { /**  * 调用该方法表示客户端已经介接入成功  * 同时再accept接收新的客户端连接  * @param result  * @param attachment  */ @Override public void completed(AsynchronousSocketChannel result,       AsyncTimeServerHandler attachment) {  // 此时还要继续调用accept方法是因为，completed方法表示上一个客户端连接完成，而下一个新的客户端需要连接  // 如此形成新的循环：每接收一个客户端的成功连接之后，再异步接收新的客户端连接  attachment.asynchronousServerSocketChannel.accept(attachment, this);  // 预分配1M的缓冲区  ByteBuffer buffer = ByteBuffer.allocate(1024);  // 调用read方法异步读，传入CompletionHandler类型参数异步回调读事件  result.read(buffer, buffer, new ReadCompletionHandler(result)); } @Override public void failed(Throwable exc, AsyncTimeServerHandler attachment) {  exc.printStackTrace();  // 让服务线程不再阻塞  attachment.latch.countDown(); }}

（3）服务端read事件异步回调处理器ReadCompletionHandler：异步回调处理客户端请求数据

/** * 服务端read事件异步处理器 * completed异步回调处理客户端请求数据 */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数组  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 java.util.Date(     System.currentTimeMillis()).toString() : "BAD ORDER";   // 发送当前时间给客户端   doWrite(currentTime);  } catch (UnsupportedEncodingException e) {   e.printStackTrace();  } } private void doWrite(String currentTime) {  if (currentTime != null && currentTime.trim().length() > 0) {   byte[] bytes = (currentTime).getBytes();   ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);   writeBuffer.put(bytes);   writeBuffer.flip();   // write异步回调，传入CompletionHandler类型参数   channel.write(writeBuffer, writeBuffer,     new CompletionHandler<Integer, ByteBuffer>() {      @Override      public void completed(Integer result, ByteBuffer buffer) {       // 如果没有发送完成，继续发送       if (buffer.hasRemaining()) {        channel.write(buffer, buffer, this);       }      }      @Override      public void failed(Throwable exc, ByteBuffer attachment) {       try {        channel.close();       } catch (IOException e) {        // TODO 只要是I/O异常就需要关闭链路，释放资源       }      }     });  } } @Override public void failed(Throwable exc, ByteBuffer attachment) {  try {   this.channel.close();  } catch (IOException e) {   e.printStackTrace();   // TODO 只要是I/O异常就需要关闭链路，释放资源  } }}

（4）服务端启动TimeServer

/** * AIO 异步非阻塞服务端 * 不需要单独开线程去处理read、write等事件 * 只需要关注complete-handlers中的回调completed方法 */public class TimeServer { public static void main(String[] args) throws IOException {  int port = 8086;  AsyncTimeServerHandler timeServer = new AsyncTimeServerHandler(port);  new Thread(timeServer, "AIO-AsyncTimeServerHandler").start(); }}

（5）启动服务端

服务端Console：

使用命令netstat查看8086端口是否监听

三、AIO客户端

（1）客户端AIO异步回调处理任务：

  • 打开AsynchronousSocketChannel通道，连接服务端
  • 发送服务端指令
  • 回调处理服务端应答
/** * 客户端AIO异步回调处理任务 * -打开AsynchronousSocketChannel通道，连接服务端 * -发送服务端指令 * -回调处理服务端应答 */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);  client.connect(new InetSocketAddress(host, port), this, this);  try {   // 防止异步操作都没完成，连接线程就结束退出   latch.await();  } catch (InterruptedException e1) {   e1.printStackTrace();  }  try {   client.close();  } catch (IOException e) {   e.printStackTrace();  } } /**  * 发送请求完成异步回调  * @param result  * @param attachment  */ @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 buffer) {      if (buffer.hasRemaining()) {       client.write(buffer, buffer, this);      } else {       ByteBuffer readBuffer = ByteBuffer.allocate(1024);       // 回调服务端应答消息       client.read(readBuffer, readBuffer,         new CompletionHandler<Integer, ByteBuffer>() {          @Override          public void completed(Integer result, ByteBuffer buffer) {           buf......
原文转载：http://www.shaoqun.com/a/852298.html
跨境电商：https://www.ikjzd.com/
Zozo：https://www.ikjzd.com/w/2180
 
custommade：https://www.ikjzd.com/w/2514
 
二类电商：https://www.ikjzd.com/w/1457
 
前言　　之前已经整理过了BIO、NIO两种I/O的相关博文，每一种I/O都有其特点，但相对开发而言，肯定是要又高效又简单的I/O编程才是真正需要的，在之前的NIO博文（深入学习Netty（2）——传统NIO编程）中就已经介绍过NIO编程的缺点（相比较而言的缺点：同步非阻塞，需要单独开启线程不断轮询），所以才会有真正的异步非阻塞I/O出现，这就是此篇博文需要介绍的AIO编程。　　参考资料《Netty
lithium：https://www.ikjzd.com/w/2505
干货！亚马逊小白必备基础扫盲知识：https://www.ikjzd.com/articles/104238
亚马逊站内广告如何创建？（附图文实操详解）：https://www.ikjzd.com/articles/104239
国务院：2019年底前出台对跨境电商等企业的认证标准！：https://www.ikjzd.com/articles/104240
想提升亚马逊listing转化率，这些数据你每天都要看！：https://www.ikjzd.com/articles/104241
校花喂我奶我脱她胸罩 我吃了校花一节课的奶：http://lady.shaoqun.com/a/247349.html
口述：老婆被闺蜜洗脑后求我参与换妻：http://www.30bags.com/m/a/249608.html
口述：我和小姨子那些旧事儿（上）(11/12)：http://www.30bags.com/m/a/249644.html
不再由贝索斯掌舵 亚马逊将驶向何方？：https://www.ikjzd.com/articles/146429
独立日0单？亚马逊订单大跌，大卖7000万库存待清理：https://www.ikjzd.com/articles/146428
词缀en的常见意义和用法：http://lady.shaoqun.com/a/405834.html
2021深圳锦绣中华民俗村泼水节夜场攻略：http://www.30bags.com/a/476661.html
发帖者 时间：

没有评论:

发表评论

订阅： 博文评论 (Atom)