【3月15日】BIO、伪异步IO以及NIO编程实践

1.引言

从java的I/O体系发展的历史看，先有java.io，后有java.nio。前者一般称之为IO,后者称之为NIO（New IO）。但是又由于其特性前者又成为BIO(Block IO),后者对应为NIO(No-block IO)。

2. IO,NIO,NIO2.0

2.1 IO

java的IO通过java.io包下的接口和类来支持。在jav.io包下，主要包括输入、输出两种IO流，每种输入、输出流又可以分为字符流和字节流两大类。其中字节流以字节为单位来处理输入、输出的操作；字符流以则以字符作为基本的操作单位。除此之外，java的IO流使用了一种装饰器设计模式。它将java的IO流分成底层的节点流和上层的处理流–其中节点流和底层的底层的物理储存节点直接关联–不同的物流节点获取节点流的方式可能存在差异；但是通过把不然的节点流包装成统一的处理流，从而让程序统一处理输入、输出成为了可能。

2.1.1 输入流和输出流

按照流向来分，可以分为输入流（以InputStream和Reader作为基类）和输出流（以OutStream和Writer为基类）：

输入流：只能从中读取数据，而不能向其写入数据。
输出流：只能从中写入数据，而不能向其读入数据。

InputStream和Reader、OutStream和Writer都是抽象基类，无法直接创建实例。

2.1.2 字节流和字符流

字节流和字符流的用法几乎完全一样，区别在于两者操作的数据单位不同–字节流操作的数据但是是8位的字节，而字符流操作的是16为的字符。

2.1.3 节点流和处理流

按照流的角色来分，可以分为节点流和处理流。

节点流是向/从一个特定的IO设备（如磁盘、网络）读写数据的流，这种流也被成为低端流。
处理流对于一个已存在的流进行连接和封装，通过封装后的流来实现数据的读写功能，也成为高端流。
使用处理流进行输入/输出时，程序并不会直接连接到实际的数据源，并没有和实际的输入/输出节点连接。使用节点流的一个明显的好处是，只要使用相同的处理流，程序就可以采用完全相同的输入/输出代码来访问不同的数据源。
识别处理流也很简单，只要流的构造器参数不是一个物流节点，而是已经存在的流，那么这种流就一定是处理流；所有的字节流都是以物理IO节点作为构造器参数的。

2.2 NIO

在IO的输入/输出流中，都是阻塞式的。以BufferedReader为例，当BufferedReader读取输入流中的数据时，如果没有读到有效的数据，程序将在此处阻塞该线程的执行。不仅如此，传统的输入/输出流都是通过字节的移动来处理的（及时不是直接的去处理字节流，底层实现还是依赖于字节流的处理），也就是说，面向流的输入/输出系统一次只能处理一个字节，因此效率不高。

从JDK1.4开始，Java提供了一系列改进的输入/输出处理的新功能，这些功能被统称为NIO，位于java.nio的包及子包下。
NIO采用内存映射文件的方式来处理输入/输出，NIO将文件或文件的一段趋于映射到内存中，这样就可以像访问内存一样来访问文件了（这种方式模拟了操作系统上虚拟内存的概念），提高了效率。
NIO的包结构如下：

java.nio:主要包含各种与Buffer相关的类；
java.nio.channels:主要包含Channel和Selector相关的类；
java.nio,charset:主要是与字符集相关的类；
java.nio.channels.spi:主要包含Channle相关服务提供者的编程接口；
java.nio.channels.sap:主要包含字符集相关服务提供者的编程接口；

Channel和Buffer是NIO中的两个核心概念，Channel是对传统输入/输出系统的模拟，在NIO中所有数据都需要通过Channel进行传输；Channel与传统的Stream（流）的区别在于它提供了一个map()方法，用以将数据映射到内存中。因此也说IO是面向“流”的处理，而NIO是面向“块”处理

Buffer可以理解为是一个容器，本质上是一个数组，Channel中的读写对象都需要首先放在Buffer中。

2.3 NIO2.0

java 7 对原有的NIO进行了重大的改进，主要包括以下两个方面：
- 提供了全面的文件IO和文件系统访问支持
- 基于异步Channel的IO

3 IO和NIO的主要区别

IO	NIO
面向流	面向缓冲（块）
阻塞	非阻塞
无	选择器

3.1 面向流与面向缓冲

Java IO和NIO之间第一个最大的区别是，IO是面向流的，NIO是面向缓冲区的。 Java IO面向流意味着每次从流中读一个或多个字节，直至读取所有字节，它们没有被缓存在任何地方。此外，它不能前后移动流中的数据。如果需要前后移动从流中读取的数据，需要先将它缓存到一个缓冲区。 Java NIO的缓冲导向方法略有不同。数据读取到一个它稍后处理的缓冲区，需要时可在缓冲区中前后移动。这就增加了处理过程中的灵活性。但是，还需要检查是否该缓冲区中包含所有您需要处理的数据。而且，需确保当更多的数据读入缓冲区时，不要覆盖缓冲区里尚未处理的数据。

3.2 阻塞与非阻塞IO

Java IO的各种流是阻塞的。这意味着，当一个线程调用read() 或 write()时，该线程被阻塞，直到有一些数据被读取，或数据完全写入。该线程在此期间不能再干任何事情了。Java NIO的非阻塞模式，使一个线程从某通道发送请求读取数据，但是它仅能得到目前可用的数据，如果目前没有数据可用时，就什么都不会获取，而不是保持线程阻塞，所以直至数据变的可以读取之前，该线程可以继续做其他的事情。非阻塞写也是如此。一个线程请求写入一些数据到某通道，但不需要等待它完全写入，这个线程同时可以去做别的事情。线程通常将非阻塞IO的空闲时间用于在其它通道上执行IO操作，所以一个单独的线程现在可以管理多个输入和输出通道（channel）

选择器（Selectors）

Java NIO的选择器允许一个单独的线程来监视多个输入通道，你可以注册多个通道使用一个选择器，然后使用一个单独的线程来“选择”通道：这些通道里已经有可以处理的输入，或者选择已准备写入的通道。这种选择机制，使得一个单独的线程很容易来管理多个通道。

4 BIO、伪异步IO以及NIO编程实践

本章实例主要来自于李林峰先生的《Netty权威指南》一书中。源码放在上我的github上。

4.1 传统的BIO编程

网络编程的基本模型是C/S模型，也就是两个进程之间进行的通信，其中服务端提供位置信息（绑定的IP和监听的接口），客户端通过连接操作向服务端监听的地址发起连接请求，通过三次握手建立连接；如果连接成功，双方就可以通过Socket进行通信。

以经典的TimeServer为例，通过代码分析各种IO模式下的编程。

4.1.2 同步阻塞式IO创建的TimeServer

package com.njust.bio;import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;public class TimeServer {public static void main(String[] args) throws IOException {// 默认值为8080int port = 8080;// 如果有参数，port设定读取参数的值if(args!=null && args.length>0){try{port = Integer.valueOf(args[0]);}catch (NumberFormatException e){// 采用默认值}}ServerSocket  server = null;try{// 参数port指定服务器要绑定的端口（服务器要监听的端口）server = new ServerSocket(port);System.out.println("the time server is start in port : " + port);Socket socket = null;// 通过一个无限循环来监听客户端的连接while (true){//当服务器执行ServerSocket的accept()方法时，如果连接请求队列为空，// 服务器就会一直等待，直到接收到了客户连接才从accept()方法返回。socket = server.accept();// TimeServerHandler是一个Runablenew Thread(new TimeServerHandler(socket)).start();}}finally {if(server!=null){System.out.println("the time server is close");server.close();server = null;}}}}

TimeServer根据传入的参数设置监听的端口如果没有入参，则使用默认的8080。通过 new ServerSocket(port) 来创建 ServerSocket，如果端口没有被占用，服务器监听port成功。并通过while的无限循环来监听客户端的连接，如果没有客户端接入，则主线程会阻塞在ServerSocket的accept上。

4.1.2 同步阻塞式IO创建的TimeServerHandler

package com.njust.bio;import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;public class TimeServerHandler implements Runnable {private Socket socket;public TimeServerHandler(Socket socket) {this.socket = socket;}@Overridepublic 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 java.util.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 e2){e2.printStackTrace();}}}}
}

可以看出，TimeServerHandler实现了Runable接口。

4.1.2 同步阻塞式IO创建的TimeClient

package com.njust.bio;import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;public class TimeClient {public static void main(String[] args) {int port = 8080;if(args!=null && args.length>0){try{port = Integer.valueOf(args[0]);}catch (NumberFormatException e){// 不进行处理}}BufferedReader in = null;PrintWriter out = null;Socket socket = null;try{socket = new Socket("127.0.0.1", port);// 输入流 服务端提供in = new BufferedReader(new InputStreamReader(socket.getInputStream()));//  通过现有的 OutputStream 创建新的 PrintWriter:定义输出流的位置// 输出流 输出给服务端out = new PrintWriter(socket.getOutputStream(), true);String currentTime = null;out.println("QUERY TIME ORDER");System.out.println("Send order 2 server succeed");String resp = in.readLine();System.out.println("Now is : " + resp);}catch (Exception e){if(in != null){try {in.close();}catch (IOException e1){e1.printStackTrace();}}if(out != null){out.close();out = null;}if(socket != null){try{socket.close();}catch (IOException e2){e2.printStackTrace();}}}}
}

4.2 伪异步的IO编程

采用线程池和任务队列可以实现一种叫做伪异步的IO通信框架，具体实现如下.

4.2.1 伪异步的IO创建的TimeServer

package com.njust.io2;import com.njust.bio.TimeServerHandler;import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;public class TimeServer {public static void main(String[] args) throws IOException {// 默认值为8080int port = 8080;// 如果有参数，port设定读取参数的值if(args!=null && args.length>0){try{port = Integer.valueOf(args[0]);}catch (NumberFormatException e){// 采用默认值}}ServerSocket  server = null;try{// 参数port指定服务器要绑定的端口（服务器要监听的端口）server = new ServerSocket(port);System.out.println("the time server is start in port : " + port);Socket socket = null;// 创建I/O任务线程池TimeServerHandlerExecutorPool singleExecutor = new TimeServerHandlerExecutorPool(50, 1000);// 通过一个无限循环来监听客户端的连接while (true){//当服务器执行ServerSocket的accept()方法时，如果连接请求队列为空，// 服务器就会一直等待，直到接收到了客户连接才从accept()方法返回。socket = server.accept();// TimeServerHandler是一个Runable//new Thread(new TimeServerHandlerExecutorPool(socket)).start();// 将socket封装成一个tasksingleExecutor.execute(new TimeServerHandler(socket));}}finally {if(server!=null){System.out.println("the time server is close");server.close();server = null;}}}}

4.2.2 4.2.1 伪异步的IO创建的TimeServerHanderExecutorPool

package com.njust.io2;import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;public class TimeServerHandlerExecutorPool  {private ExecutorService executorService;public TimeServerHandlerExecutorPool(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);}
}

4.3 NIO编程

4.3.1 NIO创建的TimeServer

package com.njust.nio;import java.io.IOException;public class NioTimeServer {public static void main(String[] args) throws IOException {// 默认值为8080int port = 8080;// 如果有参数，port设定读取参数的值if (args != null && args.length > 0) {try {port = Integer.valueOf(args[0]);} catch (NumberFormatException e) {// 采用默认值}}MutiplexerTimeServer timeServer = new MutiplexerTimeServer(port);new Thread(timeServer, "NIO-MutiplexerTimeServer-001").start();}
}

4.3.2 NIO创建的 MutiplexerTimeServer

package com.njust.nio;import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
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.util.Iterator;
import java.util.Set;public class MutiplexerTimeServer implements Runnable {private Selector selector;private ServerSocketChannel serverSocketChannel;private boolean stop;/*构造方法，在构造方法中进行资源的初始化，创建多路复用器Selector,ServerSocketChannel;* 对Channel的TCP参数进行配置，例如：* 1.serverSocketChannel.configureBlocking(false)设置为异步非阻塞* 2.serverSocketChannel.bind(new InetSocketAddress(port),1024)backlog设置为1024* 3.serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT)，将serverSocketChannel注册到selector，* 并监控SelectionKey.OP_ACCEPT的操作位** 如果初始化失败（例如端口被占用），则退出*/// 初始化多路复用器，并绑定监听器public MutiplexerTimeServer(int port) {try{selector = Selector.open();serverSocketChannel = ServerSocketChannel.open();// 配置为非阻塞serverSocketChannel.configureBlocking(false);//绑定serverSocketChannel.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/*通过while循环体循环遍历selector，休眠时间设置为1000，即1s。* 无论是否有读写等事件发生，selector每隔1s就被唤醒一次*/public void run() {while (!stop){try{// 选择一组键，其相应的通道已为 I/O 操作准备就绪selector.select(1000);//  返回此选择器的已选择键集。Set<SelectionKey> selectionKeys = selector.selectedKeys();Iterator<SelectionKey> it = selectionKeys.iterator();SelectionKey key = null;while(it.hasNext()){key = it.next();it.remove();try{handleInput(key);}catch (Exception e){if(key!=null){key.cancel();if(key.channel()!=null){key.channel().close();}}}}}catch (Throwable t){t.printStackTrace();}}//多路复用器关闭后，所有注册在上面的Channel和Pipe等资源都会自动的去注册并关闭，所以不需要重复释放资源if(selector!=null){try{selector.close();}catch (IOException e){e.printStackTrace();}}}/*处理新介入的客户端请求信息，通过SelectionKey key即可获知网络事件的类型*/private void handleInput(SelectionKey key) throws IOException{if(key.isValid()){//处理新接入的请求信息if(key.isAcceptable()) {// 接受新连接ServerSocketChannel ssc = (ServerSocketChannel) key.channel();/* 接受到此通道套接字的连接。如果此通道处于非阻塞模式，那么在不存在挂起的连接时，此方法将直接返回 null。否则，在新的连接可用或者发生 I/O 错误之前会无限期地阻塞它。不管此通道的阻塞模式如何，此方法返回的套接字通道（如果有）将处于阻塞模式。此方法执行的安全检查与 ServerSocket 类的 accept 方法执行的安全检查完全相同。也就是说，如果已安装了安全管理器，则对于每个新的连接，此方法都会验证安全管理器的checkAccept 方法是否允许使用该连接的远程端点的地址和端口号。 */SocketChannel sc = ssc.accept();//到这里，相当于完成了TCP的三次握手，TCP的物理链路层正式建立// 设置为异步非阻塞sc.configureBlocking(false);// 将新连接注册到selector中，并监控sc.register(selector, SelectionKey.OP_READ);}// 读取客户端的请求信息if(key.isReadable()){//读取数据SocketChannel sc = (SocketChannel)key.channel();ByteBuffer readBuffer = ByteBuffer.allocate(1024);// 将字节序列从此通道中读入给定的缓冲区。// 返回：读取的字节数，可能为零，如果该通道已到达流的末尾，则返回 -1int readBytes = sc.read(readBuffer);if(readBytes>0){/*0 <= 标记 <= 位置 <= 限制 <= 容量*  (Buffer)缓冲区是特定基本类型元素的线性有限序列。除内容外，缓冲区的基本属性还包括容量、限制和位置：*  缓冲区的容量 是它所包含的元素的数量。缓冲区的容量不能为负并且不能更改。*  缓冲区的限制 是第一个不应该读取或写入的元素的索引。缓冲区的限制不能为负，并且不能大于其容量。*  缓冲区的位置 是下一个要读取或写入的元素的索引。缓冲区的位置不能为负，并且不能大于其限制。*  对于每个非 boolean 基本类型，此类都有一个子类与之对应。 */// 反转此缓冲区。首先将限制(limit)设置为当前位置(position)，然后将位置(position)设置为 0。// 如果已定义了标记，则丢弃该标记。readBuffer.flip(); //也就是说调用flip之后，读写指针指到缓存头部，并且设置了最多只能读出之前写入的数据长度(而不是整个缓存的容量大小)。byte[] bytes = new byte[readBuffer.remaining()];//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 java.util.Date(System.currentTimeMillis()).toString():"BAD ORDER";doWrite(sc, currentTime);}else if (readBytes<0){// 关闭链路key.cancel();sc.close();}else ; //读到0字节，忽略}}}private  void doWrite(SocketChannel channel, String response) throws IOException{if(response!=null && response.trim().length()>0){byte[] bytes = response.getBytes();ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);writeBuffer.put(bytes);//此方法将给定的源 bytes 数组的所有内容传输到此缓冲区中。writeBuffer.flip();channel.write(writeBuffer);}}
}