JDK7网络异步IO

在Reactor模式中，虽然可以采用non-blocking I/O模式，使用Selector注册感兴趣的I/O事件和读取感兴趣的I/O事件，I/O调用者向I/O系统请求一个I/O调用时，I/O立即返回给调用者一个反馈，这些反馈无外乎两大类型，请求已经被执行并且有结果返回，或者当前的通道缓存中无数据可用。第二种情况下，为保险起见，编写程序时需要写一个循环只到有数据被读取为止。在I/O系统处理一个I/O请求时，不可能同时处理第二个I/O操作。

 

异步I/O与同步I/O的最大不同是：当调用者向异步I/O系统请求一个I/O操作的时候，I/O系统自身会调用系统资源启动一个线程去处理该请求，当这个请求被处理完成后，通知调用者，并返回结果给调用者。在这个过程之中，调用者在自身的线程中可以去干一些别的工作。这种I/O处理方式提高了应用程序的可扩展性和性能。

 

JDK7中的NIO2版本引入了异步IO的功能。支持文件异步I/O操作和网络异步I/O操作。针对于网络部分的主要3个类和一个接口。

 

主要的类

1,  AsynchronousChannelGroup，这个类与一组AsynchronousSocketChannel相关联，并且与I/O系统交互，当有I/O请求被处理完成后，会把处理结果通知给一个实现了CompletionHandler接口的对象派发给相关的AsynchronousSocketChannel. 这个类在在使用时需要制定一个ExecutorService对象，ExecutorService对象中的线程池是用来应对与I/O系统交互及与通知调用者时需要的线程开销。

 

2, AsynchronousServerSocketChannel，异步网络ServerSocket，主要用来绑定端口，并接受客户端的连接。accept可以返回一个Future<AsynchronousSocketChannel>对象，调用Future对象的get方法得到一个传入的套接字 的通道对象，如

AsynchronousServerSocketChannel listener = .....
listener.bind(new InetSocketAddress(9001));
Future<AsynchronousSocketChannel> future = listener.accept();
AsynchronousSocketChannel channel = future.get();
..

在Future对象中拿到传入的套接字通道对象后，才可以再次调用accept方法拿到新的套接字通道，否则会抛出AcceptPendingException，虽然是异步的，不过这个方法实质上类似于同步的。

 

另外一种拿到套接字通道对象的方法是使用CompletionHandler的回调方式。这种方式也是Sun鼓励的方式。一旦套接字通道建立完成，I/O系统会调用CompletionHandler对象的completed方法，把已经建立连接的套接字通道传入到completed方法的第一参数当中去。 这个通道建立后，可以让服务器AsynchronousServerSocketChannel对象再次接受新的套接字通道。接着可以用这个刚刚建立的套接字通道来进行数据通讯的工作。

listener.accept(null, new CompletionHandler<AsynchronousSocketChannel,Void>() {
      public void completed(AsynchronousSocketChannel ch, Void att) {
          // accept the next connection
          listener.accept(null, this);

          // handle this connection
          ....
          .....
      }
      public void failed(Throwable exc, Void att) {
          ...
      }
  });

 

 

3, AsynchronousSocketChannel，真正的数据通讯类，使用write方法向channel的缓冲中写入数据，使用read方法读取channel缓冲中的数据，读和写的I/O操作完成后，I/O系统会发出一个通知，把读取、写入的字节数传入到CompletionHandler<Integer, A>对象中的completed方法的第一个参数。与AsynchronousServerSocketChannel的accept方法类型，如果一个I/O操作完成，而另外一个I/O操作又发起的话，会抛出PendingReadException或者PendingWriteException.因此在用AsynchronousSocketChannel处理I/O操作时，要非常小心，确保一个I/O操作发起前，要确保这个通道上的上一个I/O操作已经完成。

 

 

4, CompletionHandler接口，实现这个接口的对象在异步I/O操作中承担着通知的作用，当操作完成后，I/O系统会把通知的内容发送到completed方法中的第一个参数，并且I/O系统会调用completed方法。CompletionHandler按照使用场合可以分成3种：

AsynchronousServerSocket类中中用于接受传入连接套接字通道后的接入套接字通道完成通知，采用泛形声明的方式

CompletionHandler<AsynchronousSocketChannel, A>

 

AsynchronousSocketChannel写数据后的写入完成通知

CompletionHandler<Integer, A>

 

AsynchronousSocketChannel读取数据后的读取完成通知，如写入通知的声明相同。

 

异步读和写的问题

使用通道来读取数据时，首先是从通道的缓冲区读取一定量的数据到ByteBuffer中目标缓存中，如果目标缓存没有剩余的字节可供写入，直接返回零给读取通知的completed方法的第一个参数，如果这个参数的值为-1，表示缓冲区中没有可供读取的字节或者当前通道中的套接字输入流已经关闭。因为读写都是异步的，所以一端的通道进行I/O操作后如果立即关闭，另外一个端的通道很有可能仍然处于相对应的I/O操作过程之中，这样的情形下，会抛出一个远程通道被强制关闭的IOException。因此，在通道接收数据时，需要根据completed方法的第一个参数来判断是否需要关闭当前的通道，如果不能及时关闭通道，容易引起内存泄露的问题。

 

I/O系统何时把通道缓冲区中的字节传输到远程通道中的缓冲区里，这个是由I/O系统决定，而对于异步I/O的开发人员来说，是不可控制的，假如代码在客户端向通道里写入了3个字节序列，这3个序列在远程的接收端通道里可能会被接收成3个同样的字节序列，也有可能是2个序列，也有可能是1个序列，但是不伦是接受到几个序列，这些接受到的序列的内容和发送序列的内容是一样的。

 

无论读和写，都是不是线程安全的，必须等到I/O系统通知调用者后，才可以进行下一个操作，这个在多线程环境下，需要格外小心。

 

异步Echo Server

public class AsynEchoServer {

	private int port = 9999;

	private int backlog = 50;

	private int threadPoolSize = 20;

	private int initialSize = 5;

	public void start() throws IOException {
		ExecutorService executor = Executors.newFixedThreadPool(threadPoolSize);
		AsynchronousChannelGroup group = AsynchronousChannelGroup.withCachedThreadPool(executor, initialSize);
		AsynchronousServerSocketChannel listener = AsynchronousServerSocketChannel.open(group);
		listener.bind(new InetSocketAddress(port), backlog);
		listener.accept(listener, new CompletionHandler<AsynchronousSocketChannel, AsynchronousServerSocketChannel>() {

			@Override
			public void completed(AsynchronousSocketChannel channel, AsynchronousServerSocketChannel listener) {
				listener.accept(listener, this);
				ByteBuffer buffer = ByteBuffer.allocate(512);
				channel.read(buffer, buffer, new EchoHandler(channel, buffer));
			}

			@Override
			public void failed(Throwable exc, AsynchronousServerSocketChannel listener) {
				exc.printStackTrace();
				try {
					listener.close();
				} catch (IOException e) {
					e.printStackTrace();
				} finally {
					System.exit(-1);
				}
			}
		});
	}

	/**
	 * @param args
	 * @throws IOException
	 */
	public static void main(String[] args) throws IOException {
		AsynEchoServer server = new AsynEchoServer();
		server.start();
	}

	// getter & setters
}

 

class EchoHandler implements CompletionHandler<Integer, ByteBuffer> {
	private static Charset utf8 = Charset.forName("utf-8");
	AsynchronousSocketChannel channel;
	ByteBuffer buffer;

	public EchoHandler(AsynchronousSocketChannel channel, ByteBuffer buffer) {
		this.channel = channel;
		this.buffer = buffer;
	}

	@Override
	public void completed(Integer result, ByteBuffer buff) {
		if (result == -1) {
			try {
				channel.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		} else if (result > 0) {
			buffer.flip();
			String msg = utf8.decode(buffer).toString();
			System.out.println("echo: " + msg);
			Future<Integer> w = channel.write(utf8.encode(msg));
			try {
				w.get();
			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (ExecutionException e) {
				e.printStackTrace();
			}
			
			buffer.clear();
			channel.read(buff, buff, this);
		}
	}

	@Override
	public void failed(Throwable exc, ByteBuffer buff) {
		// TODO Auto-generated method stub
	}
}

 

EchoClient

class EchoClient {
	private static Charset utf8 = Charset.forName("utf-8");
	private int port = 9999;
	private String remoteHost;
	private String[] message;
	private AsynchronousSocketChannel channel;

	public static void main(String args[]) throws Exception {
		if (args.length >= 2) {
			String msgs[] = new String[args.length - 1];
			System.arraycopy(args, 1, msgs, 0, msgs.length);
			EchoClient client = new EchoClient(args[0], msgs);
			client.connect();
			client.sendAndReceive();
			Thread.sleep(3000);
			client.close();
			Thread.sleep(3000);
		} else {
			System.out.println("usage EchoClient [remotehost] [messages .... ]");
		}
	}

	public void sendAndReceive() throws InterruptedException, ExecutionException {
		ByteBuffer buffer = ByteBuffer.allocate(512);
		for (String msg : this.message) {
			Future<Integer> w = channel.write(utf8.encode(msg));
			w.get();
		}
		
		channel.read(buffer, buffer, new ReceiverHandler(channel, buffer));
	}

	public void close() throws IOException {
		channel.shutdownInput();
		channel.shutdownOutput();
	}

	public EchoClient(String remoteHost, String[] message) {
		super();
		this.remoteHost = remoteHost;
		this.message = message;
	}

	public void connect() throws IOException, InterruptedException, ExecutionException {
		channel = AsynchronousSocketChannel.open();
		Future<Void> r = channel.connect(new InetSocketAddress(this.remoteHost, this.port));
		r.get();
	}

	//  getter & setters
}

 

class ReceiverHandler implements CompletionHandler<Integer, ByteBuffer> {
	private static Charset utf8 = Charset.forName("utf-8");
	private AsynchronousSocketChannel channel;
	private ByteBuffer buffer;

	@Override
	public void completed(Integer result, ByteBuffer buff) {
		
		if (result > 0) {			
			buffer.flip();
			System.out.println(utf8.decode(buffer));
			buffer.clear();
			channel.read(buff, buff, this);
		}else if (result==-1){
			try {
				channel.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	@Override
	public void failed(Throwable exc, ByteBuffer buff) {
		exc.printStackTrace();
	}

	public ReceiverHandler(AsynchronousSocketChannel channel, ByteBuffer buffer) {
		super();
		this.channel = channel;
		this.buffer = buffer;
	}
}

 

 

参考资料

http://www.artima.com/lejava/articles/more_new_io.html

http://www.kegel.com/c10k.html