Java 学习系列： IO相关的BIO,NIO与AIO

介绍： 操作系统的IO指其内部与外界的交互(Input/Output)，网络IO通常是使用某种网络协议(TCP/IP,UDP/IP,etc)监听计算机的端口，进而实现数据交互。

进行IO操作，通常包括1)发起IO请求，询问要处理的数据是否准备好了(比如与某个服务端口建立socket连接) 2)执行真正的IO操作(比如通过socket传输数据)

阻塞/非阻塞：它是根据上面步骤1的处理状态来进行区分，当发起IO请求时，如果不能立即执行读写操作就会阻塞一直等待，而非阻塞的话会立即返回。

同步/异步：首先同步和异步是针对应用程序和内核的交互而言的，其次它是根据上面步骤2的处理状态来进行区分，同步的话，用户操作进程会等待操作系统IO操作执行完成返回; 异步的话用户进程触发IO操作后无须等待立即返回去做自己的事情，而操作系统完成指定的IO操作任务后会进行回调通知。

Java 在IO编程层面上有三种方式，分别是BIO，NIO，和AIO

1 . BIO概念说明 ：阻塞型，面向流操作，它意味着控制权只到调用操作结束了才会回到调用者手里，在此之前会一直被阻塞。服务器实现模式为一个连接一个线程，即客户端有连接请求时服务器端就需要启动一个对应的线程进行处理，jdk1.4之前的唯一IO操作方法。

示例代码：

/* 服务端 */
package test.demo.io;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.ServerSocket;
import java.net.Socket;

import javax.net.ServerSocketFactory;

public class BioServer {
private Socket socket = null;
private ServerSocket serverSocket = null;

public BioServer(int port) throws IOException {
serverSocket = ServerSocketFactory.getDefault().createServerSocket(port);
int threadNum = 0;
while (true) {
try{
//监听直到接受连接后返回一个新Socket对象
socket = serverSocket.accept();//阻塞
threadNum++;
//new一个线程处理连接请求
new Thread(new Worker(socket, threadNum)).start();
}
catch (Throwable e) {  //防止发生异常搞死服务器
e.printStackTrace();
}
}
}

private static class Worker implements Runnable{
private int threadNum;
private Socket socket;

public Worker(Socket socket, int threadNum){
this.socket = socket;
this.threadNum = threadNum;
System.out.println("Start thread: " + threadNum);
}

@Override
public void run() {
BufferedReader reader = null;
PrintWriter writer = null;

try {
reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
writer = new PrintWriter(socket.getOutputStream());
String inputStr = reader.readLine();//没有内容会阻塞
while(inputStr!= null && !inputStr.equals("exit")){
System.out.println("thread->" + threadNum + ": " + inputStr);
inputStr = reader.readLine();
}

writer.flush();

if(writer != null){
writer.close();
}

if(reader != null){
reader.close();
}
} catch (IOException e) {
e.printStackTrace();
}
}

}

public static void main(String[] args) {
try {
new BioServer(7777);
} catch (IOException e) {
e.printStackTrace();
}
}
}

/* 客户端 */
package test.demo.io;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;

public class BioClient {

private static class Worker implements Runnable {

private String address = "localhost";
private int port = 7777;

public Worker() {
super();
}

@Override
public void run() {
Socket socket = null;
BufferedReader reader = null;
PrintWriter writer = null;

try {
// 创建一个Socket并连接到指定的目标服务器
socket = new Socket(address, port);
reader = new BufferedReader(new InputStreamReader(System.in));
writer = new PrintWriter(socket.getOutputStream());

String message = reader.readLine(); // 没有内容会阻塞
do{
writer.println(message);
message = reader.readLine();
}while(!message.equals("exit"));

writer.flush();

} catch (IOException e) {
e.printStackTrace();
} finally {
try {
if (writer != null) {
writer.close();
}

if (reader != null) {
reader.close();
}

if (socket != null) {
socket.close();
}
} catch (IOException e) {
e.printStackTrace();
}
}
}

}

public static void main(String[] args) {
for(int i=0; i<5; i++){
new Thread(new Worker()).start();
}
}
}

2 . NIO概念说明 ：非阻塞同步，jdk1.4添加，面向缓冲区操作，它会立即返回调用者无需等待。通过Selectors实现的轮询模式，即客户端发送的连接请求都会注册到多路复用器上，多路复用器轮询到连接有I/O请求时才启动一个线程进行处理。

示例代码：

/* 服务端 */
package test.demo.io;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.SelectorProvider;
import java.util.Iterator;
import java.util.Set;

public class NioServer {

ServerSocketChannel channel = null;

public NioServer(int port) throws IOException {
channel = ServerSocketChannel.open();
// 绑定监听端口
channel.socket().bind(new InetSocketAddress(port));
// 设置为非阻塞形式
channel.configureBlocking(false);
}

@SuppressWarnings("rawtypes")
public void waitForConnection() throws IOException {
Selector acceptSelector = SelectorProvider.provider().openSelector();
channel.register(acceptSelector, SelectionKey.OP_ACCEPT);
while (acceptSelector.select() > 0) {
// 某客户已经准备好可以进行I/O操作了，获取其ready键集合
Set readKeys = acceptSelector.selectedKeys();
Iterator iter = readKeys.iterator();
while (iter.hasNext()) {
SelectionKey sk = (SelectionKey) iter.next();
iter.remove();
if (sk.isAcceptable()) {
dealConnection(sk);
}
}
}
}

/**
* @throws IOException
*
*/
private void dealConnection(SelectionKey sk) throws IOException{
ServerSocketChannel server = (ServerSocketChannel) sk.channel();
SocketChannel inChannel = server.accept();
ByteBuffer buffer = ByteBuffer.allocate(128);
buffer.clear();
int bytesRead = inChannel.read(buffer);
while (bytesRead != -1) {
System.out.println("Read " + bytesRead);
buffer.flip();
while(buffer.hasRemaining()){
System.out.print((char) buffer.get());
}
buffer.clear();
bytesRead = inChannel.read(buffer);
}
}

public static void main(String[] args) {
try {
NioServer server = new NioServer(7777);
server.waitForConnection();
} catch (IOException e) {
e.printStackTrace();
}

}
}

/* 客户端 */
package test.demo.io;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;

public class NioClient {

private String address = "localhost";
private int port = 7777;

public NioClient() {
super();
}

public void work() {
SocketChannel channel = null;
try {
InetSocketAddress socketAddress = new InetSocketAddress(address, port);
channel = SocketChannel.open(socketAddress);
channel.configureBlocking(false);
BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
String message = reader.readLine();
ByteBuffer buffer = ByteBuffer.allocate(64);
buffer.clear();
buffer.put(message.getBytes());
channel.write(buffer);
} catch (IOException e) {
e.printStackTrace();

} finally {
if (channel != null) {
try {
channel.close();
} catch (IOException e) {
System.err.println(e.toString());
}
}
}
}

public static void main(String[] args) {
for(int i=0; i<1; i++){
new NioClient().work();
}
}
}

3 . AIO概念说明 ：异步非阻塞，服务器实现模式为一个有效请求一个线程，客户端的I/O请求都是由OS先完成了再通知服务器应用去启动线程进行处理

示例代码：

/* 服务端 */
package test.demo.io;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class AioServer {

private final Integer port;

public AioServer(int port) {
this.port = port;
}

public void listen() {
try {
ExecutorService executorService = Executors.newCachedThreadPool();
AsynchronousChannelGroup threadGroup = AsynchronousChannelGroup.withCachedThreadPool(executorService, 1);

try (AsynchronousServerSocketChannel server = AsynchronousServerSocketChannel.open(threadGroup)) {
server.bind(new InetSocketAddress(port));

System.out.println("Echo listen on:" + port);
server.accept(null,
new CompletionHandler<AsynchronousSocketChannel, Object>() {
// 这里构造buffer是，用allocateDirect方法，下面打印客户端来的数据时会出错，why?
final ByteBuffer echoBuffer = ByteBuffer.allocate(1024);

public void completed(AsynchronousSocketChannel result, Object attachment) {
System.out.println("Client data is coming ....");
try {
echoBuffer.clear();
result.read(echoBuffer).get();
// 打印从客户端接收到的数据
System.out.println("Data is: " + new String(echoBuffer.array()));
//清空buffer里面的缓存
echoBuffer.flip();
// 发送数据到客户端
result.write(ByteBuffer.wrap("Welcome Hinsteny!".getBytes()));
echoBuffer.flip();
} catch (InterruptedException | ExecutionException e) {
System.out.println(e.toString());
} finally {
try {
result.close();
server.accept(null, this);
} catch (Exception e) {
System.out.println(e.toString());
}
}
}

@Override
public void failed(Throwable exc, Object attachment) {
System.out.println("server failed: " + exc);
}
});

try {
// Wait for ever
Thread.sleep(Integer.MAX_VALUE);
} catch (InterruptedException ex) {
System.out.println(ex);
}
}
} catch (IOException e) {
System.out.println(e);
}
}

public static void main(String args[]) {
new AioServer(8000).listen();
}

}

/* 客户端 */
package test.demo.io;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;

public class AioClient {

private String address = "127.0.0.1";
private int port = 8000;

private final AsynchronousSocketChannel client;

public AioClient() throws Exception {
client = AsynchronousSocketChannel.open();
}

public void start() throws Exception {
client.connect(new InetSocketAddress(address, port), null,
// 建立连接后的回调
new CompletionHandler<Void, Object>() {
@Override
public void completed(Void result, Object attachment) {
try {
System.out.println("Connect to channel successed and start send data to server...");
client.write(ByteBuffer.wrap("Hinsteny is come here ..".getBytes())).get();
} catch (Exception ex) {
ex.printStackTrace();
}
}

@Override
public void failed(Throwable exc, Object attachment) {
exc.printStackTrace();
}
});

final ByteBuffer bb = ByteBuffer.allocate(1024);
client.read(bb, null,
// 读操作结束后的回调
new CompletionHandler<Integer, Object>() {
@Override
public void completed(Integer result, Object attachment) {
System.out.println("Message size is:" + result);
System.out.println("Message content is:" + new String(bb.array()));
}

@Override
public void failed(Throwable exc, Object attachment) {
exc.printStackTrace();
}
});

try {
// Wait for ever
Thread.sleep(Integer.MAX_VALUE);
} catch (InterruptedException ex) {
System.out.println(ex);
}

}

public static void main(String args[]) throws Exception {
new AioClient().start();
}
}