Java网络编程（4） - BIO、NIO、AIO的区别，阻塞、非阻塞、同步、异步的区别，什么是NIO同步非阻塞详解？什么是AIO异步非阻塞？NIO的三大组件是什么？
多种IO区别（BIO、NIO、AIO）
传统的Socket链接还是伪异步，都是用的BIO的链接，而JDK为了解决这种通信的问题推出了NIO和AIO去解决这个问题。
|
BIO（blocking I/O） |
NIO（non-blocking I/O） |
AIO（Asychronous I/O） |
|
同步、阻塞 |
同步、非阻塞 |
异步、非阻塞 |
|
面向流 |
面向缓冲 |
面向系统 |
|
服务器实现模式为一个链接一个线程，这种IO的问题可以通过伪异步的方式去优化。 |
服务器实现模式为一个请求一个线程，客户端发送的链接请求都会注册到多路复用器上，多路复用器到链接I/O请求时，才启动一个线程去处理。 |
也称为NIO 2.0，服务器实现模式为一个有效请求一个线程，客户端的I/O请求都是由OS先完成了再通知服务器应用去启动线程处理。 |
|
适合连接数小且固定的架构，对服务器资源要求比较好，并发局限于应用中，JDK1.5之前是唯一的选择，但呈现直观简单，易于理解。 |
适合连接数多且链接数据比较短的轻操作的架构，比如聊天服务器，并发局限于应用中，编程比较复杂。 |
适合链接数多且链接比较长的重操作的架构，比如相册服务器，充分调用OS参与并发操作，编程复杂。 |
|
JDK1.5以前 |
JDK1.5支持 |
JDK1.7支持 |
阻塞、非阻塞、同步、异步的区别
|
概念 |
解释 |
示例 |
|
阻塞 |
应用程序在获取网络数据时，如果网络传输很慢，那么程序一直等待，直到传输完成。 |
ABC：A交代B去挑水，B没干完这件事，AB都不准回来。C去买菜。 |
|
非阻塞 |
应用程序直接可以获取已经准备就绪的数据，无需等待，数据在缓冲区加载完毕就直接在缓冲区获取。 |
ABC：A交代B去挑水，A放了个远程摄像机监督B，A去干别的事，B完事，通知A。C去买菜。 |
|
同步 |
面向应用程序，应用程序会直接参与IO读写操作，并且应用程序会直接阻塞到某个方法上，直到数据准备就绪。或者采用轮询的方式实时检查数据的就绪状态，如果就绪则获取数据。 |
ABC：A交代线程B去挑水，B没干完这事，ABC就都一直一起等。 |
|
异步 |
面向操作系统，所有IO读写操作交给操作系统处理，应用程序不关心IO读写，而可以去做其他事情，但操作系统完成IO读写操作时，会给应用程序发出通知，应用程序拿走数据即可。 |
ABC：A交代B去挑水，A放了个远程摄像机监督B，A去干别的事，B完事，通知A。C去买菜。 |
NIO（non-blocking I/O）同步非阻塞详解与案例
NIO就是利用了多路复用技术，在IO编程过程中，需要同时处理多个客户端接入请求时，可以利用多线程或IO多路复用技术进行处理。
IO多路复用技术会把多个IO阻塞复用到同一个select的阻塞上，从而使得系统在单线程的情况下，可以同时处理多个客户端的请求。
与传统的多线程/多进程模型相比，IO多路复用的最大优势在于系统开销低，系统不需要创建新的额外进程或现场，也吧需要维护这些进程和线程，降低了系统的维护工作量，节省了系统资源。
NIO本质上是避免原始的TCP建立连接使用3次握手的操作，减少连接开销，而是连接到Selector。所以NIO设计了三大组件来实现这个结果。
Buffer：缓冲区就是内存的一块区域，把数据写存储到内存，然后一次性写入，类似数据库的批量操作，这样大大的提高了数据的读写速度。
NIO的数据是先到缓冲区的，而缓冲区是一个对象，包含一些写入或者要读取的数据，在NIO类库中加入Buffer对象，体现了新库和原IO的一个重要区别。
在面向流的IO中，可以将数据直接写入或读取到Stream对象中。在NIO库中，所有的数据都是用华冲去处理的读写。
缓冲区本质就是一个数组，通常是一个字节数组（ByteBuffer），也可以使用其他类型的数组。这个数组为缓冲区提供了数据的访问读写等操作属性，如位置、容量等。
备注：实际上每一种Java基本类型都对应了一种缓冲区，除了Boolean类。
Channel：称为管道、通道，就像自来水管道一样，网络数据通过Channel读取和写入，通道和流不同之处在于通道是双向的，而流只是一个方向上移动（一个流必须是InputStream或OutputStream的子类），而通道可以用于读、写或者二者同时进行，最关键的是可以与多路分复用器结合起来，有多种状态位，方便多路复用器去识别。
事实上通道分为两大类，一是网络读写（SelectableChannel）；二是用于文件操作（FileChannel）；无论哪一类都是SelectableChannel的子类。
Selector：称为选择器、多路复用器，是NIO的基础，提供了选择一个就绪的任务能力。就是不断的轮询注册在其上的通道（Channel），如果某个通道发生了读写操作，这个通道就处于就绪状态，会被Selector轮询出来，然后通过SelectionKey可以取得就绪的Channel集合，从而进行后续的IO操作。
一个Selector可以负责成千上万的Channel通道，理论中没有上限，这也是JDK使用了epool代替了传统的select实现，获得链接句柄没有限制，意味着只要一个线程负责Selector轮询，就可以介入成千上万个客户端，这就是JDK对NIO类库的优化。
Selector类似一个管理者（Master），管理了成千上万个通道，然后轮询那个通道的数据已经准备好了，就通知CPU执行IO的读写或写入操作。
Selector模式就是，当IO事件（Channel）注册到选择器以后，Selector会分类给每一个Channel一个Key。Selector以轮询的方式查找注册的所有IO事件（Channel），当IO事件（Channel）准备就绪后，Selector就会识别，通过Key值来查找到相对应的Channel，进行相关的数据处理操作（从管道读或写，写数据到缓冲区）。
缓冲区（Buffer）使用示例
|
/** * NIO的数据是到缓冲区 * 每一种基本类型都对应了一种NIO缓冲区，除了Boolean */ public class N03NIOBuffer { public static void main(String[] args) {
//基本操作 ============== //指定长度的缓冲区 IntBuffer intBuffer = IntBuffer.allocate(10); intBuffer.put(111);//位置：0 -> 1 intBuffer.put(222);//位置：1 -> 2 intBuffer.put(333);//位置：2 -> 3
System.out.println(intBuffer); System.out.println("容量：" + intBuffer.capacity()); System.out.println("限制：" + intBuffer.limit()); System.out.println("下标1的元素：" + intBuffer.get(1));
//把位置复位为0，也就是position位置：3 -> 0 intBuffer.flip();
System.out.println("---------------------------------");
System.out.println(intBuffer); System.out.println("容量：" + intBuffer.capacity()); System.out.println("限制：" + intBuffer.limit()); System.out.println("下标1的元素：" + intBuffer.get(1));
intBuffer.put(2, 3); System.out.println("position不变：" + intBuffer);
System.out.println("---------------------------------");
for (int i = 0; i < intBuffer.limit(); i++) { System.out.println(intBuffer.get()); }
//wrap使用 ============== /* * wrap方法会包裹一个数组，一般这种用法不会先初始化缓存对象的长度， * 因为没有意义，最后还会被wrap所包裹的数组覆盖掉，并且wrap方法会修改缓冲区对象的时候，数组本身也会跟着发生变化。 */ int [] arr = new int[] {1,2,3}; IntBuffer wrap = IntBuffer.wrap(arr); System.out.println(wrap);
//标识容量为数组的长度，但是可操作元素只有实际进入缓存的元素长度 IntBuffer wrap2 = IntBuffer.wrap(arr,0,1); System.out.println(wrap2);
System.out.println("---------------------------------");
//其他方法 ============== IntBuffer intBuffer2 = IntBuffer.allocate(10); int [] arr2 = new int[] {1,2,3}; intBuffer2.put(arr2); System.out.println(intBuffer2);
//复制 IntBuffer duplicate = intBuffer2.duplicate(); System.out.println(duplicate);
//重置位置 // intBuffer2.flip(); // intBuffer2.position(0); System.out.println(intBuffer2); System.out.println("可读数据：" + intBuffer2.remaining());
int [] arr3 = new int[intBuffer2.remaining()]; //将缓冲区数据放入到数组中 intBuffer2.get(arr3); for (int i : arr3) { System.out.print(i + " "); } } } |
NIO（non-blocking I/O）同步非阻塞案例
|
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;
/** * NIO（同步非阻塞）服务端 */ public class N04NIOServer { public static void main(String[] args) { new Thread(new NServer(2222)).start(); } } class NServer implements Runnable {
//多路复用器，管理所有的Client链接管道 private Selector selector;
//建立读缓冲区，用于接收数据 private ByteBuffer readBuffer = ByteBuffer.allocate(10);
//建立写缓冲区，用于接收数据 private ByteBuffer writeBuffer = ByteBuffer.allocate(10);
public NServer(int port) { try { //打开多路复用器，让所有都注册到这里面来 this.selector = Selector.open();
//打开服务器链接通道 ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
//把服务器链接通道设置为非阻塞模式 serverSocketChannel.configureBlocking(false);
//设置服务端的端口和地址 serverSocketChannel.bind(new InetSocketAddress(port));
//把服务器打开的通道注册到多路复用器上，都交给其管理，并监听阻塞事件 serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT);
System.out.println("服务端启动...");
} catch (Exception e) { e.printStackTrace(); } }
@Override public void run() { while (true) { try { //让多路复用器开始监听 this.selector.select();
//获得多路复用器已经选择的结果集 Iterator<SelectionKey> keys = this.selector.selectedKeys().iterator();
while (keys.hasNext()) { //获取到选择的元素 SelectionKey key = keys.next();
//从容器移除。 keys.remove();
//验证是否有效 if (key.isValid()) { //阻塞状态 if (key.isAcceptable()) { this.accept(key); } //可读状态 if(key.isReadable()) { this.read(key); } //可写状态 if(key.isReadable()) { this.write(key); } } }
} catch (Exception e) { e.printStackTrace(); } } }
/** * 阻塞状态处理 */ private void accept(SelectionKey key) { try { //获取到服务通道 ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
//执行阻塞方法 SocketChannel socketChannel = serverSocketChannel.accept();
//设置非阻塞模式 socketChannel.configureBlocking(false);
//注册到多路复用器，并设置读取标识 socketChannel.register(this.selector, SelectionKey.OP_READ); } catch (Exception e) { e.printStackTrace(); } }
/** * 可读状态处理 */ private void read(SelectionKey key) { try { //清空缓冲区的旧数据 this.readBuffer.clear();
//获取到之前注册的socket通道对象 SocketChannel socketChannel = (SocketChannel) key.channel();
//读取数据 int read = socketChannel.read(this.readBuffer);
//没有数据 if(read == -1) { key.channel().close(); key.cancel(); return ; }
//有数据则进行读取，读取之前需要进行复位（position和limit进行复位） readBuffer.flip();
//根据缓冲区的数据读取 byte [] data = new byte[this.readBuffer.remaining()]; this.readBuffer.get(data);
System.out.println("服务端：" + new String(data));
} catch (Exception e) { e.printStackTrace(); } }
/** * 可写状态处理 */ private void write(SelectionKey key) { try { //获取到服务通道 // ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
//注册到多路复用器 // serverSocketChannel.register(this.selector, SelectionKey.OP_WRITE); } catch (Exception e) { e.printStackTrace(); } } } |
|
import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.SocketChannel;
/** * NIO（同步非阻塞）客户端 */ public class N04NIOClient { public static void main(String[] args) { System.out.println("客户端启动..."); //打开链接通道 try (SocketChannel socketChannel = SocketChannel.open();){
//链接地址 InetSocketAddress inetSocketAddress = new InetSocketAddress(2222);
//建立缓冲区，写读数据 ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
//开始链接 socketChannel.connect(inetSocketAddress);
while (true) { //将数据写到缓冲区，并复位缓冲区 byteBuffer.put("111".getBytes()); byteBuffer.flip();
//写出数据 socketChannel.write(byteBuffer);
//清空缓冲区 byteBuffer.clear();
} } catch (Exception e) { e.printStackTrace(); } } } |
AIO（Asychronous I/O）异步非阻塞案例
在NIO基础上引入了异步通道的概念，并提供了异步文件和异步套接字通道的实现，从而在真正意义上实现了异步非阻塞，NIO只是非阻塞而非异步。
AIO不需要通过多路复用器对注册的通道进行轮询操作即可实现异步读写，客户端的I/O请求都是由OS先完成了再通知服务器应用去启动线程处理，从而简化了NIO模型。
|
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;
/** * AIO（异步非阻塞）服务端 */ public class N05AIOServer { public static void main(String[] args) { new AServer(2222); } } class AServer { //线程池 private ExecutorService executorService;
//通信通道线程组 private AsynchronousChannelGroup threadGroup;
//通信通道 private AsynchronousServerSocketChannel serverSocketChannel;
public AServer(int port) { try {
//初始化线程池 executorService = Executors.newCachedThreadPool();
//初始化通信通道线程池组 threadGroup = AsynchronousChannelGroup.withCachedThreadPool(executorService, 1);
//打开服务端通道 serverSocketChannel = AsynchronousServerSocketChannel.open(threadGroup);
//以上的初始化中，已经把通道都交给线程池做管理了。
//服务端绑定端口 serverSocketChannel.bind(new InetSocketAddress("127.0.0.1",port));
System.out.println("服务端启动...");
/* * 进行阻塞 * 第一个参数：运行这个服务端的对象，this，即AServer * 第二个参数：处理业务的类 */ serverSocketChannel.accept(this, new AServerHandler());
//会发现不在需要自己去做Selector监听，API内部会自动完成
/* * 一直阻塞，不让服务器停止 * 要知道这是异步的，如果把Thread.sleep(Integer.MAX_VALUE);注释掉，会发现启动就停止。 * 在生产环境中，这个异步线程的生命是随着如Tomcat容器的生命周期而结束的。 */ Thread.sleep(Integer.MAX_VALUE); } catch (Exception e) { e.printStackTrace(); } }
public AsynchronousServerSocketChannel getServerSocketChannel() { return serverSocketChannel; } } /** * 服务端的实际处理 * 第一个：通道 * 第二个：运行服务端的类 */ class AServerHandler implements CompletionHandler<AsynchronousSocketChannel,AServer> {
@Override public void completed(AsynchronousSocketChannel result, AServer attachment) {
//在运行时，将这个注释掉，会发现3个客户端的链接，只有1个是链接了。 //但有下一个客户端接入的时候，直接调用Server的accept方法，这样反复执行下去，保证多个客户端都可以阻塞 attachment.getServerSocketChannel().accept(attachment, this);
read(result); }
/** * 读取数据 */ private void read(AsynchronousSocketChannel result) { //读取数据 ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
result.read(byteBuffer,byteBuffer,new CompletionHandler<Integer, ByteBuffer>() {
@Override public void completed(Integer resultSize, ByteBuffer attachment) { //重置标识位 attachment.flip(); System.out.println("服务端收到的数据：" + new String(attachment.array())); System.out.println("服务端收到的数据长度：" + resultSize);
//回应客户端 write(result); }
@Override public void failed(Throwable exc, ByteBuffer attachment) { exc.printStackTrace(); } });
}
/** * 响应数据 */ protected void write(AsynchronousSocketChannel result) { //把数据放到缓冲区 ByteBuffer byteBuffer = ByteBuffer.allocate(1024); byteBuffer.put("欢迎客户端的到来".getBytes()); byteBuffer.flip();
try { //写出给客户端 result.write(byteBuffer).get(); } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); } }
@Override public void failed(Throwable exc, AServer attachment) { exc.printStackTrace(); } } |
|
import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.AsynchronousSocketChannel;
/** * AIO（异步非阻塞）客户端 */ public class N05AIOClient { public static void main(String[] args) throws Exception { AClient c1 = new AClient(); AClient c2 = new AClient(); AClient c3 = new AClient();
c1.connect(); c2.connect(); c3.connect();
new Thread(c1).start(); new Thread(c2).start(); new Thread(c3).start();
Thread.sleep(1000);
c1.wirte("c1 1111111"); c2.wirte("c2 2222222222"); c3.wirte("c2 33333333333333"); } } /** * 客户端 */ class AClient implements Runnable {
private AsynchronousSocketChannel socketChannel;
/** * 初始化时，打开通道，并开始链接 */ public AClient() throws IOException { super(); this.socketChannel = AsynchronousSocketChannel.open(); }
/** * 开始链接 */ public void connect() { socketChannel.connect(new InetSocketAddress("127.0.0.1",2222)); }
/** * 写出数据给服务端 */ public void wirte(String text) throws Exception { socketChannel.write(ByteBuffer.wrap(text.getBytes())); //读取服务端的响应数据 read(); }
/** * 读取数据 */ private void read() throws Exception { //读取数据到缓冲区 ByteBuffer byteBuffer = ByteBuffer.allocate(1024); socketChannel.read(byteBuffer).get(); byteBuffer.flip();
//获得数据 byte[] data = new byte[byteBuffer.remaining()]; byteBuffer.get(data); System.out.println("服务端响应：" + new String(data)); }
/* * 让客户端一直运行 */ @Override public void run() { while (true) {
} } } |