BIO-NIO-AIO

一、同步和异步

同步和异步是针对应用程序和操作系统的内核交互而言的，同步指的是用户进程触发IO操作并等待或者轮询地区查看IO操作是否就绪，而异步是值用户触发IO操作以后便开始做自己地事情，而当IO操作已经完成地时候会得到IO完成地通知。

以银行取款为例：
同步：自己亲自持银行卡到银行取钱(使用同步IO时，JAVA自己处理IO读写)
异步：委托小弟到银行取钱，我干其他事，然后给我（使用异步IO时，JAVA将IO读写委托给OS处理，需要将数据缓冲区地址和大小传给OS(银行卡和密码)，需要支持异步IO操作API）

二、阻塞和非阻塞

阻塞和非阻塞是针对于进程在访间数据的时候，根据IO操作的就绪状态来采取的不同方式，说白了是一种读取或者写入操作方法的实现方式，阻塞方式下读取或者写入方法将一直等待，而非阻塞方式下，读取或者写入方法会立即返回一个状态值。

以银行取款为例：
阻塞：ATM排队取款，你只能等待（使用阻塞旧时，JAVA调用会一直阻塞到读写完成才返回)
非阻塞：柜台取款，取个号，然后坐在椅子上做其它事，等号广播会通知你办理，没到号你就不能去，你可以不断问大堂经理排到了没有，大堂经理如果说还没到你就不能去（使用非阻塞IO时，如果不能读写JAVA调用会马上返回，当IO事件分发器通知可读写时再继续进行读写，不断循环直到读写完成）

各IO比较：

三、BIO

网络编程的基本模型是C/S模型，即两个进程间的通信。服务端提供IP和监听端口，客户端通过连接操作想服务端监听的地址发起连接请求，通过三次握手连接，如果连接成功建立，双方就可以通过套接字进行通信。传统的同步阻塞模型开发中，ServerSocket负责绑定IP地址，启动监听端口；Socket负责发起连接操作。连接成功后，双方通过输入和输出流进行同步阻塞式通信。

传统的 IO 大致可以分为4种类型：

1.InputStream、OutputStream 基于字节操作的 IO
2.Writer、Reader 基于字符操作的 IO
3.File 基于磁盘操作的 IO
4.Socket 基于网络操作的 IO

本文例子：客户端发送一段算式的字符串到服务器，服务器计算后返回结果到客户端。BIO 就是传统的 java.io包，它是基于流模型实现的，交互的方式是同步、阻塞方式，也就是说在读入输入流或者输出流时，在读写动作完成之前，线程会一直阻塞在那里，它们之间的调用时可靠的线性顺序。它的有点就是代码比较简单、直观；缺点就是IO 的效率和扩展性很低，容易成为应用性能瓶颈。该模型最大的问题就是缺乏弹性伸缩能力，当客户端并发访问量增加后，服务端的线程个数和客户端并发访问数呈1:1的正比关系，Java中的线程也是比较宝贵的系统资源，线程数量快速膨胀后，系统的性能将急剧下降，随着访问量的继续增大，系统最终就死掉了。
传统的BIO模型（同步阻塞I/O）：

BIO代码思路：

Socket 网络通信过程简单来说分为下面 4 步：
1.建立服务端并且监听客户端请求
2.客户端请求，服务端和客户端建立连接
3.两端之间可以传递数据
4.关闭资源

服务器端：

1.创建 ServerSocket 对象并且绑定地址（ip）和端口号(port)：server.bind(new InetSocketAddress(host, port))
2.通过 accept()方法监听客户端请求
3.连接建立后，通过输入流读取客户端发送的请求信息
4.通过输出流向客户端发送响应信息
5.关闭相关资源

客户端：

1.创建Socket 对象并且连接指定的服务器的地址（ip）和端口号(port)：socket.connect(inetSocketAddress)
2.连接建立后，通过输出流向服务器端发送请求信息
3.通过输入流获取服务器响应的信息
4.关闭相关资源

同步阻塞式I/O创建的Server源码：

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
/*** BIO服务端源码*/
final class ServerNormal {//默认的端口号private static int DEFAULT_PORT = 12345;//单例的ServerSocketprivate static ServerSocket server;//根据传入参数设置监听端口，如果没有参数调用以下方法并使用默认值public static void start() throws IOException{//使用默认值start(DEFAULT_PORT);}//这个方法不会被大量并发访问，不太需要考虑效率，直接进行方法同步就可以public synchronized static void start(int port) throws IOException{if(server != null) {return;}try{//通过构造函数创建ServerSocket//如果端口合法且空闲，服务端就监听成功server = new ServerSocket(port);System.out.println("服务器已启动，端口号：" + port);//通过无线循环监听客户端连接//如果没有客户端接入，将阻塞在accept操作上。while(true){Socket socket = server.accept();//当有新的客户端接入时，会执行下面的代码//然后创建一个新的线程处理这条Socket链路new Thread(new ServerHandler(socket)).start();}}finally{//一些必要的清理工作if(server != null){System.out.println("服务器已关闭。");server.close();server = null;}}}
}

客户端消息处理线程ServerHandler源码：

import com.company.Calculator;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;
/*** 客户端线程* 用于处理一个客户端的Socket链路*/
public class ServerHandler implements Runnable{private Socket socket;public ServerHandler(Socket socket) {this.socket = socket;}@Overridepublic void run() {BufferedReader in = null;PrintWriter out = null;try{in = new BufferedReader(new InputStreamReader(socket.getInputStream()));out = new PrintWriter(socket.getOutputStream(),true);String expression;String result;while(true){//通过BufferedReader读取一行//如果已经读到输入流尾部，返回null,退出循环//如果得到非空值，就尝试计算结果并返回if((expression = in.readLine())==null) {break;}System.out.println("服务器收到消息：" + expression);try{Calculator calculator=new Calculator(expression);result = calculator.cal();}catch(Exception e){result = "计算出错：" + e.getMessage();}out.println(result);}}catch(Exception e){e.printStackTrace();}finally{//一些必要的清理工作if(in != null){try {in.close();} catch (IOException e) {e.printStackTrace();}in = null;}if(out != null){out.close();out = null;}if(socket != null){try {socket.close();} catch (IOException e) {e.printStackTrace();}socket = null;}}}
}

同步阻塞式I/O创建的Client源码：

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;
/*** 阻塞式I/O创建的客户端*/
public class Client {//默认的端口号private static int DEFAULT_SERVER_PORT = 12345;private static String DEFAULT_SERVER_IP = "127.0.0.1";public static void send(String expression){send(DEFAULT_SERVER_PORT,expression);}public static void send(int port,String expression){System.out.println("算术表达式为：" + expression);Socket socket = null;BufferedReader in = null;PrintWriter out = null;try{socket = new Socket(DEFAULT_SERVER_IP,port);in = new BufferedReader(new InputStreamReader(socket.getInputStream()));out = new PrintWriter(socket.getOutputStream(),true);out.println(expression);System.out.println("___结果为：" + in.readLine());}catch(Exception e){e.printStackTrace();}finally{//一下必要的清理工作if(in != null){try {in.close();} catch (IOException e) {e.printStackTrace();}in = null;}if(out != null){out.close();out = null;}if(socket != null){try {socket.close();} catch (IOException e) {e.printStackTrace();}socket = null;}}}
}

放到同一个程序（jvm）中运行：

package com.company.BIO;
import java.io.IOException;
import java.util.Random;
/*** 测试方法* @version 1.0*/
public class Test {//测试主方法public static void main(String[] args) throws InterruptedException {//运行服务器new Thread(new Runnable() {@Overridepublic void run() {try {ServerNormal.start();} catch (IOException e) {e.printStackTrace();}}}).start();//避免客户端先于服务器启动前执行代码Thread.sleep(100);//运行客户端char operators[] = {'+','-','*','/'};Random random = new Random(System.currentTimeMillis());new Thread(new Runnable() {@SuppressWarnings("static-access")@Overridepublic void run() {while(true){//随机产生算术表达式String expression = random.nextInt(10)+""+operators[random.nextInt(4)]+(random.nextInt(10)+1);Client.send(expression);try {Thread.sleep(random.nextInt(1000));} catch (InterruptedException e) {e.printStackTrace();}}}}).start();}
}

伪异步I/O编程
为了改进这种一连接一线程的模型，我们可以使用线程池来管理这些线程（需要了解更多请参考前面提供的文章），实现1个或多个线程处理N个客户端的模型（但是底层还是使用的同步阻塞I/O），通常被称为“伪异步I/O模型“。
伪异步I/O模型图：

只需要将新建线程的地方，交给线程池管理即可，只需要改动刚刚的Server代码

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/*** BIO服务端源码__伪异步I/O*/final class ServerNormal {//默认的端口号private static int DEFAULT_PORT = 12345;//单例的ServerSocketprivate static ServerSocket server;//线程池懒汉式的单例private static ExecutorService executorService = Executors.newFixedThreadPool(60);//根据传入参数设置监听端口，如果没有参数调用以下方法并使用默认值public static void start() throws IOException{//使用默认值start(DEFAULT_PORT);}//这个方法不会被大量并发访问，不太需要考虑效率，直接进行方法同步就行了public synchronized static void start(int port) throws IOException{if(server != null) {return;}try{//通过构造函数创建ServerSocket//如果端口合法且空闲，服务端就监听成功server = new ServerSocket(port);System.out.println("服务器已启动，端口号：" + port);//通过无线循环监听客户端连接//如果没有客户端接入，将阻塞在accept操作上。while(true){Socket socket = server.accept();//当有新的客户端接入时，会执行下面的代码//然后创建一个新的线程处理这条Socket链路executorService.execute(new ServerHandler(socket));}}finally{//一些必要的清理工作if(server != null){System.out.println("服务器已关闭。");server.close();server = null;}}}
}

测试运行结果与上面是一样的。
我们知道，如果使用CachedThreadPool线程池（不限制线程数量），其实除了能自动帮我们管理线程（复用），看起来也就像是1:1的客户端：线程数模型，而FixedThreadPool我们就有效的控制了线程的最大数量，保证了系统有限的资源的控制，实现了N:M的伪异步I/O模型。但是，正因为限制了线程数量，如果发生大量并发请求，超过最大数量的线程就只能等待，直到线程池中的有空闲的线程可以被复用。而对Socket的输入流就行读取时，会一直阻塞，直到发生：

1.有数据可读
2.可用数据以及读取完毕
3.发生空指针或I/O异常
4.所以在读取数据较慢时（比如数据量大、网络传输慢等），大量并发的情况下，其他接入的消息，只能一直等待，这就是最大的弊端。

后面介绍的NIO，就能解决这个难题。

四、NIO

NIO 是 Java 1.4 引入的 java.nio 包，提供了 Channel（管道）、Selector（多路复用器）、Buffer（缓冲区）等新的抽象，可以构建多路复用的、同步非阻塞 IO 程序，同时提供了更接近操作系统底层高性能的数据操作方式。NIO的最重要的地方是当一个连接创建后，不需要对应一个线程，这个连接会注册到多路复用器上面，所以所有的连接只需要一个线程就可以搞定，当这个线程中的多路复用器进行轮询的时候，发现连接上有请求的话，才开启一个线程进行处理，也就是一个请求一个线程模式。在NIO的处理方式中，当一个请求来的话，开启线程进行处理，可能会等待后端应用的资源（JDBC链接等），其实这个线程就被阻塞了，当并发上来的话，还是会有BIO一样的问题。NIO方式适用于链接数目多且连接比较短（轻操作）的架构，比如聊天服务器。

工作原理图：

主要参数：
1.Channel：通道，连接客户端和服务端的一个管道，通道不同于流的地方就是通道是双向的，可以用于读、写和同时读写操作。底层的操作系统的通道一般都是全双工的，所以全双工的Channel比流能更好的映射底层操作系统的API。

Channel主要分两大类：

1. SelectableChannel：用户网络读写
2. FileChannel：用于文件操作

Channel主要实现类：

1. SocketChannel：一个客户端发起TCP连接的Channel
2. ServerSocketChannel：一个服务端监听新连接的TCP Channel，对于每一个新的Client连接，都会建立一个对应的SocketChannel
3. FileChannel：从文件中读写数据

2.Selector：Selector是NIO中最为重要的组件之一，我们常常说的多路复用器就是指的Selector组件。Selector组件用于轮询一个或多个NIO Channel的状态是否处于可读、可写。通过轮询的机制就可以管理多个Channel，也就是说可以管理多个网络连接。可以想象成一个环状传送带，上面可以接入很多管道，selector还可以对每个管道设置感兴趣的颜色（连接（红色），读（黄色），写（蓝色），接收数据）。当Selector开始轮询的时候Selector这个传送带就一直转动，当某个管道被传送到感兴趣事件检查点的时候，selector会检查改管道当前颜色（即事件）之前是否被注册成了感兴趣颜色（事件），如果感兴趣，那么Selector就可以对这个管道做处理了，比如把管道传给别的线程，让别的线程完成读写操作。一个Selector可以同时轮询多个Channel，因为JDK使用了epoll()代替传统的select实现，所以没有最大连接句柄1024/2048的限制。所以，只需要一个线程负责Selector的轮询，就可以接入成千上万的客户端。

轮询机制

首先，需要将Channel注册到Selector上，这样Selector才知道需要管理哪些Channel，接着Selector会不断轮询其上注册的Channel，如果某个Channel发生了读或写的时间，这个Channel就会被Selector轮询出来，然后通过SelectionKey可以获取就绪的Channel集合，进行后续的IO操作。

属性操作

创建Selector

通过open()方法，我们可以创建一个Selector对象。

Selector selector = Selector.open();

注册Channel到Selector中

我们需要将Channel注册到Selector中，才能够被Selector管理。

channel.configureBlocking(false);
SelectionKey key = channel.register(selector, SelectionKey.OP_READ);

某个Channel要注册到Selector中，那么该Channel必须是非阻塞，所有上面代码中有个configureBlocking()的配置操作。在register(Selector selector, int interestSet)方法的第二个参数，标识一个interest集合，意思是Selector对哪些事件感兴趣，可以监听四种不同类型的事件：

Selector可以监听的事件类型（可使用 SelectionKey 的四个常量表示）：

读: SelectionKey.OP_READ （1）

写:SelectionKey.OP_WRITE（4）

连接 : SelectionKey.OP_CONNECT（8）

接收: SelectionKey.OP_ACCEPT （16）

Connect事件：连接完成事件( TCP 连接 )，仅适用于客户端，对应 SelectionKey.OP_CONNECT。

Accept事件：接受新连接事件，仅适用于服务端，对应 SelectionKey.OP_ACCEPT 。

Read事件：读事件，适用于两端，对应 SelectionKey.OP_READ ，表示 Buffer 可读。

Write事件：写时间，适用于两端，对应 SelectionKey.OP_WRITE ，表示 Buffer 可写。

Channel触发了一个事件，表明该时间已经准备就绪：

一个Client Channel成功连接到另一个服务器，成为“连接就绪”

一个ServerSocket准备好接收新进入的接，称为“接收就绪”

一个有数据可读的Channel，称为“读就绪”

一个等待写数据的Channel，称为”写就绪“

当然，Selector是可以同时对多个事件感兴趣的，我们使用或运算即可组合多个事件：

int interestSet = SelectionKey.OP_READ | SelectionKey.OP_WRITE;

Selector其他一些操作
选择Channel

//当Selector执行select()方法就会产生阻塞，
//等到注册在其上的Channel准备就绪就会立即返回，返回准备就绪的数量。
public abstract int select() throws IOException;
//select(long timeout)则是在select()的基础上增加了超时机制。
public abstract int select(long timeout) throws IOException;
//selectNow()立即返回，不产生阻塞。
public abstract int selectNow() throws IOException;

有一点非常需要注意： select 方法返回的 int 值，表示有多少 Channel 已经就绪。
自上次调用select 方法后有多少 Channel 变成就绪状态。如果调用 select 方法，因为有一个 Channel 变成就绪状态则返回了 1 ；若再次调用 select 方法，如果另一个 Channel 就绪了，它会再次返回1。

获取可操作的Channel

Set selectedKeys = selector.selectedKeys();

当有新增就绪的Channel，调用select()方法，就会将key添加到Set集合中。

3.ByteBuffer：字节缓冲区，本质上是一个连续的字节数组，Selector感兴趣的事件发生后对管道的读操作所读到的数据都存储在ByteBuffer中，而对管道的写操作也是以ByteBuffer为源头，值得注意的是ByteBuffer可以有多个。想想看，我们使用的所有基本类型都可以转换成byte字节，比如Integer类型占4字节，那么传递数字 1 ByteBuffer底层数组被占用了4个格子。

属性
Buffer中有4个非常重要的属性：

capacity、limit、position、mark

capacity属性：容量，Buffer能够容纳的数据元素的最大值，在Buffer初始化创建的时候被赋值，而且不能被修改。

上图中，初始化Buffer的容量为8（图中从0~7，共8个元素），所以capacity = 8

limit属性：代表Buffer可读可写的上限。
写模式下：limit 代表能写入数据的上限位置，这个时候limit = capacity
读模式下：在Buffer完成所有数据写入后，通过调用flip()方法，切换到读模式，此时limit等于Buffer中实际已经写入的数据大小。因为Buffer可能没有被写满，所以limit<=capacity

position属性：代表读取或者写入Buffer的位置。默认为0。
写模式下：每往Buffer中写入一个值，position就会自动加1，代表下一次写入的位置。
读模式下：每往Buffer中读取一个值，position就自动加1，代表下一次读取的位置。

从上图就能很清晰看出，读写模式下capacity、limit、position的关系了。

mark属性：
代表标记，通过mark()方法，记录当前position值，将position值赋值给mark，在后续的写入或读取过程中，可以通过reset()方法恢复当前position为mark记录的值。

这几个重要属性讲完，我们可以再来回顾下：

0 <= mark <= position <= limit <= capacity

现在应该很清晰这几个属性的关系了~

Buffer常见操作

创建Buffer

 allocate(int capacity)

ByteBuffer buffer = ByteBuffer.allocate(1024);
int count = channel.read(buffer);

例子中创建的ByteBuffer是基于堆内存的一个对象。

wrap(array)

wrap方法可以将数组包装成一个Buffer对象:

ByteBuffer buffer = ByteBuffer.wrap("hello world".getBytes());
channel.write(buffer);

allocateDirect(int capacity)

通过allocateDirect方法也可以快速实例化一个Buffer对象，和allocate很相似，这里区别的是allocateDirect创建的是基于堆外内存的对象。堆外内存不在JVM堆上，不受GC的管理。堆外内存进行一些底层系统的IO操作时，效率会更高。

Buffer写操作
Buffer写入可以通过put()和channel.read(buffer)两种方式写入。
通常我们NIO的读操作的时候，都是从Channel中读取数据写入Buffer，这个对应的是Buffer的写操作。
Buffer读操作
Buffer读取可以通过get()和channel.write(buffer)两种方式读入。
还是同上，我们对Buffer的读入操作，反过来说就是对Channel的写操作。读取Buffer中的数据然后写入Channel中。
其他常见方法

rewind()：重置position位置为0，可以重新读取和写入buffer，一般该方法适用于读操作，可以理解为对buffer的重复读。

flip()：很常用的一个方法，一般在写模式切换到读模式的时候会经常用到。也会将position设置为0，然后设置limit等于原来写入的position。

clear()：重置buffer中的数据，该方法主要是针对于写模式，因为limit设置为了capacity，读模式下会出问题。

clear()：重置buffer中的数据，该方法主要是针对于写模式，因为limit设置为了capacity，读模式下会出问题。

public final Buffer rewind() {position = 0;mark = -1;return this;
}
public final Buffer flip() {limit = position;position = 0;mark = -1;return this;
}
public final Buffer clear() {position = 0;limit = capacity;mark = -1;return this;
}
public final Buffer mark() {mark = position;return this;
}
public final Buffer reset() {int m = mark;if (m < 0)throw new InvalidMarkException();position = m;return this;
}

常用的读写方法可以用一张图总结一下：

NIO提供了与传统BIO模型中的Socket和ServerSocket相对应的SocketChannel和ServerSocketChannel两种不同的套接字通道实现。新增的着两种通道都支持阻塞和非阻塞两种模式。
示意图：

使用NIO步骤：（服务端）

首先：创建一个传送带（selector）

然后：创建一个管道(channel)，设置管道为非阻塞，绑定端口

然后：把管道放到传送带上

再然后：启动传送带其次：传送带感兴趣事件检查点查获一个感兴趣管道，转给其他线程对管道进行非阻塞读写

最后：全使用完，关闭管道

NIO服务端通信序列图：

Nio客户端通信时序图：

NIO代码思路：

将 Channel 注册到 Selector 中。

调用 Selector 的 select() 方法，这个方法会阻塞；

到注册在 Selector 中的某个 Channel 有新的 TCP 连接或者可读写事件的话，这个 Channel 就会处于就绪状态，会被 Selector 轮询出来。

然后通过 SelectionKey 可以获取就绪 Channel 的集合，进行后续的 I/O 操作。

NIO创建的Server源码：

package com.company.NIO;
class Server {private static ServerHandle serverHandle;public static void start(){int DEFAULT_PORT = 12345;start(DEFAULT_PORT);}public static synchronized void start(int port){if(serverHandle!=null) {serverHandle.stop();}serverHandle = new ServerHandle(port);new Thread(serverHandle,"Server").start();}public static void main(String[] args){start();}
}

serverHandle:

package com.company.NIO;import com.company.Calculator;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.util.Iterator;
import java.util.Set;/*** NIO服务端*/
class ServerHandle implements Runnable{private Selector selector;private ServerSocketChannel serverChannel;private volatile boolean started;/*** 构造方法* @param port 指定要监听的端口号*/public ServerHandle(int port) {try{//创建选择器selector = Selector.open();//打开监听通道serverChannel = ServerSocketChannel.open();//如果为 true，则此通道将被置于阻塞模式；如果为 false，则此通道将被置于非阻塞模式serverChannel.configureBlocking(false);//开启非阻塞模式//绑定端口 backlog设为1024serverChannel.socket().bind(new InetSocketAddress(port),1024);//监听客户端连接请求serverChannel.register(selector, SelectionKey.OP_ACCEPT);//标记服务器已开启started = true;System.out.println("服务器已启动，端口号：" + port);}catch(IOException e){e.printStackTrace();System.exit(1);}}public void stop(){started = false;}@Overridepublic void run() {//循环遍历selectorwhile(started){try{//无论是否有读写事件发生，selector每隔1s被唤醒一次selector.select(1000);//阻塞,只有当至少一个注册的事件发生的时候才会继续.
//              selector.select();Set<SelectionKey> keys = selector.selectedKeys();Iterator<SelectionKey> it = keys.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();}}//selector关闭后会自动释放里面管理的资源if(selector != null) {try{selector.close();}catch (Exception e) {e.printStackTrace();}}}private void handleInput(SelectionKey key) throws IOException{if(key.isValid()){//处理新接入的请求消息if(key.isAcceptable()){ServerSocketChannel ssc = (ServerSocketChannel) key.channel();//通过ServerSocketChannel的accept创建SocketChannel实例//完成该操作意味着完成TCP三次握手，TCP物理链路正式建立SocketChannel sc = ssc.accept();//设置为非阻塞的sc.configureBlocking(false);//注册为读sc.register(selector, SelectionKey.OP_READ);}//读消息if(key.isReadable()){SocketChannel sc = (SocketChannel) key.channel();//创建ByteBuffer，并开辟一个1M的缓冲区ByteBuffer buffer = ByteBuffer.allocate(1024);//读取请求码流，返回读取到的字节数int readBytes = sc.read(buffer);//读取到字节，对字节进行编解码if(readBytes>0){//将缓冲区当前的limit设置为position=0，用于后续对缓冲区的读取操作buffer.flip();//根据缓冲区可读字节数创建字节数组byte[] bytes = new byte[buffer.remaining()];//将缓冲区可读字节数组复制到新建的数组中buffer.get(bytes);String expression = new String(bytes,"UTF-8");System.out.println("服务器收到消息：" + expression);//处理数据String result = null;try{Calculator calculator=new Calculator(expression);result = calculator.cal().toString();}catch(Exception e){result = "计算错误：" + e.getMessage();}//发送应答消息doWrite(sc,result);}//没有读取到字节 忽略
//              else if(readBytes==0);//链路已经关闭，释放资源else if(readBytes<0){key.cancel();sc.close();}}}}//异步发送应答消息private void doWrite(SocketChannel channel,String response) throws IOException{//将消息编码为字节数组byte[] bytes = response.getBytes();//根据数组容量创建ByteBufferByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);//将字节数组复制到缓冲区writeBuffer.put(bytes);//flip操作writeBuffer.flip();//发送缓冲区的字节数组channel.write(writeBuffer);//****此处不含处理“写半包”的代码}
}

client:

package com.company.NIO;class Client {private static String DEFAULT_HOST = "127.0.0.1";private static int DEFAULT_PORT = 12345;private static ClientHandle clientHandle;public static void start(){start(DEFAULT_HOST,DEFAULT_PORT);}public static synchronized void start(String ip,int port){if(clientHandle!=null) {clientHandle.stop();}clientHandle = new ClientHandle(ip,port);new Thread(clientHandle,"Server").start();}//向服务器发送消息public static boolean sendMsg(String msg) throws Exception{if("q".equals(msg)) {return false;}clientHandle.sendMsg(msg);return true;}public static void main(String[] args){start();}
}

clientHandle:

package com.company.NIO;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.SocketChannel;
import java.util.Iterator;
import java.util.Set;
/*** NIO客户端*/
public class ClientHandle implements Runnable{private String host;private int port;private Selector selector;private SocketChannel socketChannel;private volatile boolean started;public ClientHandle(String ip,int port) {this.host = ip;this.port = port;try{//创建选择器selector = Selector.open();//打开监听通道socketChannel = SocketChannel.open();//如果为 true，则此通道将被置于阻塞模式；如果为 false，则此通道将被置于非阻塞模式socketChannel.configureBlocking(false);//开启非阻塞模式started = true;}catch(IOException e){e.printStackTrace();System.exit(1);}}public void stop(){started = false;}@Overridepublic void run() {try{doConnect();}catch(IOException e){e.printStackTrace();System.exit(1);}//循环遍历selectorwhile(started){try{//无论是否有读写事件发生，selector每隔1s被唤醒一次selector.select(1000);//阻塞,只有当至少一个注册的事件发生的时候才会继续.
//              selector.select();Set<SelectionKey> keys = selector.selectedKeys();Iterator<SelectionKey> it = keys.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(Exception e){e.printStackTrace();System.exit(1);}}//selector关闭后会自动释放里面管理的资源if(selector != null) {try{selector.close();}catch (Exception e) {e.printStackTrace();}}}private void handleInput(SelectionKey key) throws IOException{if(key.isValid()){SocketChannel sc = (SocketChannel) key.channel();if(key.isConnectable()){if(sc.finishConnect()) {;} else {System.exit(1);}}//读消息if(key.isReadable()){//创建ByteBuffer，并开辟一个1M的缓冲区ByteBuffer buffer = ByteBuffer.allocate(1024);//读取请求码流，返回读取到的字节数int readBytes = sc.read(buffer);//读取到字节，对字节进行编解码if(readBytes>0){//将缓冲区当前的limit设置为position=0，用于后续对缓冲区的读取操作buffer.flip();//根据缓冲区可读字节数创建字节数组byte[] bytes = new byte[buffer.remaining()];//将缓冲区可读字节数组复制到新建的数组中buffer.get(bytes);String result = new String(bytes,"UTF-8");System.out.println("客户端收到消息：" + result);}//没有读取到字节 忽略
//              else if(readBytes==0);//链路已经关闭，释放资源else if(readBytes<0){key.cancel();sc.close();}}}}//异步发送消息private void doWrite(SocketChannel channel,String request) throws IOException{//将消息编码为字节数组byte[] bytes = request.getBytes();//根据数组容量创建ByteBufferByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);//将字节数组复制到缓冲区writeBuffer.put(bytes);//flip操作writeBuffer.flip();//发送缓冲区的字节数组channel.write(writeBuffer);//****此处不含处理“写半包”的代码}private void doConnect() throws IOException{if(socketChannel.connect(new InetSocketAddress(host,port))) {;} else {socketChannel.register(selector, SelectionKey.OP_CONNECT);}}public void sendMsg(String msg) throws Exception{socketChannel.register(selector, SelectionKey.OP_READ);doWrite(socketChannel, msg);}
}

NIO Test代码:

import org.junit.Test;
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.util.Date;
import java.util.Iterator;
import java.util.Scanner;/*** 一、使用 NIO 完成网络通信的三个核心：* * 1. 通道（Channel）：负责连接*         *      java.nio.channels.Channel 接口：*            |--SelectableChannel*               |--SocketChannel*               |--ServerSocketChannel*                 |--DatagramChannel* *               |--Pipe.SinkChannel*                |--Pipe.SourceChannel* * 2. 缓冲区（Buffer）：负责数据的存取* * 3. 选择器（Selector）：是 SelectableChannel 的多路复用器。用于监控 SelectableChannel 的 IO 状况* */
public class TestNonBlockingNIO {//客户端@Testpublic void client() throws IOException{//1. 获取通道SocketChannel sChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 9898));//2. 切换非阻塞模式sChannel.configureBlocking(false);//3. 分配指定大小的缓冲区ByteBuffer buf = ByteBuffer.allocate(1024);//4. 发送数据给服务端Scanner scan = new Scanner(System.in);while(scan.hasNext()){String str = scan.next();buf.put((new Date().toString() + "\n" + str).getBytes());buf.flip();sChannel.write(buf);buf.clear();}//5. 关闭通道sChannel.close();}//服务端@Testpublic void server() throws IOException{//1. 获取通道ServerSocketChannel ssChannel = ServerSocketChannel.open();//2. 切换非阻塞模式ssChannel.configureBlocking(false);//3. 绑定连接ssChannel.bind(new InetSocketAddress(9898));//4. 获取选择器Selector selector = Selector.open();//5. 将通道注册到选择器上, 并且指定“监听接收事件”ssChannel.register(selector, SelectionKey.OP_ACCEPT);//6. 轮询式的获取选择器上已经“准备就绪”的事件while(selector.select() > 0){//说明至少有一个准备就绪了//7. 获取当前选择器中所有注册的“选择键(已就绪的监听事件)”Iterator<SelectionKey> it = selector.selectedKeys().iterator();while(it.hasNext()){//8. 获取准备“就绪”的是事件SelectionKey sk = it.next();//9. 判断具体是什么事件准备就绪if(sk.isAcceptable()){//10. 若“接收就绪”，获取客户端连接SocketChannel sChannel = ssChannel.accept();//11. 切换非阻塞模式sChannel.configureBlocking(false);//12. 将该通道注册到选择器上sChannel.register(selector, SelectionKey.OP_READ);}else if(sk.isReadable()){//13. 获取当前选择器上“读就绪”状态的通道SocketChannel sChannel = (SocketChannel) sk.channel();/**14. 读取数据*/ByteBuffer buf = ByteBuffer.allocate(1024);int len = 0;while((len = sChannel.read(buf)) > 0 ){buf.flip();System.out.println(new String(buf.array(), 0, len));buf.clear();}}//15. 取消选择键 SelectionKeyit.remove();}}}
}

与传统IO比较：

传统IO一般是一个线程等待连接，连接过来之后分配给processor线程，processor线程与通道连接后如果通道没有数据过来就会阻塞（线程被动挂起）不能做别的事情。NIO则不同，首先：在Selector线程轮询的过程中就已经过滤掉了不感兴趣的事件，其次：在processor处理感兴趣事件的read和write都是非阻塞操作即直接返回的，线程没有被挂起。
传统IO的管道是单向的，NIO的管道是双向的
两者都是同步的，也就是Java程序亲力亲为的去读写数据，不管传统IO还是NIo都需要read和write方法，这些都是Java程序调用的而不是系统帮我们调用的。NIO2.0里这点得到了改观，即使用异步非阻塞AsynchronousXXX四个类来处理。

五、AIO

AIO 是 Java 1.7 之后引入的包，是 NIO 的升级版本，提供了异步非堵塞的 IO 操作方式，所以人们叫它AIO（Asynchronous IO），异步 IO是基于事件和回调机制实现的，也就是应用操作之后会直接返回，不会堵塞在那里，当后台处理完成，操作系统会通知相应的线程进行后续的操作。与NIO不同，当进行读写操作时，只须直接调用API的read或write方法即可·这两种方法均为异步的，对于读操作而言，当有可读取时，操作系统会将可读的流传入read方法的缓冲区，通知应用程序；对于写操作而言，当操作系统将write方法传递的写入完毕时，操作系统主动涌知应用程序。即可以理解为，read/write方法都是异步的，完成后会主动调用回调函数。在JDK1.7中，这部分内容称作NIO.2。异步非阻塞，服务器实现模式为一个有效请求一个线程，客户端的I/O请求都是由OS先完成了在通知服务器应用去启动线程进行处理，AIO方式适用于链接数目多且连接比较长的架构，比如相册服务器，重复调用OS参与并发操作。 异步的套接字通道时真正的异步非阻塞I/O，对应于UNIX网络编程中的事件驱动I/O（AIO）。他不需要过多的Selector对注册的通道进行轮询即可实现异步读写，从而简化了NIO的编程模型。

流程图：

服务端代码：
AsyncServerHandler:

package com.company.AIO.server;import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.util.concurrent.CountDownLatch;
public class AsyncServerHandler implements Runnable {public CountDownLatch latch;public AsynchronousServerSocketChannel channel;public AsyncServerHandler(int port) {try {//创建服务端通道channel = AsynchronousServerSocketChannel.open();//绑定端口channel.bind(new InetSocketAddress(port));System.out.println("服务器已启动，端口号：" + port);} catch (IOException e) {e.printStackTrace();}}@Overridepublic void run() {//CountDownLatch初始化//它的作用：在完成一组正在执行的操作之前，允许当前的现场一直阻塞//此处，让现场在此阻塞，防止服务端执行完成后退出//也可以使用while(true)+sleep//生成环境就不需要担心这个问题，以为服务端是不会退出的latch = new CountDownLatch(1);//用于接收客户端的连接channel.accept(this,new AcceptHandler());try {latch.await();} catch (InterruptedException e) {e.printStackTrace();}}
}

AcceptHandler:

package com.company.AIO.server;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
//作为handler接收客户端连接
public class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel, AsyncServerHandler> {@Overridepublic void completed(AsynchronousSocketChannel channel,AsyncServerHandler serverHandler) {//继续接受其他客户端的请求Server.clientCount++;System.out.println("连接的客户端数：" + Server.clientCount);serverHandler.channel.accept(serverHandler, this);//创建新的BufferByteBuffer buffer = ByteBuffer.allocate(1024);//异步读  第三个参数为接收消息回调的业务Handlerchannel.read(buffer, buffer, new ReadHandler(channel));}@Overridepublic void failed(Throwable exc, AsyncServerHandler serverHandler) {exc.printStackTrace();serverHandler.latch.countDown();}
}

ReadHandler:

package com.company.AIO.server;
import com.company.Calculator;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
public class ReadHandler implements CompletionHandler<Integer, ByteBuffer> {//用于读取半包消息和发送应答private AsynchronousSocketChannel channel;public ReadHandler(AsynchronousSocketChannel channel) {this.channel = channel;}//读取到消息后的处理@Overridepublic void completed(Integer result, ByteBuffer attachment) {//flip操作attachment.flip();//根据byte[] message = new byte[attachment.remaining()];attachment.get(message);try {String expression = new String(message, "UTF-8");System.out.println("服务器收到消息: " + expression);String calrResult = null;try{Calculator calculator=new Calculator(expression);calrResult = calculator.cal().toString();}catch(Exception e){calrResult = "计算错误：" + e.getMessage();}//向客户端发送消息doWrite(calrResult);} catch (UnsupportedEncodingException e) {e.printStackTrace();}}//发送消息private void doWrite(String result) {byte[] bytes = result.getBytes();ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);writeBuffer.put(bytes);writeBuffer.flip();//异步写数据 参数与前面的read一样channel.write(writeBuffer, writeBuffer,new CompletionHandler<Integer, ByteBuffer>() {@Overridepublic void completed(Integer result, ByteBuffer buffer) {//如果没有发送完，就继续发送直到完成if (buffer.hasRemaining()) {channel.write(buffer, buffer, this);} else{//创建新的BufferByteBuffer readBuffer = ByteBuffer.allocate(1024);//异步读  第三个参数为接收消息回调的业务Handlerchannel.read(readBuffer, readBuffer, new ReadHandler(channel));}}@Overridepublic void failed(Throwable exc, ByteBuffer attachment) {try {channel.close();} catch (IOException e) {}}});}@Overridepublic void failed(Throwable exc, ByteBuffer attachment) {try {this.channel.close();} catch (IOException e) {e.printStackTrace();}}
}

Server:

package com.company.AIO.server;
/*** AIO服务端*/
public class Server {private static int DEFAULT_PORT = 12345;private static AsyncServerHandler serverHandle;public volatile static long clientCount = 0;public static void start(){start(DEFAULT_PORT);}public static synchronized void start(int port){if(serverHandle!=null) {return;}serverHandle = new AsyncServerHandler(port);new Thread(serverHandle,"Server").start();}public static void main(String[] args){Server.start();}
}

客户端代码：
AsyncClientHandler：

package com.company.AIO.client;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.concurrent.CountDownLatch;
public class AsyncClientHandler implements CompletionHandler<Void, AsyncClientHandler>, Runnable {private AsynchronousSocketChannel clientChannel;private String host;private int port;private CountDownLatch latch;public AsyncClientHandler(String host, int port) {this.host = host;this.port = port;try {//创建异步的客户端通道clientChannel = AsynchronousSocketChannel.open();} catch (IOException e) {e.printStackTrace();}}@Overridepublic void run() {//创建CountDownLatch等待latch = new CountDownLatch(1);//发起异步连接操作，回调参数就是这个类本身，如果连接成功会回调completed方法clientChannel.connect(new InetSocketAddress(host, port), this, this);try {latch.await();} catch (InterruptedException e1) {e1.printStackTrace();}try {clientChannel.close();} catch (IOException e) {e.printStackTrace();}}//连接服务器成功//意味着TCP三次握手完成@Overridepublic void completed(Void result, AsyncClientHandler attachment) {System.out.println("客户端成功连接到服务器...");}//连接服务器失败@Overridepublic void failed(Throwable exc, AsyncClientHandler attachment) {System.err.println("连接服务器失败...");exc.printStackTrace();try {clientChannel.close();latch.countDown();} catch (IOException e) {e.printStackTrace();}}//向服务器发送消息public void sendMsg(String msg){byte[] req = msg.getBytes();ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);writeBuffer.put(req);writeBuffer.flip();//异步写clientChannel.write(writeBuffer, writeBuffer,new WriteHandler(clientChannel, latch));}
}

ReadHandler:

package com.company.AIO.client;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.concurrent.CountDownLatch;
public class ReadHandler implements CompletionHandler<Integer, ByteBuffer> {private AsynchronousSocketChannel clientChannel;private CountDownLatch latch;public ReadHandler(AsynchronousSocketChannel clientChannel,CountDownLatch latch) {this.clientChannel = clientChannel;this.latch = latch;}@Overridepublic void completed(Integer result,ByteBuffer buffer) {buffer.flip();byte[] bytes = new byte[buffer.remaining()];buffer.get(bytes);String body;try {body = new String(bytes,"UTF-8");System.out.println("客户端收到结果:"+ body);} catch (UnsupportedEncodingException e) {e.printStackTrace();}}@Overridepublic void failed(Throwable exc,ByteBuffer attachment) {System.err.println("数据读取失败...");try {clientChannel.close();latch.countDown();} catch (IOException e) {}}
}

WriteHandler:

package com.company.AIO.client;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.concurrent.CountDownLatch;
public class WriteHandler implements CompletionHandler<Integer, ByteBuffer> {private AsynchronousSocketChannel clientChannel;private CountDownLatch latch;public WriteHandler(AsynchronousSocketChannel clientChannel,CountDownLatch latch) {this.clientChannel = clientChannel;this.latch = latch;}@Overridepublic void completed(Integer result, ByteBuffer buffer) {//完成全部数据的写入if (buffer.hasRemaining()) {clientChannel.write(buffer, buffer, this);}else {//读取数据ByteBuffer readBuffer = ByteBuffer.allocate(1024);clientChannel.read(readBuffer,readBuffer,new ReadHandler(clientChannel, latch));}}@Overridepublic void failed(Throwable exc, ByteBuffer attachment) {System.err.println("数据发送失败...");try {clientChannel.close();latch.countDown();} catch (IOException e) {}}
}

Client:

package com.company.AIO.client;
import java.util.Scanner;
public class Client {private static String DEFAULT_HOST = "127.0.0.1";private static int DEFAULT_PORT = 12345;private static AsyncClientHandler clientHandle;public static void start(){start(DEFAULT_HOST,DEFAULT_PORT);}public static synchronized void start(String ip,int port){if(clientHandle!=null) {return;}clientHandle = new AsyncClientHandler(ip,port);new Thread(clientHandle,"Client").start();}//向服务器发送消息public static boolean sendMsg(String msg) throws Exception{if("q".equals(msg)) {return false;}clientHandle.sendMsg(msg);return true;}@SuppressWarnings("resource")public static void main(String[] args) throws Exception{Client.start();System.out.println("请输入请求消息：");Scanner scanner = new Scanner(System.in);while(Client.sendMsg(scanner.nextLine()));}
}

六、附一张总结图

参考文章1
参考文章2
参考文章3
参考文章4

BIO-NIO-AIO相关推荐

关于BIO | NIO | AIO的讨论
关于BIO | NIO | AIO的讨论一直存在,有时候也很容易让人混淆,就我的理解,给出一个解释: BIO | NIO | AIO,本身的描述都是在Java语言的基础上的.而描述IO,我们需要从两个 ...
Netty序章之BIO NIO AIO演变
Netty序章之BIO NIO AIO演变 Netty是一个提供异步事件驱动的网络应用框架,用以快速开发高性能.高可靠的网络服务器和客户端程序.Netty简化了网络程序的开发,是很多框架和公司都在使用 ...
【面试】迄今为止把同步/异步/阻塞/非阻塞/BIO/NIO/AIO讲的这么清楚的好文章（快快珍藏）...
网上有很多讲同步/异步/阻塞/非阻塞/BIO/NIO/AIO的文章,但是都没有达到我的心里预期,于是自己写一篇出来. 常规的误区假设有一个展示用户详情的需求,分两步,先调用一个HTTP接口拿到详情数 ...
也谈BIO | NIO | AIO （Java版--转）
http://my.oschina.net/bluesky0leon/blog/132361 关于BIO | NIO | AIO的讨论一直存在,有时候也很容易让人混淆,就我的理解,给出一个解释: BI ...
迄今为止把同步/异步/阻塞/非阻塞/BIO/NIO/AIO讲的这么清楚的好文章
来源:编程新说网上有很多讲同步/异步/阻塞/非阻塞/BIO/NIO/AIO的文章,但是都没有达到我的心里预期,于是自己写一篇出来. 常规的误区假设有一个展示用户详情的需求,分两步,先调用一个HTT ...
IO: BIO ? NIO ? AIO?
IO的方式通常分为几种,同步阻塞的BIO.同步非阻塞的NIO.异步非阻塞的AIO. 一.BIO 在JDK1.4出来之前,我们建立网络连接的时候采用BIO模式,需要先在服务端启动一个ServerSock ...
内核aio_今天来说说令人让人傻傻分不清的BIO,NIO,AIO
| 作者:新一. | 简书:https://www.jianshu.com/u/b3263fc54bce | 知乎:https://www.zhihu.com/people/qing-ni-chi-y ...
bio阻塞的缺点_java 中的 BIO/NIO/AIO 详解
java 的 IO 演进之路我们在前面学习了 linux 的 5 种 I/O 模型详解下面我们一起来学习下如何使用 java 实现 BIO/NIO/AIO 这 3 种不同的网络 IO 模型编程. ...
Java的IO：BIO | NIO | AIO
原文: http://my.oschina.net/bluesky0leon/blog/132361 BIO | NIO | AIO,本身的描述都是在Java语言的基础上的.而描述IO,我们需要从两个 ...
Java IO(BIO, NIO, AIO) 总结
文章转载自:JavaGuide 目录 BIO,NIO,AIO 总结同步与异步阻塞和非阻塞 1. BIO (Blocking I/O) 1.1 传统 BIO 1.2 伪异步 IO 1.3 代码示例 ...