Netty-入门AIO编程

在介绍Netty编程之前， 我们不得不限介绍一下Unix系统下的事件驱动IO模型；在JDK1.7之后出现的NIO2.0也引入了这个异步非阻塞通道，他不需要通过多路复用器对注册通道进行轮询操作即可实现异步读写，从而简化了NIO的编程模型；NIO2.0提供的异步的套接字通道是真正的异步I/O，在异步I/O操作的时候可以传递信号变量，当操作完成之后回调相关的方法。

 

异步I/O也被称作AIO。

 

 

下面介绍一下AIO程序，类图如下：

 

 

 

源码：

 

package com.techstar.aio;

public class AIOTimeServer {

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

 

package com.techstar.aio;

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 asyncServerSocketChannel = null;
	
	
	
	public AsyncTimeServerHandler(int port) {
		this.port = port;
		try {
			this.asyncServerSocketChannel = AsynchronousServerSocketChannel.open();
			this.asyncServerSocketChannel.bind(new InetSocketAddress(this.port));
			System.out.println("The time server  is started! the port is " + this.port);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}



	@Override
	public void run() {
		
		latch = new CountDownLatch(1);
		this.doAccept();
		
		try {
			latch.await();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}
	
	private void doAccept() {
		this.asyncServerSocketChannel.accept(this, new AcceptCompletionHandler());
	}

}

 

package com.techstar.aio;

import java.nio.ByteBuffer;
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) {
		attachment.asyncServerSocketChannel.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();
	}

}

 

package com.techstar.aio;

import java.io.IOException;
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) {
		super();
		if (this.channel == null)
			this.channel = channel;
	}

	@Override
	public void completed(Integer result, ByteBuffer attachment) {

		try {
			attachment.flip();
			byte[] body = new byte[attachment.remaining()];
			attachment.get(body);
			String order = new String(body, "UTF-8");
			System.out.println("the client request Order = " + order );
			
			String currentTime = "QUERY TIME ORDER".equals(order)?new Date(System.currentTimeMillis()).toString() : "BAD ORDER";
			
			this.doWrite(currentTime);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}

	@Override
	public void failed(Throwable exc, ByteBuffer attachment) {
		try {
			channel.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		
	}
	

	private void doWrite(String response) {
		if (response != null && response.trim().length() > 0) {
			byte[] msg = response.getBytes();
			
			ByteBuffer writeBuffer = ByteBuffer.allocate(msg.length);
			writeBuffer.put(msg);
			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) {
						e.printStackTrace();
					}
					
				}
			});
		}
	}

}

 

 

 

 

 

 

客户端代码

package com.techstar.aio;

public class AIOTimeClient {

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

 

package com.techstar.aio;

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 int port;

	private String host;

	private AsynchronousSocketChannel channel;

	private CountDownLatch latch;

	public AsyncTimeClientHandler(int port, String host) {
		this.port = port;
		this.host = host;

		try {
			channel = AsynchronousSocketChannel.open();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	@Override
	public void run() {
		this.latch = new CountDownLatch(1);

		this.channel.connect(new InetSocketAddress(this.host, port), this, this);
		System.out.println("connect to the server。。。");
		try {
			this.latch.await();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		try {
			this.channel.close();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	@Override
	public void completed(Void result, AsyncTimeClientHandler attachment) {
		byte[] order = "QUERY TIME ORDER".getBytes();
		this.doCompleted(order);
	}

	@Override
	public void failed(Throwable exc, AsyncTimeClientHandler attachment) {
		try {
			this.channel.close();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

	}

	private void doCompleted(byte[] msg) {

		ByteBuffer writeBuffer = ByteBuffer.allocate(msg.length);
		writeBuffer.put(msg);
		writeBuffer.flip();
		channel.write(writeBuffer, writeBuffer, new CompletionHandler<Integer, ByteBuffer>() {

			@Override
			public void completed(Integer result, ByteBuffer attachment) {
				if (attachment.hasRemaining()) {
					try {
						System.out.println("发送消息指令:" + new String(msg, "UTF-8"));
					} catch (UnsupportedEncodingException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
					channel.write(attachment, attachment, this);
				} else {
					ByteBuffer readBuffer = ByteBuffer.allocate(1024);
					channel.read(readBuffer, readBuffer, new CompletionHandler<Integer, ByteBuffer>() {

						@Override
						public void completed(Integer result, ByteBuffer attachment) {
							attachment.flip();
							byte[] body = new byte[attachment.remaining()];
							
							attachment.get(body);
							String time;
							try {
								time = new String(body, "UTF-8");
								System.out.println("Now is:" + time);
								latch.countDown();
							} catch (UnsupportedEncodingException e) {
								e.printStackTrace();
							}
						}

						@Override
						public void failed(Throwable exc, ByteBuffer attachment) {
							try {
								channel.close();
							} catch (IOException e) {
								e.printStackTrace();
							}
						}
						
					});
				}
			}

			@Override
			public void failed(Throwable exc, ByteBuffer attachment) {
				try {
					channel.close();
				} catch (IOException e) {
					e.printStackTrace();
				}

			}
		});
	}

}