Java 网络编程 —— 异步通道和异步运算结果
从 JDK7 开始，引入了表示异步通道的 AsynchronousSockerChannel 类和 AsynchronousServerSocketChannel 类，这两个类的作用与 SocketChannel 类和 ServerSockelChannel 相似，区别在于异步通道的一些方法总是采用非阻塞模式，并且它们的非阻塞方法会立即返回一个 Future 对象，用来存放方法的异步运算结果
AsynchronousSocketChannel 类有以下非阻塞方法：
// 连接远程主机
Future<Void> connect(SocketAddress remote);
// 从通道中读入数据,存放到ByteBuffer中
// Future对象包含了实际从通道中读到的字节数
Future<Inleger> read(ByteBuffer dst);
// 把ByteBuffer的数据写入通道
// Future对象包含了实际写入通道的字节数
Future<Integer> write(ByteBuffer src);
AsynchronousServerSocketChannel 类有以下非阻塞方法：
// 接受客户连接请求
// Future对象包含连接建立成功后创建的AsynchronousSockelChannel对象
Future<AsynchronousSocketChannel> accept();
使用异步通道，可以使程序并行执行多个异步操作，例如：
SocketAddress socketAddress = ...;
AsynchronousSocketChannel client = AsynchronousSocketChannel.open();
//请求建立连接
Future<Void> connected = client.connect(socketAddress);
ByteBuffer byteBuffer = ByteBuffer.allocate(128);
//执行其他操作
//...
//等待连接完成
connected.get();
//读取数据
Future<Integer> future = client.read(byteBuffer);
//执行其他操作
//...
//等待从通道读取数据完成
future.get();
byteBuffer.flip();
WritableByteChannel out = Channels.newChannel(System.out);
out.write(byteBuffer);
下例的代码演示了异步通道的用法，它不断接收用户输入的域名并尝试建立连接，最后打印建立连接所花费的时间。如果程序无法连接到指定的主机，就打印相关错误信息。如果用户输入 bye，就结束程序
//表示连接一个主机的结果
class PingResult {
InetSocketAddress address;
long connectStart; //开始连接时的时间
long connectFinish = 0; //连接成功时的时间
String failure;
Future<Void> connectResult; //连接操作的异步运算结果
AsynchronousSocketChannel socketChannel;
String host;
final String ERROR = "连接失败";
PingResult(String host) {
try {
this.host = host;
address = new InetSocketAddress(InetAddress.getByName(host), 80);
} catch (IOException x) {
failure = ERROR;
}
}
//打印连接一个主机的执行结果
public void print() {
String result;
if (connectFinish != 0) {
result = Long.toString(connectFinish - connectStart) + "ms";
} else if (failure != null) {
result = failure;
} else {
result = "Timed out";
}
System,out,println("ping "+ host + "的结果" + ":" + result);
}
public class PingClient {
//存放所有PingResult结果的队列
private LinkedList<PingResult> pingResults = new Linkedlist<PingResult>();
boolean shutdown = false;
ExecutorService executorService;
public PingClient() throws IOException {
executorService = Executors.newFixedThreadPool(4);
executorService.execute(new Printer());
receivePingAddress();
}
}
public static void main(String args[]) throws IOException {
new PingClient();
}
/*接收用户输入的主机地址，由线程池执行PingHandler任务 */
public void receivePingAddress() {
try {
BufferedReader localReader = new BufferedReader(new InputStreamReader(System.in));
String msg = null;
//接收用户输入的主机地址
while((msg = localReader.readLine()) != null) {
if(msg.equals("bye")) {
shutdown = true;
executorService.shutdown();
break;
}
executorService.execute(new PingHandler(msg));
}
} catch(IOException e) {}
}
/* 尝试连接特定主机,生成一个PingResult对象,把它加入PingResults结果队列中 */
public class PingHandler implements Runnable {
String msg;
public PingHandler(String msg) {
this.msg = msg;
}
public void run() {
if(!msg.equals("bye")) {
PingResult pingResult = new PingResult(msg);
AsynchronousSocketChannel socketChannel = null;
try {
socketChannel = AsynchronousSocketChannel.open();
pingResult.connectStart = System.currentTimeMillis();
synchronized (pingResults) {
//向pingResults队列加入一个PingResult对象
pingResults.add(pingResult);
pingResults,notify();
}
Future<Void> connectResult = socketChannel.connect(pingResult.address);
pingResult.connectResult = connectResult;
} catch (Exception x) {
if (socketChannel != null) {
try { socketChannel.close();} catch (IOException e) {)
}
pingResult.failure = pingResult.ERROR;
}
}
}
}
/* 打印PingResults结果队列中已经执行完毕的任务的结果 */
public class Printer implements Runnable {
public void run() {
PingResult pingResult = null;
while(!shutdown) {
synchronized (pingResults) {
while (!shutdown && pingResults.size() == 0 ) {
try {
pingResults.wait(100);
} catch(InterruptedException e) {
e.printStackTrace();
}
}
if(shutdown && pingResults.size() == 0 ) break;
pingResult = pingResults.getFirst();
try {
if(pingResult.connectResult != null) {
pingResult.connectResult.get(500, TimeUnit,MILLISECONDS);
} catch(Exception e) {
pingResult.failure = pingResult.ERROR;
}
}
if(pingResult.connectResult != null && pingResult.connectResult.isDone()) {
pingResult.connectFinish = System.currentTimeMillis();
}
if(pingResult,connectResult != null && pingResult.connectResult.isDone() || || pingResult,failure != null) {
pingResult.print();
pingResults.removeFirst();
try {
pingResult.socketChannel.close();
} catch (IOException e) {}
}
}
}
}
}
}
PingClient 类定义了两个表示特定任务的内部类：
PingHandler：负责通过异步通道去尝试连接客户端输入的主机地址，并且创建一个PingResult对象，它包含了连接操作的异步运算结果，再将其加入PingResults结果队列Printer：负责打印PingResults结果队列已经执行完毕的任务结果，打印完毕的PingResult对象会从队列中删除