Socket编程（三）（AIO）

骑猪看日落 2022-05-11 14:33 233阅读 0赞

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

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

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

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

代码如下：

服务端：

public class TimeServer {
    	public static void main(String[] args) throws IOException {
    		int port = 8080;
    		AsyncTimeServerHandler timeServer = new AsyncTimeServerHandler(port);
    		new Thread(timeServer, "AIO-AsyncTimeServerHandler-001").start();
    	}
    }

首先创建异步的时间服务器处理类，然后启动线程将AsyncTimeServerHandler启动：

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));
    			// bind 一个监听端口
    			System.out.println("时间服务器在端口启动 : " + 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() {
    		asynchronousServerSocketChannel.accept(this, new AcceptCompletionHandler());// 处理接受消息的通知。 
    	}
    
    }

首先在构造方法中，先创建一个异步的服务端通道 AsynchronousServerSocketChannel，然后调用它的bind方法绑定监听端口。如果端口合法且未被占用则绑定成功打印。

![70][]

在run方法中首先通过new CountDownLatch(1)初始化CountDownLatch对象，它的作用是在完成一组正在执行的操作之前，允许当前的线程一直阻塞。然后调用doAccept()方法，由于是异步操作，我们可以传递一个CompletionHandler<AsynchronousServerSocketChannel,? super A>类型的handler实例接受accept操作成功的通知消息。

![70 1][]

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);
    		result.read(buffer, buffer, new ReadCompletionHandler(result));
    	}
    
    	@Override
    	public void failed(Throwable exc, AsyncTimeServerHandler attachment) {
    		exc.printStackTrace();
    		attachment.latch.countDown();
    	}
    }

CompletionHandler有两个方法：

1.public void completed(AsynchronousSocketChannel result, AsyncTimeServerHandler attachment)；

首先从attachment获取成员变量AsynchronousServerSocketChannel然后继续调用accept方法。（调用AsynchronousServerSocketChannel的accept方法后，如果有新的客户端连接接入，系统将回调我们传入的CompletionHandler实例的completed方法，表示新的客户端已经接入成功。因为一个AsynchronousServerSocketChannel可以接收成千上万个客户端，所以需要继续调用它的accept方法，接收其他的客户端连接，最终形成一个循环，每当接受一个客户读连接成功之后，再异步接收新的客户端连接。）

然后创建新的ByteBuffer分配1M的缓冲区。然后调用AsynchronousSocketChannel的read方法进行异步读操作。

ByteBuffer dst：接收缓冲区，用于从异步Channel中读取数据包；

A attachment：异步Channel携带的附件，通知回调的时候作为入参使用；

CompletionHandler<Integer,? super A> handler：接收通知回调的业务Handler，在本例中是ReadCompletionHandler。

2.public void failed(Throwable exc, AsyncTimeServerHandler attachment)；

当异常发生时对异常Trowable进行判断，并且关闭链路释放资源。

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) {
    		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("服务器接收到的消息 : " + req);
    			String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(req)
    					? new java.util.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 && 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 buffer) {
    					// 如果没有发送完成，继续发送
    					if (buffer.hasRemaining())
    						channel.write(buffer, buffer, this);
    				}
    
    				@Override
    				public void failed(Throwable exc, ByteBuffer attachment) {
    					try {
    						channel.close();
    					} catch (IOException e) {
    						// ingnore on close
    					}
    				}
    			});
    		}
    	}
    }

这部分与之前的NIO大致相同，主要看下他的write方法：

![70 2][]

通过判断是否有剩余字节可写，如果有说明没有发送完继续发送，直到发送成功。

客户端：

public class TimeClient {
    	public static void main(String[] args) {
    		int port = 8080;
    		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;
    
    	public AsyncTimeClientHandler(String host, int port) {
    		this.host = host;
    		this.port = port;
    		try {
    			client = AsynchronousSocketChannel.open();
    		} 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 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) {
    							buffer.flip();
    							byte[] bytes = new byte[buffer.remaining()];
    							buffer.get(bytes);
    							String body;
    							try {
    								body = new String(bytes, "UTF-8");
    								System.out.println("当前时间 : " + body);
    								latch.countDown();
    							} catch (UnsupportedEncodingException e) {
    								e.printStackTrace();
    							}
    						}
    
    						@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, 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();
    		}
    	}
    
    	@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();
    		}
    	}
    }

首先在构造方法中通过AsynchronousSocketChannel的open方法创建一个新的AsynchronousSocketChannel对象。然后在run方法中创建CountDownLatch对象进行等待，防止异步操作没有执行完就完成线程退出。然后通过AsynchronousSocketChannel的connect方法发起异步操作。

它有两个参数分别是：

A attachment：AsynchronousSocketChannel的附件，用于回调通知时作为入参被传递，调用者可自定义。

CompletionHandler<Void,? super A> handler：异步操作回调通知接口，由调用者实现。

在本例中，两个参数都为自身。

![70 3][]

在异步连接成功后会回调completed方法，在这个方法中我们会创建请求消息体，对其进行编码，然后复制到发送缓冲区writeBuffer中，调用AsynchronousSocketChannel的write方法进行异步写。与服务端类似，我们可以是实现new CompletionHandler<Integer, ByteBuffer>接口用于写操作完成后的回调。如果发送缓冲区中任由尚未发送的字节，将继续异步发送，如果已经发送完成，则执行异步读取操作。

![70 4][]

由于read操作是异步的，所以我们通过匿名类实现new CompletionHandler<Integer, ByteBuffer>接口当读取完成被JDK回调是，构造应答消息。从CompletionHandler的ByteBuffer中读取应答消息然后打印。

![70 5][]

![70 6][]

![70 7][]

## 不同I/O模型对比： ##

![70 8][]

![70 9][]

## 不选择Java原生NIO编程的原因： ##

1.NIO的类库和API繁杂，使用麻烦，需要熟练掌握Selector、ServerSocketChannel、SocketChannel、ByteBuffer等。

2.需要具备其他额外技能做铺垫，如熟悉Java多线程编程。

3.可靠性能力补齐，工作量和难度都非常大。

4.JDK NIO的BUG，例如epoll bug，它会导致Selector空轮询，导致CPU 100%。

## 为什么选择 Netty： ##

1.API使用简单，开发门槛低；

2.功能强大，预置了多种编解码功能，支持多种主流协议；

3.定制能力强，可以通过ChannelHandler对通信框架进行灵活地扩展；

4.性能高，通过与其他业界主流的NIO框架对比，Netty综合性能最优；

5.成熟、稳定，Netty修复了已经发现的所有JDK NIO BUG；

6.社区活跃，版本迭代周期短。

7.经历大规模的商业应用考验，质量得到验证。

[70]: /images/20220505/5ce8d8908a3045cc8fce28e84e694ea9.png
[70 1]: /images/20220505/9a138e9adb6e412b963287aaec0c2a68.png
[70 2]: /images/20220505/7c04b8982e884a01a55b5577cc5e2c54.png
[70 3]: /images/20220505/23db60309a87478f9ecf567db75e7b0d.png
[70 4]: /images/20220505/72e9614a840b40eea9fe579010a87ded.png
[70 5]: /images/20220505/3b75b0e030c14676a088f8a5e1f1bff0.png
[70 6]: /images/20220505/0bd9d5f74beb4a5aac7fc8b29f2f0ed5.png
[70 7]: /images/20220505/fb8eb81a7e324b9384fc178c7520b9ed.png
[70 8]: /images/20220505/b23fd0c470e14e04a598b3127812e8f8.png
[70 9]: /images/20220505/8f60545997034ad8bfa2575686078f08.png