当前位置 博文首页 > 一只小菜鸡：java AIO示例以及流程详解
服务器端：
package nio;
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 = AsynchronousServerSocketChannel.open();
//绑定端口号
asynchronousServerSocketChannel.bind(new InetSocketAddress(port));
System.out.println("the time server is start on port :" + port);
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void run() {
//是一个并发工具类，他允许一个线程等待另一个线程完成后再执行
//这里是为了防止主线程启动完成后关闭
latch = new CountDownLatch(1);
doAccept();
try {
//阻塞直到count down to 0
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public void doAccept(){
//该方法是异步的接受来自该通道的客户端的连接请求，连接成功后调用CompletionHandler的completed或者failed方法
asynchronousServerSocketChannel.accept(this, new AcceptCompletionHandler());
}
}?
客户端连接请求完成回调的handler：
package nio;
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) {
//服务端已经接收客户端成功了，为什么还要调用accept方法？因为一个channel可以接收成千上万个客户端
//当调用asynchronousServerSocketChannel.accept(this, new AcceptCompletionHandler())方法后，又有新的
//客户端连接接入，所以需要继续调用他的accept方法，接受其它客户端的接入，最终形成一个循环
attachment.asynchronousServerSocketChannel.accept(attachment,this);
ByteBuffer buffer = ByteBuffer.allocate(1024);
//异步读操作，参数的定义：第一个参数：接收缓冲区，用于异步从channel读取数据包；
//第二个参数：异步channel携带的附件，通知回调的时候作为入参参数，这里是作为ReadCompletionHandler的入参
//通知回调的业务handler，也就是数据从channel读到ByteBuffer完成后的回调handler，这里是ReadCompletionHandler
result.read(buffer, buffer, new ReadCompletionHandler(result));
}
@Override
public void failed(Throwable exc, AsyncTimeServerHandler attachment) {
exc.printStackTrace();
AsyncTimeServerHandler result = (AsyncTimeServerHandler) attachment;
result.latch.countDown();
}
}从channel写缓存ByteBuffer完成回调的handler：
package nio;
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;
public class ReadCompletionHandler implements CompletionHandler<Integer, ByteBuffer> {
private AsynchronousSocketChannel channel;
public ReadCompletionHandler(AsynchronousSocketChannel channel){
if(this.channel == null)
this.channel = channel;
}
/**
* 业务处理，从ByteBuffer读取业务数据，做业务操作
* @param result
* @param attachment
*/
@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("the time server receive order:" + req );
String currentTime = "query time order".equals(req) ? new Date(System.currentTimeMillis()).toString(): "bad order";
dowrite(currentTime);
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
/**
* 往客户端的写操作
* @param currentTime
*/
public void dowrite(String currentTime){
if(currentTime != null && currentTime.trim().length() > 0){
byte[] bytes = currentTime.getBytes();
//分配一个写缓存
ByteBuffer write = ByteBuffer.allocate(bytes.length);
System.out.println("reponsbody=" + currentTime);
//将返回数据写入缓存
write.put(bytes);
write.flip();
//将缓存写进channel
channel.write(write, write, 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，并释放与channel相关联的一切资源
channel.close();
} catch (IOException e) {
e.printStackTrace();
}
}
});
}
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
try {
//读，关闭channel，并释放与channel相关联的一切资源
this.channel.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}服务器端启动类：
package nio;
public class TimeServer {
/**
* 启动服务端，采用异步非阻塞模式
* @param args
*/
public static void main(String[] args) {
int port = 8010;
new Thread(new AsyncTimeServerHandler(8010), "AIO-AsyncTimeServerHandler-001").start();
}
}客户端请求服务端，同时也作为回调类：
package nio;
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 {
//初始化一个AsynchronousSocketChannel
client = AsynchronousSocketChannel.open();
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void run() {
latch = new CountDownLatch(1);
//连接服务端，并将自身作为连接成功时的回调handler
client.connect(new InetSocketAddress(host, port), this, this);
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
/**
* 连接服务端成功时的回调
* @param result
* @param attachment
*/
@Override
public void completed(Void result, AsyncTimeClientHandler attachment) {
//请求参数
byte [] req = "query time order".getBytes();
//分配写缓存区
ByteBuffer write = ByteBuffer.allocate(req.length);
//往写缓存区写请求body
write.put(req);
write.flip();
//将缓存中的数据写到channel，同时使用匿名内部类做完成后回调
client.write(write, write, new CompletionHandler<Integer, ByteBuffer>() {
@Override
public void completed(Integer result, ByteBuffer byteBuffer) {
//如果缓存数据中还有数据，接着写
if(byteBuffer.hasRemaining()){
client.write(byteBuffer, byteBuffer, this);
}else{
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
//读取服务端的返回到缓存，采用匿名内部类做写完缓存后的回调handler
client.read(readBuffer, readBuffer, new CompletionHandler<Integer, ByteBuffer>() {
/**
* 从缓存中读取数据，做业务处理
* @param result
* @param buffer
*/
@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("now body is:" + body);
latch.countDown();
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
/**
* 从缓存读取数据失败
* 关闭client，释放channel相关联的一切资源
* @param exc
* @param attachment
*/
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
try {
client.close();
latch.countDown();
} catch (IOException e) {
e.printStackTrace();
}
}
});
}
}
/**
* 缓存写入channel失败
* 关闭client，释放channel相关联的一切资源
* @param exc
* @param attachment
*/
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
{
try {
client.close();
latch.countDown();
} catch (IOException e) {
e.printStackTrace();
}
}
}
});
}
/**
* 连接服务端失败
* @param exc
* @param attachment
*/
@Override
public void failed(Throwable exc, AsyncTimeClientHandler attachment) {
{
try {
client.close();
latch.countDown();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}客户端启动类：
package nio;
public class TimeClient {
/**
* 启动客户端，采用异步非阻塞模式
* @param args
*/
public static void main(String[] args) {
String host = "127.0.0.1";
int port = 8010;
new Thread(new AsyncTimeClientHandler(host, port), "AIO-AsyncTimeClientHandler-001").start();
}
}?
cs