mina中的AIO

Posted 2021-09-04 健康平安的活着

一 AIO

Nio2.0的异步套接字通道是真正的异步非阻塞io，Java nio 2.0的主要改进就是引入了异步IO（包括文件和网络）。

二 AIO案例

1.client

package com.cf.demo.aio2;

public class AioClient {

	public static void main(String[] args) {
		int port = 7080;
		new Thread(new AioClientAsyncHandler("127.0.0.1", port), "Client-001").start();;
	}
}

 

package com.cf.demo.aio2;


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 AioClientAsyncHandler implements CompletionHandler<Void, AioClientAsyncHandler>, Runnable {
	private String host;
	private int port;
	private CountDownLatch latch;
	private AsynchronousSocketChannel client;
	
	public AioClientAsyncHandler(String host, int port) {
		this.host = host;
		this.port = port;
		try {
			client = AsynchronousSocketChannel.open();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	public void run() {
		latch = new CountDownLatch(1);
		client.connect(new InetSocketAddress(host, port), this, this);
		
		try {
			latch.await();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		
		try {
			client.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public void completed(Void result, AioClientAsyncHandler attachment) {
		byte[] request = "jack".getBytes();
		ByteBuffer writeBuffer = ByteBuffer.allocate(request.length);
		writeBuffer.put(request);
		writeBuffer.flip();
		
		client.write(writeBuffer, writeBuffer, new CompletionHandler<Integer, ByteBuffer>() {

			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>() {

						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("client receive:" + body);
							} catch (UnsupportedEncodingException e) {
								e.printStackTrace();
							}
						}

						public void failed(Throwable exc, ByteBuffer buffer) {
							try {
								client.close();
								latch.countDown();
							} catch (IOException e) {
								e.printStackTrace();
							}
						}
					});
				}
				
			}

			public void failed(Throwable exc, ByteBuffer buffer) {
				try {
					client.close();
					latch.countDown();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
			
		});
	}

	public void failed(Throwable exc, AioClientAsyncHandler attachment) {
		try {
			client.close();
			latch.countDown();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
}

2.server

package com.cf.demo.aio2;

public class Aioserver {

	public static void main(String[] args) {
		int port = 7080;
		//创建异步的处理类
		new Thread(new AioServerAsyncHandler(port), "Aio-Server-001").start();
	}
}

package com.cf.demo.aio2;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.util.concurrent.CountDownLatch;

public class AioServerAsyncHandler implements Runnable {
	int port;
	CountDownLatch latch;
	AsynchronousServerSocketChannel asynchronousServerSocketChannel;
	
	public AioServerAsyncHandler(int port) {
		this.port = port;
		
		try {
			//异步服务端通道
			asynchronousServerSocketChannel = AsynchronousServerSocketChannel.open();
			//绑定监听端口
			asynchronousServerSocketChannel.bind(new InetSocketAddress(port));
			System.out.println("Server start:" + port);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
	public void run() {
		//CountDownLatch对象运行在完成一组正在执行的操作之前，当前线程一直阻塞
		latch = new CountDownLatch(1);
		//接收客户端的连接
		doAccept();
		try {
			latch.await();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	public void doAccept() {
		//传递一个实现CompletionHandler<AsynchronousSocketChannel, ?>类型的handler实例来接收accept操作成功的通知消息
		asynchronousServerSocketChannel.accept(this, new AioServerAcceptHandler());
	}
}