package joanna.yan.aio;public class TimeServer {public static void main(String[] args) {int port=9090;if(args!=null&&args.length>0){try {port=Integer.valueOf(args[0]);} catch (Exception e) {// 采用默认值
            }}AsyncTimeServerHandler timeServer=new AsyncTimeServerHandler(port);new Thread(timeServer,"AIO-AsyncTimeServerHandler-001").start();}
}

package joanna.yan.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 asynchronousServerSocketChannel;public AsyncTimeServerHandler(int port){this.port=port;try {//创建一个异步的服务端通道AsynchronousServerSocketChannelasynchronousServerSocketChannel=AsynchronousServerSocketChannel.open();asynchronousServerSocketChannel.bind(new InetSocketAddress(port));System.out.println("The time server is start in port: "+port);} catch (IOException e) {e.printStackTrace();}}@Overridepublic void run() {/**初始化CountDownLatch对象。 *它的作用是，在完成一组正在执行的操作之前，允许当前的线程一直阻塞。*在本例中，我们让线程在此阻塞，防止服务器执行完成退出。*在实际项目应用中，不需要启动独立的线程来处理AsynchronousServerSocketChannel，这里仅仅是个demo演示。*/latch=new CountDownLatch(1);doAccept();try {latch.await();} catch (InterruptedException e) {e.printStackTrace();}}/*** 接收客户端的连接。* 由于这里是异步操作。我们可以传递一个CompletionHandler<AsynchronousSocketChannel,? super A>类型* 的handler实例接收accept操作成功的通知消息*/public void doAccept() {asynchronousServerSocketChannel.accept(this, new AcceptCompletionHandler());}
}

package joanna.yan.aio;import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
/*** 接收accept操作成功的通知消息* @author Administrator**/
public class AcceptCompletionHandler implements CompletionHandler<AsynchronousSocketChannel,AsyncTimeServerHandler>{@Overridepublic void completed(AsynchronousSocketChannel result,AsyncTimeServerHandler attachment) {/** 疑惑：既然已经接收客户端成功了，为什么还要再次调用accept方法呢？* 原因：当我们调用AsynchronousServerSocketChannel的accept方法后，如果有新的客户端连接接入，* 系统将回调我们传入的CompletionHandler实例的completed方法，表示新的客户端已经接入成功，* 因为一个AsynchronousServerSocketChannel可以接收成千上万个客户端，所以我们需要继续调用它的accep方法，* 接收其他的客户端连接，最终形成一个循环。* 每当接收一个客户读连接成功之后，再异步接收新的客户端连接。*/attachment.asynchronousServerSocketChannel.accept(attachment, this);ByteBuffer buffer=ByteBuffer.allocate(1024);/** 参数一：接收缓冲区，用于从异步Channel中读取数据包；* 参数二：异步Channel携带的附件，通知回调的时候作为入参使用；* 参数三：接收通知回调的业务handler*/result.read(buffer, buffer, new ReadCompletionHandler(result));}@Overridepublic void failed(Throwable exc, AsyncTimeServerHandler attachment) {exc.printStackTrace();attachment.latch.countDown();}
}

package joanna.yan.aio;import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.Date;
/*** 主要用于读取半包消息和发送应答。* 本例不对半包读写进行具体说明，在后面的Netty半包处理中会介绍。* @author Administrator**/
public class ReadCompletionHandler implements CompletionHandler<Integer, ByteBuffer>{private AsynchronousSocketChannel channel;public ReadCompletionHandler(AsynchronousSocketChannel channel){if(this.channel==null){this.channel=channel;}}@Overridepublic 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("The time server receive order : "+req);String currentTime="QUERY TIME ORDER".equalsIgnoreCase(req) ? new Date(System.currentTimeMillis()).toString() : "BAD　ORDER";doWrite(currentTime);} catch (UnsupportedEncodingException e) {e.printStackTrace();}}/*** 发生异常的时候调用。* 对异常Throwable进行判断，如果是I/O异常，就关闭链路，释放资源；* 如果是其它异常，按照业务自己的逻辑进行处理。* 本例作为简单demo，没有对异常进行分类判断，只要发生了读写异常，就关闭链路，释放资源。*/@Overridepublic 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>() {@Overridepublic void completed(Integer result, ByteBuffer buffer) {//如果没有发送完成，继续发送if(buffer.hasRemaining()){channel.write(buffer, buffer, this);}}@Overridepublic void failed(Throwable exc, ByteBuffer attachment) {try {channel.close();} catch (IOException e) {//ingnore on close
                    }}});}}
}

package joanna.yan.aio;public class TimeClient {public static void main(String[] args) {int port=9090;if(args!=null&&args.length>0){try {port=Integer.valueOf(args[0]);} catch (Exception e) {// 采用默认值
            }}/** 通过一个独立的I/O线程创建异步时间服务器客户端handler。* 在实际项目中，我们不需要独立的线程创建异步连接对象，因为底层都是通过JDK的系统回调实现的。*/new Thread(new AsyncTimeClientHandler("127.0.0.1", port),"AIO-AsyncTimeClientHandle-001").start();/** 需要指出的是，正如之前的NIO例程，我们并没有完整的处理网络的半包读写，在对例程进行功能测试的是还没有问题，* 但是，如果对代码稍加改造，进行压力或者性能测试，就会发现输出结果存在问题。* 这里只集中将NIO的入门知识，后面会详细讲到半包读写*/}
}

package joanna.yan.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 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();}}@Overridepublic 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>() {//用于写操作完成后的回调
@Overridepublic 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>(){//当读取完成被JDK回调时，构造应答消息。
@Overridepublic 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("Now is : "+body);latch.countDown();} catch (UnsupportedEncodingException e) {e.printStackTrace();}}@Overridepublic void failed(Throwable exc, ByteBuffer attachment) {try {client.close();//让AsyncTimeClientHandler线程执行完毕，客户端退出执行
                                latch.countDown();} catch (IOException e) {//ingnore on close
                            }}   });}}@Overridepublic void failed(Throwable exc, ByteBuffer attachment) {try {client.close();latch.countDown();} catch (IOException e) {//ingnore on close
                }}});}@Overridepublic void failed(Throwable exc, AsyncTimeClientHandler attachment) {exc.printStackTrace();try {client.close();latch.countDown();} catch (IOException e) {//ingnore on close
        }}@Overridepublic void run() {//创建CountDownLatch进行等待，防止异步操作没有执行完成线程就退出。latch=new CountDownLatch(1);/** 参数二：AsynchronousSocketChannel的附件，用于回调通知时作为入参被传递，调用者可以自定义* 参数三：异步参数回调通知接口，由调用者实现。*/client.connect(new InetSocketAddress(host, port), this, this);try {latch.await();} catch (InterruptedException e) {e.printStackTrace();}try {client.close();} catch (IOException e) {e.printStackTrace();}}
}