Java——BIO、NIO、AIO
在工作中，无论什么项目我们都会用到网络传输，tomcat等server服务器、Dubbo等Rpc协议框架、Redis等NOsql数据库、各种MQ的中间件等等。这篇文章是学习《Netty权威指南》后的总结，从Java基础的BIO、NIO、AIO进行学习。
在说JavaIO以前，先看看unix提供的5种IO模型：
1，阻塞IO模型：
2，非阻塞IO模型：
3，IO复用模型：
4，信号驱动IO模型：
5，异步IO模型：
小结：最基本的1对1IO阻塞模型，通过使用轮询调用、selector管家进行协调、buffer缓存空间进行解耦、信号进行主动通知等方面进行优化实现的各种模型。
一，Java最基础的BIO，网络编程基于Client/Server模型，BIO就是其中最基础的阻塞模型，通过ServerSocket和Socket建立通道，进行通讯。下边我们以“查询服务器时间”例子，来进行实现。
public class TimeServer {
public static void main(String[] args) throws IOException {
int port = 8090;
if (args != null && args.length > 0) {
try {
port = Integer.valueOf(args[0]);
}catch (NumberFormatException e){
}
}
ServerSocket serverSocket = null;
try {
serverSocket = new ServerSocket(port);
System.out.println("The time server is start in port : " + port);
Socket socket =null;
while (true){
socket =serverSocket.accept();
new Thread(new TimeServerHandler(socket)).start();
}
}finally {
if(serverSocket != null){
System.out.println("The time server close");
serverSocket.close();
serverSocket =null;
}
}
}
}
public class TimeServerHandler implements Runnable {
private Socket socket;
public TimeServerHandler(Socket socket) {
this.socket = socket;
}
@Override
public void run() {
BufferedReader in = null;
PrintWriter out = null;
try {
in = new BufferedReader(new InputStreamReader(this.socket.getInputStream()));
out = new PrintWriter(this.socket.getOutputStream(), true);
String currentTime = null;
String body = null;
while (true) {
body = in.readLine();
if (body == null) {
break;
}
System.out.println("the time server receive order : " + body);
currentTime = "query time order".equalsIgnoreCase(body) ? new Date(System.currentTimeMillis()).toString() : "bad order";
out.println(currentTime);
}
} catch (Exception e) {
if (in != null) {
try {
in.close();
} catch (IOException e1) {
e1.printStackTrace();
}
}
if (out != null) {
out.close();
out = null;
}
if (this.socket != null) {
try {
this.socket.close();
} catch (IOException e1) {
e1.printStackTrace();
}
this.socket = null;
}
} finally {
}
}
}
public class TimeClient {
public static void main(String[] args) {
int port = 8090;
if(args !=null && args.length >0){
try {
port = Integer.valueOf(args[0]);
}catch (NumberFormatException e){
}
}
Socket socket =null;
BufferedReader in =null;
PrintWriter out = null;
try {
socket =new Socket("127.0.0.1",port);
in =new BufferedReader(new InputStreamReader(socket.getInputStream()));
out = new PrintWriter(socket.getOutputStream(),true);
out.println("QUERY TIME ORDER");
System.out.println("Send order 2 server succeed.");
String resp = in.readLine();
System.out.println("Now is :" + resp);
} catch (UnknownHostException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}finally {
if(out !=null){
out.close();
out =null;
}
if(in !=null){
try {
in.close();
}catch (IOException e){
e.printStackTrace();
}
in = null;
}
if(socket !=null){
try {
socket.close();
} catch (IOException e) {
e.printStackTrace();
}
socket = null;
}
}
}
}
二，伪异步IO：为了解决同步阻塞IO一个链路需要一个线程来处理的问题，可以利用引入线程池，可以使用线程池中的线程处理多个client客户端，提高线程的利用率。但是如果由于通选非常慢，耗尽了线程池中的线程，其它的client处理请求就会在队列中排队阻塞了，所以仅仅是理想情况下，看起来异步了。来看伪异步“查询服务器时间”的例子。
public class TimeServer {
public static void main(String[] args) throws IOException {
int port = 8090;
if (args != null && args.length > 0) {
try {
port = Integer.valueOf(args[0]);
}catch (NumberFormatException e){
}
}
ServerSocket serverSocket = null;
try {
serverSocket = new ServerSocket(port);
System.out.println("The time server is start in port : " + port);
Socket socket =null;
//thread pool
TimeServerHandlerExecutePool singleExecutePool = new TimeServerHandlerExecutePool(50,10000);
while (true){
socket =serverSocket.accept();
singleExecutePool.execute(new TimeServerHandler(socket));
}
}finally {
if(serverSocket != null){
System.out.println("The time server close");
serverSocket.close();
serverSocket =null;
}
}
}
}
public class TimeServerHandlerExecutePool {
private ExecutorService executorService;
public TimeServerHandlerExecutePool(int maxPoolSize, int queueSize) {
executorService = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(), maxPoolSize, 120L, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(queueSize));
}
public void execute(Runnable task) {
executorService.execute(task);
}
}
public class TimeServerHandler implements Runnable {
private Socket socket;
public TimeServerHandler(Socket socket) {
this.socket = socket;
}
@Override
public void run() {
BufferedReader in = null;
PrintWriter out = null;
try {
in = new BufferedReader(new InputStreamReader(this.socket.getInputStream()));
out = new PrintWriter(this.socket.getOutputStream(), true);
String currentTime = null;
String body = null;
while (true) {
body = in.readLine();
if (body == null) {
break;
}
System.out.println("the time server receive order : " + body);
currentTime = "query time order".equalsIgnoreCase(body) ? new Date(System.currentTimeMillis()).toString() : "bad order";
out.println(currentTime);
}
} catch (Exception e) {
if (in != null) {
try {
in.close();
} catch (IOException e1) {
e1.printStackTrace();
}
}
if (out != null) {
out.close();
out = null;
}
if (this.socket != null) {
try {
this.socket.close();
} catch (IOException e1) {
e1.printStackTrace();
}
this.socket = null;
}
} finally {
}
}
}
public class TimeClient {
public static void main(String[] args) {
int port = 8090;
if(args !=null && args.length >0){
try {
port = Integer.valueOf(args[0]);
}catch (NumberFormatException e){
}
}
Socket socket =null;
BufferedReader in =null;
PrintWriter out = null;
try {
socket =new Socket("127.0.0.1",port);
in =new BufferedReader(new InputStreamReader(socket.getInputStream()));
out = new PrintWriter(socket.getOutputStream(),true);
out.println("QUERY TIME ORDER");
System.out.println("Send order 2 server succeed.");
String resp = in.readLine();
System.out.println("Now is :" + resp);
} catch (UnknownHostException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}finally {
if(out !=null){
out.close();
out =null;
}
if(in !=null){
try {
in.close();
}catch (IOException e){
e.printStackTrace();
}
in = null;
}
if(socket !=null){
try {
socket.close();
} catch (IOException e) {
e.printStackTrace();
}
socket = null;
}
}
}
}
三，NIO（非阻塞IO）：在JDK1.4以后引入了NIO来解决BIO的一些问题，我们来看它是通过引入什么解决的（根据文章前边的几种IO模型联想一下）。
1，Buffer缓存区，IO的读写，都直接往Buffer中进行操作，这样极大的解耦了两端；
2，Channel（通道）：通道是双向的，可以既进行读又进行写两种操作，和流不一样必须是inputstream或者outputstream的子类。
3，Selector（多路复用器）：Selector会不断轮询注册在之上的Channel，如果Channel发生了读或者写事件就出处于就绪状态，Seletor就会轮询出来，然后进行处理。
好，通过“查询服务器时间”例子来看下。
public class TimeServer {
public static void main(String[] args) throws IOException {
int port = 8090;
if (args != null && args.length > 0) {
try {
port = Integer.valueOf(args[0]);
} catch (NumberFormatException e) {
}
}
MultiplexerTimeServer timeServer = new MultiplexerTimeServer(port);
new Thread(timeServer, "NIO-MultiplexerTimeServer-001").start();
}
}
public class MultiplexerTimeServer implements Runnable {
private Selector selector;
private ServerSocketChannel serverSocketChannel;
private volatile boolean stop;
/**
* 初始化多路复用器
*
* @param port
*/
public MultiplexerTimeServer(int port) {
try {
selector = Selector.open();
serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);
serverSocketChannel.socket().bind(new InetSocketAddress(port), 1024);
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
System.out.println("The time server is start in port : " + port);
} catch (IOException e) {
e.printStackTrace();
System.exit(1);
}
}
public void stop() {
this.stop = true;
}
@Override
public void run() {
while (!stop) {
try {
selector.select(1000);
Set<SelectionKey> selectionKeySet = selector.selectedKeys();
Iterator<SelectionKey> it = selectionKeySet.iterator();
SelectionKey selectionKey = null;
while (it.hasNext()) {
selectionKey = it.next();
it.remove();
try {
handlerInput(selectionKey);
} catch (Exception e) {
if (selectionKey != null) {
selectionKey.cancel();
if (selectionKey.channel() != null) {
selectionKey.channel().close();
}
}
}
}
} catch (Throwable t) {
t.printStackTrace();
}
}
//多路复用器关闭后，所有注册上边的channel和pipe等资源都会被自动去注册并关闭，所以不必重复释放资源
if (selector != null) {
try {
selector.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
/**
* hold key the handler step
* @param key
* @throws IOException
*/
private void handlerInput(SelectionKey key) throws IOException {
if (key.isValid()) {
//处理新接入的请求消息
if (key.isAcceptable()) {
//accept new connection
ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
SocketChannel socketChannel = serverSocketChannel.accept();
socketChannel.configureBlocking(false);
//add the new connection to selector
socketChannel.register(selector, SelectionKey.OP_READ);
}
if (key.isReadable()) {
//read the data
SocketChannel socketChannel = (SocketChannel) key.channel();
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
int readBytes = socketChannel.read(readBuffer);
if (readBytes > 0) {
readBuffer.flip();
byte[] bytes = new byte[readBuffer.remaining()];
readBuffer.get(bytes);
String body = new String(bytes, "UTF-8");
System.out.println("the time server receive order: " + body);
String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new Date(System.currentTimeMillis()).toString() : "bad order";
doWrite(socketChannel, currentTime);
} else if (readBytes < 0) {
//对链路进行关闭
key.cancel();
socketChannel.close();
} else {
//read 0 byte data
}
}
}
}
/**
* socketChannel write response infomation
* @param socketChannel
* @param response
* @throws IOException
*/
private void doWrite(SocketChannel socketChannel, String response) throws IOException {
if (response != null && response.trim().length() > 0) {
byte[] bytes = response.getBytes();
ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
writeBuffer.put(bytes);
writeBuffer.flip();
socketChannel.write(writeBuffer);
}
}
}
public class TimeClient {
public static void main(String[] args) {
int port = 8090;
if(args !=null && args.length >0){
try {
port = Integer.valueOf(args[0]);
}catch (NumberFormatException e){
}
}
new Thread(new TimeClientHandler("127.0.0.1",port),"Timer-Client-001").start();
}
}
public class TimeClientHandler implements Runnable {
private String host;
private int port;
private Selector selector;
private SocketChannel socketChannel;
private volatile boolean stop;
/**
* 初始化客户端
* @param host
* @param port
*/
public TimeClientHandler(String host, int port) {
this.host = host == null ? "127.0.0.1" : host;
this.port = port;
try {
selector = Selector.open();
socketChannel = SocketChannel.open();
socketChannel.configureBlocking(false);
} catch (IOException e) {
e.printStackTrace();
System.exit(1);
}
}
@Override
public void run() {
try {
doConnect();
} catch (IOException e) {
e.printStackTrace();
System.exit(1);
}
//循环
while (!stop) {
try {
selector.select(1000);
Set<SelectionKey> selectionKeys = selector.selectedKeys();
Iterator<SelectionKey> it = selectionKeys.iterator();
SelectionKey selectionKey = null;
while (it.hasNext()) {
selectionKey = it.next();
it.remove();
try {
handlerInput(selectionKey);
} catch (Exception e) {
if (selectionKey != null) {
selectionKey.cancel();
if (selectionKey.channel() != null) {
selectionKey.channel().close();
}
}
}
}
} catch (IOException e) {
e.printStackTrace();
System.exit(1);
}
}
if (selector != null) {
try {
selector.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
/**
* handler selected key
* @param key
* @throws IOException
*/
private void handlerInput(SelectionKey key) throws IOException {
if (key.isValid()) {
SocketChannel socketChannel = (SocketChannel) key.channel();
if (key.isConnectable()) {
if (socketChannel.finishConnect()) {
socketChannel.register(selector, SelectionKey.OP_READ);
doWrite(socketChannel);
} else {
//连接失败
System.exit(1);
}
}
if (key.isReadable()) {
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
int readBytes = socketChannel.read(readBuffer);
if (readBytes > 0) {
readBuffer.flip();
byte[] bytes = new byte[readBuffer.remaining()];
readBuffer.get(bytes);
String body = new String(bytes, "Utf-8");
System.out.println("Now is : " + body);
this.stop = true;
} else if (readBytes < 0) {
key.cancel();
socketChannel.close();
} else {
}
}
}
}
/**
* 进行连接
*
* @throws IOException
*/
private void doConnect() throws IOException {
//如果直接连接成功，则注册到多路复用器上，发送请求数据读应答
if (socketChannel.connect(new InetSocketAddress(host, port))) {
socketChannel.register(selector, SelectionKey.OP_READ);
doWrite(socketChannel);
} else {
socketChannel.register(selector, SelectionKey.OP_CONNECT);
}
}
/**
* write data info
*
* @param socketChannel
* @throws IOException
*/
private void doWrite(SocketChannel socketChannel) throws IOException {
byte[] req = "QUERY TIME ORDER".getBytes();
ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);
writeBuffer.put(req);
writeBuffer.flip();
socketChannel.write(writeBuffer);
if (!writeBuffer.hasRemaining()) {
System.out.println("send order 2 server success");
}
}
}
四，AIO（异步IO）：AIO在处理数据传输时，直接调用API的read或write方法，即为异步的，对于读操作而言，当有流可读取时，操作系统会将可读的流传入read方法的缓冲区，并通知应用程序；对于写操作而言，当操作系统将write方法传递的流写入完毕时，操作系统主动通知应用程序。 即可以理解为，read/write方法都是异步的，完成后会主动调用回调函数。 在JDK1.7中，这部分内容被称作NIO.2，主要在Java.nio.channels包下增加了下面四个异步通道：其中的read/write方法，会返回一个带回调函数的对象，当执行完读取/写入操作后，直接调用回调函数。下边来看看“查询服务器时间”例子。
public class TimeServer {
public static void main(String[] args) throws IOException {
int port = 8090;
if (args != null && args.length > 0) {
try {
port = Integer.valueOf(args[0]);
} catch (NumberFormatException e) {
}
}
AsyncTimeServerHandler asyncTimeServerHandler =new AsyncTimeServerHandler(port);
new Thread(asyncTimeServerHandler,"AIO-asyncTimeServerHandler-001").start();
}
}
public class AsyncTimeServerHandler implements Runnable {
CountDownLatch countDownLatch;
AsynchronousServerSocketChannel asynchronousServerSocketChannel;
private int port;
/**
* init server handler class
* @param port
*/
public AsyncTimeServerHandler(int port) {
this.port = port;
try {
asynchronousServerSocketChannel = AsynchronousServerSocketChannel.open();
asynchronousServerSocketChannel.bind(new InetSocketAddress(port));
System.out.println("Time server is start in port :" + port);
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void run() {
countDownLatch = new CountDownLatch(1);
doAccept();
try {
countDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
/**
* Accepts a connection.
*/
public void doAccept() {
asynchronousServerSocketChannel.accept(this, new AcceptCompletionHandler());
}
}
public class AcceptCompletionHandler implements CompletionHandler<AsynchronousSocketChannel,AsyncTimeServerHandler> {
/**
* success handler
* @param result
* @param attachment
*/
@Override
public void completed(AsynchronousSocketChannel result, AsyncTimeServerHandler attachment) {
attachment.asynchronousServerSocketChannel.accept(attachment,this);
ByteBuffer buffer =ByteBuffer.allocate(1024);
//注意回调
result.read(buffer,buffer,new ReadComplateHandler(result));
}
@Override
public void failed(Throwable exc, AsyncTimeServerHandler attachment) {
exc.printStackTrace();
attachment.countDownLatch.countDown();
}
}
public class ReadComplateHandler implements CompletionHandler<Integer, ByteBuffer> {
private AsynchronousSocketChannel socketChannel;
public ReadComplateHandler(AsynchronousSocketChannel socketChannel) {
if (socketChannel != null) {
this.socketChannel = socketChannel;
}
}
@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 recieve order : " + req);
String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(req) ? new Date(System.currentTimeMillis()).toString() : "bad order";
System.out.println("currentTime:" + currentTime);
doWrite(currentTime);
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
/**
* write data info to response
* @param currentTime
*/
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();
//注意回调
socketChannel.write(writeBuffer, writeBuffer, new CompletionHandler<Integer, ByteBuffer>() {
@Override
public void completed(Integer result, ByteBuffer attachment) {
//如果没有发送完，则继续发送
if (attachment.hasRemaining()) {
socketChannel.write(attachment, attachment, this);
}
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
try {
socketChannel.close();
} catch (IOException e) {
e.printStackTrace();
}
}
});
}
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
try {
this.socketChannel.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
public class TimeClient {
public static void main(String[] args) {
int port = 8090;
if(args !=null && args.length >0){
try {
port = Integer.valueOf(args[0]);
}catch (NumberFormatException e){
}
}
new Thread(new AsyncTimeClientHandler("127.0.0.1",port),"AsyncTimeClientHandler-001").start();
}
}
public class AsyncTimeClientHandler implements CompletionHandler<Void, AsyncTimeClientHandler>, Runnable {
private String host;
private int port;
private AsynchronousSocketChannel client;
private CountDownLatch latch;
/**
* init client handler
* @param host
* @param port
*/
public AsyncTimeClientHandler(String host, int port) {
this.host = host == null ? "127.0.0.1" : host;
this.port = port;
try {
client = AsynchronousSocketChannel.open();
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void run() {
latch = new CountDownLatch(1);
//回调自己，到completed
client.connect(new InetSocketAddress(host, port), this, this);
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
try {
client.close();
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void completed(Void result, AsyncTimeClientHandler attachment) {
byte[] req = "QUERY TIME ORDER".getBytes();
ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);
writeBuffer.put(req);
writeBuffer.flip();
//write回调到new CompletaionHandler
client.write(writeBuffer, writeBuffer, new CompletionHandler<Integer, ByteBuffer>() {
@Override
public void completed(Integer result, ByteBuffer attachment) {
if (attachment.hasRemaining()) {
client.write(attachment, attachment, this);
} else {
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
//read回调到new CompletionHandler
client.read(readBuffer, readBuffer, new CompletionHandler<Integer, ByteBuffer>() {
@Override
public void completed(Integer result, ByteBuffer attachment) {
attachment.flip();
byte[] bytes = new byte[attachment.remaining()];
attachment.get(bytes);
String body;
try {
body = new String(bytes, "UTF-8");
System.out.println("now is : " + body);
latch.countDown();
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
try {
client.close();
latch.countDown();
} catch (IOException e) {
e.printStackTrace();
}
}
});
}
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
try {
client.close();
latch.countDown();
} catch (IOException e) {
e.printStackTrace();
}
}
});
}
@Override
public void failed(Throwable exc, AsyncTimeClientHandler attachment) {
exc.printStackTrace();
try {
client.close();
latch.countDown();
} catch (IOException e) {
e.printStackTrace();
}
}
}
好，下边来比较一下这几种方式异同：
|
|
BIO |
伪异步IO |
NIO |
AIO |
|
客户端数：IO线程数 |
1：1 |
M:N（其中M可以大于N） |
M:1（1个线程处理多个客户端连接） |
M:0（不需要启动额外的线程，被动回调） |
|
IO类型（阻塞） |
阻塞IO |
阻塞IO |
非阻塞IO |
非阻塞IO |
|
IO类型（同步） |
同步IO |
同步IO |
同步IO（IO多路复用） |
异步IO |
|
可靠性 |
非常差 |
差 |
高 |
高 |
|
吞吐量 |
低 |
中 |
高 |
高 |
|
难度 |
简单 |
简单 |
复杂 |
复杂 |
好，网络编程的东西很多，而且比较复杂，需要不断的学习，不断的理解，当然也要善于站在巨人的肩膀上，例如各种中间件都已经实现好了，Netty也是一个非常棒的NIO框架，可以学习并为己用之。