AIO编程（NIO2.0）及代码实现

淡淡的烟草味﹌ 2022-06-08 09:48 178阅读 0赞

## AIO编程 ##

NIO 2.0引入了新的异步通道的概念，并提供了异步文件通道和异步套接字通道的实现。

异步的套接字通道时真正的异步非阻塞I/O，对应于UNIX网络编程中的事件驱动I/O（AIO）。他不需要过多的Selector对注册的通道进行轮询即可实现异步读写，从而简化了NIO的编程模型。

直接上代码吧。

###     3.1、Server端代码 ###

Server：

**\[java\]** [view plain][] [copy][view plain]

[print][view plain][?][view plain]

1.  **package** com.anxpp.io.calculator.aio.server;  
2.  /\*\* 
3.   \* AIO服务端 
4.   \* @author yangtao\_\_anxpp.com 
5.   \* @version 1.0 
6.   \*/  
7.  **public** **class** Server \{  
8.      **private** **static** **int** DEFAULT\_PORT = 12345;  
9.      **private** **static** AsyncServerHandler serverHandle;  
10.     **public** **volatile** **static** **long** clientCount = 0;  
11.     **public** **static** **void** start()\{  
12.         start(DEFAULT\_PORT);  
13.     \}  
14.     **public** **static** **synchronized** **void** start(**int** port)\{  
15.         **if**(serverHandle!=**null**)  
16.             **return**;  
17.         serverHandle = **new** AsyncServerHandler(port);  
18.         **new** Thread(serverHandle,"Server").start();  
19.     \}  
20.     **public** **static** **void** main(String\[\] args)\{  
21.         Server.start();  
22.     \}  
23. \}

AsyncServerHandler：

**\[java\]** [view plain][] [copy][view plain]

[print][view plain][?][view plain]

1.  **package** com.anxpp.io.calculator.aio.server;  
2.  **import** java.io.IOException;  
3.  **import** java.net.InetSocketAddress;  
4.  **import** java.nio.channels.AsynchronousServerSocketChannel;  
5.  **import** java.util.concurrent.CountDownLatch;  
6.  **public** **class** AsyncServerHandler **implements** Runnable \{  
7.      **public** CountDownLatch latch;  
8.      **public** AsynchronousServerSocketChannel channel;  
9.      **public** AsyncServerHandler(**int** port) \{  
10.         **try** \{  
11.             //创建服务端通道  
12.             channel = AsynchronousServerSocketChannel.open();  
13.             //绑定端口  
14.             channel.bind(**new** InetSocketAddress(port));  
15.             System.out.println("服务器已启动，端口号：" + port);  
16.         \} **catch** (IOException e) \{  
17.             e.printStackTrace();  
18.         \}  
19.     \}  
20.     @Override  
21.     **public** **void** run() \{  
22.         //CountDownLatch初始化  
23.         //它的作用：在完成一组正在执行的操作之前，允许当前的现场一直阻塞  
24.         //此处，让现场在此阻塞，防止服务端执行完成后退出  
25.         //也可以使用while(true)+sleep   
26.         //生成环境就不需要担心这个问题，以为服务端是不会退出的  
27.         latch = **new** CountDownLatch(1);  
28.         //用于接收客户端的连接  
29.         channel.accept(**this**,**new** AcceptHandler());  
30.         **try** \{  
31.             latch.await();  
32.         \} **catch** (InterruptedException e) \{  
33.             e.printStackTrace();  
34.         \}  
35.     \}  
36. \}

AcceptHandler：

**\[java\]** [view plain][] [copy][view plain]

[print][view plain][?][view plain]

1.  **package** com.anxpp.io.calculator.aio.server;  
2.  **import** java.nio.ByteBuffer;  
3.  **import** java.nio.channels.AsynchronousSocketChannel;  
4.  **import** java.nio.channels.CompletionHandler;  
5.  //作为handler接收客户端连接  
6.  **public** **class** AcceptHandler **implements** CompletionHandler<AsynchronousSocketChannel, AsyncServerHandler> \{  
7.      @Override  
8.      **public** **void** completed(AsynchronousSocketChannel channel,AsyncServerHandler serverHandler) \{  
9.          //继续接受其他客户端的请求  
10.         Server.clientCount++;  
11.         System.out.println("连接的客户端数：" + Server.clientCount);  
12.         serverHandler.channel.accept(serverHandler, **this**);  
13.         //创建新的Buffer  
14.         ByteBuffer buffer = ByteBuffer.allocate(1024);  
15.         //异步读  第三个参数为接收消息回调的业务Handler  
16.         channel.read(buffer, buffer, **new** ReadHandler(channel));  
17.     \}  
18.     @Override  
19.     **public** **void** failed(Throwable exc, AsyncServerHandler serverHandler) \{  
20.         exc.printStackTrace();  
21.         serverHandler.latch.countDown();  
22.     \}  
23. \}

ReadHandler：

**\[java\]** [view plain][] [copy][view plain]

[print][view plain][?][view plain]

1.  **package** com.anxpp.io.calculator.aio.server;  
2.  **import** java.io.IOException;  
3.  **import** java.io.UnsupportedEncodingException;  
4.  **import** java.nio.ByteBuffer;  
5.  **import** java.nio.channels.AsynchronousSocketChannel;  
6.  **import** java.nio.channels.CompletionHandler;  
7.  **import** com.anxpp.io.utils.Calculator;  
8.  **public** **class** ReadHandler **implements** CompletionHandler<Integer, ByteBuffer> \{  
9.      //用于读取半包消息和发送应答  
10.     **private** AsynchronousSocketChannel channel;  
11.     **public** ReadHandler(AsynchronousSocketChannel channel) \{  
12.             **this**.channel = channel;  
13.     \}  
14.     //读取到消息后的处理  
15.     @Override  
16.     **public** **void** completed(Integer result, ByteBuffer attachment) \{  
17.         //flip操作  
18.         attachment.flip();  
19.         //根据  
20.         **byte**\[\] message = **new** **byte**\[attachment.remaining()\];  
21.         attachment.get(message);  
22.         **try** \{  
23.             String expression = **new** String(message, "UTF-8");  
24.             System.out.println("服务器收到消息: " + expression);  
25.             String calrResult = **null**;  
26.             **try**\{  
27.                 calrResult = Calculator.cal(expression).toString();  
28.             \}**catch**(Exception e)\{  
29.                 calrResult = "计算错误：" + e.getMessage();  
30.             \}  
31.             //向客户端发送消息  
32.             doWrite(calrResult);  
33.         \} **catch** (UnsupportedEncodingException e) \{  
34.             e.printStackTrace();  
35.         \}  
36.     \}  
37.     //发送消息  
38.     **private** **void** doWrite(String result) \{  
39.         **byte**\[\] bytes = result.getBytes();  
40.         ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);  
41.         writeBuffer.put(bytes);  
42.         writeBuffer.flip();  
43.         //异步写数据 参数与前面的read一样  
44.         channel.write(writeBuffer, writeBuffer,**new** CompletionHandler<Integer, ByteBuffer>() \{  
45.             @Override  
46.             **public** **void** completed(Integer result, ByteBuffer buffer) \{  
47.                 //如果没有发送完，就继续发送直到完成  
48.                 **if** (buffer.hasRemaining())  
49.                     channel.write(buffer, buffer, **this**);  
50.                 **else**\{  
51.                     //创建新的Buffer  
52.                     ByteBuffer readBuffer = ByteBuffer.allocate(1024);  
53.                     //异步读  第三个参数为接收消息回调的业务Handler  
54.                     channel.read(readBuffer, readBuffer, **new** ReadHandler(channel));  
55.                 \}  
56.             \}  
57.             @Override  
58.             **public** **void** failed(Throwable exc, ByteBuffer attachment) \{  
59.                 **try** \{  
60.                     channel.close();  
61.                 \} **catch** (IOException e) \{  
62.                 \}  
63.             \}  
64.         \});  
65.     \}  
66.     @Override  
67.     **public** **void** failed(Throwable exc, ByteBuffer attachment) \{  
68.         **try** \{  
69.             **this**.channel.close();  
70.         \} **catch** (IOException e) \{  
71.             e.printStackTrace();  
72.         \}  
73.     \}  
74. \}

OK，这样就已经完成了，其实说起来也简单，虽然代码感觉很多，但是API比NIO的使用起来真的简单多了，主要就是监听、读、写等各种CompletionHandler。此处本应有一个WriteHandler的，确实，我们在ReadHandler中，以一个匿名内部类实现了它。

下面看客户端代码。

###     3.2、Client端代码 ###

Client：

**\[java\]** [view plain][] [copy][view plain]

[print][view plain][?][view plain]

1.  **package** com.anxpp.io.calculator.aio.client;  
2.  **import** java.util.Scanner;  
3.  **public** **class** Client \{  
4.      **private** **static** String DEFAULT\_HOST = "127.0.0.1";  
5.      **private** **static** **int** DEFAULT\_PORT = 12345;  
6.      **private** **static** AsyncClientHandler clientHandle;  
7.      **public** **static** **void** start()\{  
8.          start(DEFAULT\_HOST,DEFAULT\_PORT);  
9.      \}  
10.     **public** **static** **synchronized** **void** start(String ip,**int** port)\{  
11.         **if**(clientHandle!=**null**)  
12.             **return**;  
13.         clientHandle = **new** AsyncClientHandler(ip,port);  
14.         **new** Thread(clientHandle,"Client").start();  
15.     \}  
16.     //向服务器发送消息  
17.     **public** **static** **boolean** sendMsg(String msg) **throws** Exception\{  
18.         **if**(msg.equals("q")) **return** **false**;  
19.         clientHandle.sendMsg(msg);  
20.         **return** **true**;  
21.     \}  
22.     @SuppressWarnings("resource")  
23.     **public** **static** **void** main(String\[\] args) **throws** Exception\{  
24.         Client.start();  
25.         System.out.println("请输入请求消息：");  
26.         Scanner scanner = **new** Scanner(System.in);  
27.         **while**(Client.sendMsg(scanner.nextLine()));  
28.     \}  
29. \}

AsyncClientHandler：

**\[java\]** [view plain][] [copy][view plain]

[print][view plain][?][view plain]

1.  **package** com.anxpp.io.calculator.aio.client;  
2.  **import** java.io.IOException;  
3.  **import** java.net.InetSocketAddress;  
4.  **import** java.nio.ByteBuffer;  
5.  **import** java.nio.channels.AsynchronousSocketChannel;  
6.  **import** java.nio.channels.CompletionHandler;  
7.  **import** java.util.concurrent.CountDownLatch;  
8.  **public** **class** AsyncClientHandler **implements** CompletionHandler<Void, AsyncClientHandler>, Runnable \{  
9.      **private** AsynchronousSocketChannel clientChannel;  
10.     **private** String host;  
11.     **private** **int** port;  
12.     **private** CountDownLatch latch;  
13.     **public** AsyncClientHandler(String host, **int** port) \{  
14.         **this**.host = host;  
15.         **this**.port = port;  
16.         **try** \{  
17.             //创建异步的客户端通道  
18.             clientChannel = AsynchronousSocketChannel.open();  
19.         \} **catch** (IOException e) \{  
20.             e.printStackTrace();  
21.         \}  
22.     \}  
23.     @Override  
24.     **public** **void** run() \{  
25.         //创建CountDownLatch等待  
26.         latch = **new** CountDownLatch(1);  
27.         //发起异步连接操作，回调参数就是这个类本身，如果连接成功会回调completed方法  
28.         clientChannel.connect(**new** InetSocketAddress(host, port), **this**, **this**);  
29.         **try** \{  
30.             latch.await();  
31.         \} **catch** (InterruptedException e1) \{  
32.             e1.printStackTrace();  
33.         \}  
34.         **try** \{  
35.             clientChannel.close();  
36.         \} **catch** (IOException e) \{  
37.             e.printStackTrace();  
38.         \}  
39.     \}  
40.     //连接服务器成功  
41.     //意味着TCP三次握手完成  
42.     @Override  
43.     **public** **void** completed(Void result, AsyncClientHandler attachment) \{  
44.         System.out.println("客户端成功连接到服务器...");  
45.     \}  
46.     //连接服务器失败  
47.     @Override  
48.     **public** **void** failed(Throwable exc, AsyncClientHandler attachment) \{  
49.         System.err.println("连接服务器失败...");  
50.         exc.printStackTrace();  
51.         **try** \{  
52.             clientChannel.close();  
53.             latch.countDown();  
54.         \} **catch** (IOException e) \{  
55.             e.printStackTrace();  
56.         \}  
57.     \}  
58.     //向服务器发送消息  
59.     **public** **void** sendMsg(String msg)\{  
60.         **byte**\[\] req = msg.getBytes();  
61.         ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);  
62.         writeBuffer.put(req);  
63.         writeBuffer.flip();  
64.         //异步写  
65.         clientChannel.write(writeBuffer, writeBuffer,**new** WriteHandler(clientChannel, latch));  
66.     \}  
67. \}

WriteHandler：

**\[java\]** [view plain][] [copy][view plain]

[print][view plain][?][view plain]

1.  **package** com.anxpp.io.calculator.aio.client;  
2.  **import** java.io.IOException;  
3.  **import** java.nio.ByteBuffer;  
4.  **import** java.nio.channels.AsynchronousSocketChannel;  
5.  **import** java.nio.channels.CompletionHandler;  
6.  **import** java.util.concurrent.CountDownLatch;  
7.  **public** **class** WriteHandler **implements** CompletionHandler<Integer, ByteBuffer> \{  
8.      **private** AsynchronousSocketChannel clientChannel;  
9.      **private** CountDownLatch latch;  
10.     **public** WriteHandler(AsynchronousSocketChannel clientChannel,CountDownLatch latch) \{  
11.         **this**.clientChannel = clientChannel;  
12.         **this**.latch = latch;  
13.     \}  
14.     @Override  
15.     **public** **void** completed(Integer result, ByteBuffer buffer) \{  
16.         //完成全部数据的写入  
17.         **if** (buffer.hasRemaining()) \{  
18.             clientChannel.write(buffer, buffer, **this**);  
19.         \}  
20.         **else** \{  
21.             //读取数据  
22.             ByteBuffer readBuffer = ByteBuffer.allocate(1024);  
23.             clientChannel.read(readBuffer,readBuffer,**new** ReadHandler(clientChannel, latch));  
24.         \}  
25.     \}  
26.     @Override  
27.     **public** **void** failed(Throwable exc, ByteBuffer attachment) \{  
28.         System.err.println("数据发送失败...");  
29.         **try** \{  
30.             clientChannel.close();  
31.             latch.countDown();  
32.         \} **catch** (IOException e) \{  
33.         \}  
34.     \}  
35. \}

ReadHandler：

**\[java\]** [view plain][] [copy][view plain]

[print][view plain][?][view plain]

1.  **package** com.anxpp.io.calculator.aio.client;  
2.  **import** java.io.IOException;  
3.  **import** java.io.UnsupportedEncodingException;  
4.  **import** java.nio.ByteBuffer;  
5.  **import** java.nio.channels.AsynchronousSocketChannel;  
6.  **import** java.nio.channels.CompletionHandler;  
7.  **import** java.util.concurrent.CountDownLatch;  
8.  **public** **class** ReadHandler **implements** CompletionHandler<Integer, ByteBuffer> \{  
9.      **private** AsynchronousSocketChannel clientChannel;  
10.     **private** CountDownLatch latch;  
11.     **public** ReadHandler(AsynchronousSocketChannel clientChannel,CountDownLatch latch) \{  
12.         **this**.clientChannel = clientChannel;  
13.         **this**.latch = latch;  
14.     \}  
15.     @Override  
16.     **public** **void** completed(Integer result,ByteBuffer buffer) \{  
17.         buffer.flip();  
18.         **byte**\[\] bytes = **new** **byte**\[buffer.remaining()\];  
19.         buffer.get(bytes);  
20.         String body;  
21.         **try** \{  
22.             body = **new** String(bytes,"UTF-8");  
23.             System.out.println("客户端收到结果:"\+ body);  
24.         \} **catch** (UnsupportedEncodingException e) \{  
25.             e.printStackTrace();  
26.         \}  
27.     \}  
28.     @Override  
29.     **public** **void** failed(Throwable exc,ByteBuffer attachment) \{  
30.         System.err.println("数据读取失败...");  
31.         **try** \{  
32.             clientChannel.close();  
33.             latch.countDown();  
34.         \} **catch** (IOException e) \{  
35.         \}  
36.     \}  
37. \}

这个API使用起来真的是很顺手。

###     3.3、测试 ###

Test：

**\[java\]** [view plain][] [copy][view plain]

[print][view plain][?][view plain]

1.  **package** com.anxpp.io.calculator.aio;  
2.  **import** java.util.Scanner;  
3.  **import** com.anxpp.io.calculator.aio.client.Client;  
4.  **import** com.anxpp.io.calculator.aio.server.Server;  
5.  /\*\* 
6.   \* 测试方法 
7.   \* @author yangtao\_\_anxpp.com 
8.   \* @version 1.0 
9.   \*/  
10. **public** **class** Test \{  
11.     //测试主方法  
12.     @SuppressWarnings("resource")  
13.     **public** **static** **void** main(String\[\] args) **throws** Exception\{  
14.         //运行服务器  
15.         Server.start();  
16.         //避免客户端先于服务器启动前执行代码  
17.         Thread.sleep(100);  
18.         //运行客户端   
19.         Client.start();  
20.         System.out.println("请输入请求消息：");  
21.         Scanner scanner = **new** Scanner(System.in);  
22.         **while**(Client.sendMsg(scanner.nextLine()));  
23.     \}  
24. \}

我们可以在控制台输入我们需要计算的算数字符串，服务器就会返回结果，当然，我们也可以运行大量的客户端，都是没有问题的，以为此处设计为单例客户端，所以也就没有演示大量客户端并发。

读者可以自己修改Client类，然后开辟大量线程，并使用构造方法创建很多的客户端测试。

下面是其中一次参数的输出：

1.  服务器已启动，端口号：12345
2.  请输入请求消息：
3.  客户端成功连接到服务器...
4.  连接的客户端数：1
5.  123456\+789\+456
6.  服务器收到消息:123456\+789\+456
7.  客户端收到结果:124701
8.  9526\*56
9.  服务器收到消息:9526\*56
10. 客户端收到结果:533456
11. ...

AIO是真正的异步非阻塞的，所以，在面对超级大量的客户端，更能得心应手。

下面就比较一下，几种I/O编程的优缺点。

## 4、各种I/O的对比 ##

先以一张表来直观的对比一下：

![03][]

具体选择什么样的模型或者NIO框架，完全基于业务的实际应用场景和性能需求，如果客户端很少，服务器负荷不重，就没有必要选择开发起来相对不那么简单的NIO做服务端；相反，就应考虑使用NIO或者相关的框架了。

## 5、附录 ##

上文中服务端使用到的用于计算的工具类：

1.  package com.anxpp.utils;

1.  import javax.script.ScriptEngine;

1.  import javax.script.ScriptEngineManager;

1.  import javax.script.ScriptException;