谈谈java的bio、nio、aio模型

素颜马尾好姑娘i 2023-03-02 10:54 21阅读 0赞

**目录**

socket

IO(BIO)和NIO的区别

同步和异步

bio：同步阻塞式IO

NIO：同步非阻塞IO（工作中用的很少）

Buffer使用

NIO代码

AIO

--------------------

# socket #

Socket又称“套接字”，应用程序通常通过“套接字”向网络发出请求或者应答请求。

套接字之间的链接过程可以分为四个步骤：服务器监听、客户端请求服务器、服务器确认、客户端确认、进行通信。

# IO(BIO)和NIO的区别 #

其本质就是阻塞和非阻塞的区别。

阻塞概念：应用程序在获取网络数据的时候，如果网络传输数据很慢，那么程序就一直等着，直到传输完毕为止。

非阻塞概念：应用程序直接可以获取已经准备就绪好的数据，无需等待。

IO为同步阻塞形式，NIO为同步非阻塞形式。NIO并没有实现异步，在JDK1.7之后，升级了NIO库包，支持异步非阻塞通信模型即NIO2.0（AIO）。

# 同步和异步 #

同步和异步一般是面向操作系统与应用程序对IO操作的层面上来区别的。

同步时，应用程序会直接参与IO读写操作，并且我们的应用程序会直接阻塞到某一个方法上，直到数据准备就绪：或者采用轮询的策略实时监察数据的就绪状态，如果就绪则获取数据。

异步时，则所有的IO读写操作交给操作系统处理，与我们的应用程序没有直接关系，我们程序不需要关心IO读写，当操作系统完成了IO读写操作时，会给我们应用程序发送通知，我们的应用程序直接拿走数据即可。

同步说的是你的server服务器端的执行方式。

阻塞说的是具体的技术，接收数据的方式、状态（IO、NIO）

# bio：同步阻塞式IO #

//客户端
    public class Client {
         final static String ADDRESS = "127.0.0.1";
         final static int PORT = 8765;
         
         public static void main(String[] args) {
               
               Socket socket = null;
               BufferedReader in = null;
               PrintWriter out = null;
               
               try {
                    socket = new Socket(ADDRESS, PORT);
                    in = new BufferedReader(new  InputStreamReader(socket.getInputStream()));
                    out = new PrintWriter(socket.getOutputStream(),  true);
                    
                    //向服务器端发送数据
                    out.println("接收到客户端的请求数据...");
                    String response = in.readLine();
                    System.out.println("Client: " + response);
                    
               } catch (Exception e) {
                    e.printStackTrace();
               } finally {
                    if(in != null){
                         try {
                               in.close();
                         } catch (IOException e) {
                               e.printStackTrace();
                         }
                    }
                    if(out != null){
                         try {
                               out.close();
                         } catch (Exception e) {
                               e.printStackTrace();
                         }
                    }
                    if(socket != null){
                         try {
                               socket.close();
                         } catch (IOException e) {
                               e.printStackTrace();
                         }
                    }
                    socket = null;
               }
         }
    }
    
    
    //服务器端
    public class Server {
         final static int PROT = 8765;
         
         public static void main(String[] args) {
               
               ServerSocket server = null;
               try {
                    server = new ServerSocket(PROT);
                    System.out.println(" server start .. ");
                    //进行阻塞
                    Socket socket = server.accept();
                    //新建一个线程执行客户端的任务
                    new Thread(new ServerHandler(socket)).start();
                    
               } catch (Exception e) {
                    e.printStackTrace();
               } finally {
                    if(server != null){
                         try {
                               server.close();
                         } catch (IOException e) {
                               e.printStackTrace();
                         }
                    }
                    server = null;
               }
         }
    }
    
    
    public class ServerHandler implements Runnable{
         private Socket socket ;
         
         public ServerHandler(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 body = null;
                    while(true){
                         body = in.readLine();
                         if(body == null) break;
                         System.out.println("Server :" + body);
                         out.println("服务器端回送响的应数据.");
                    }
                    
               } catch (Exception e) {
                    e.printStackTrace();
               } finally {
                    if(in != null){
                         try {
                               in.close();
                         } catch (IOException e) {
                               e.printStackTrace();
                         }
                    }
                    if(out != null){
                         try {
                               out.close();
                         } catch (Exception e) {
                               e.printStackTrace();
                         }
                    }
                    if(socket != null){
                         try {
                               socket.close();
                         } catch (IOException e) {
                               e.printStackTrace();
                         }
                    }
                    socket = null;
               }
         }
    }

//（JDK1.5之前没有出NIO的解决办法）：BIO+线程池，降低服务器端的压力
    {
         server = new ServerSocket(PORT);
         System.out.println("server start");
         Socket socket = null;
         HandlerExecutorPool executorPool = new  HandlerExecutorPool(50, 1000);//自定义线程池
         while(true){
               socket = server.accept();
               executorPool.execute(new ServerHandler(socket));
         }
    }
    
    
    //自定义线程池
    public class HandlerExecutorPool {
         private ExecutorService executor;
         public HandlerExecutorPool(int maxPoolSize, int  queueSize){
               this.executor = new ThreadPoolExecutor(
                         Runtime.getRuntime().availableProcessors(),
                         maxPoolSize,
                         120L,
                         TimeUnit.SECONDS,
                         new  ArrayBlockingQueue<Runnable>(queueSize));
         }
         
         public void execute(Runnable task){
               this.executor.execute(task);
         }
    }

# NIO：同步非阻塞IO（工作中用的很少） #

几个概念：Buffer（缓冲区）、Channel（管道、通道）、Selector（选择器、多路复用器）

NIO的本质就是避免原始的TCP建立连接使用3次握手的操作，减少连接的开销。

![20200729095957469.png][]

## Buffer使用 ##

Buffer是一个对象，它包含一些要写入或者要读取的数据。在NIO类库中加入Buffer对象，体现了新库与原IO的一个重要的区别。在面向流的IO中，可以将数据直接写入或读取到Stream对象中。在NIO库中，所有数据都是用缓冲区处理的（读写）。缓冲区实质上是一个数组，通常它是一个字节数组（ByteBuffer），也可以使用其他类型的数据。这个数组为缓冲区提供了数据的访问读写等操作属性，如位置、容量、上限等概念，参考API文档。

Buffer类型：我们最常用的就是ByteBuffer，实际上每一种java基本类型都对应了一种缓冲区（除了Boolean类型）

ByteBuffer、CharBuffer、ShortBuffer、IntBuffer、LongBuffer、FloatBuffer、DoubleBuffer

/** 
    各种类型的buffer用法是一样的。
    pos（位置）、lim（限制）、cap（容量）
    put(x)方法会将该元素放置到当前位置，并且位置+1；但是put(x, x)方法不会改变当前位置。
    capacity()获取容量；limit()获取限制，而且是从当前位置开始获取的限制。
    get()获取当前位置的数据，并且位置+1。
    flip()复位当前位置，并且修改限制为复位之前的位置。最好在put完毕之后flip()一下。
    position(0)设置当前位置为0。
    remaining()可读数据长度，位置到限制（pos-lim）数量。
    */
    // 1 基本操作
    //创建指定长度的缓冲区
    IntBuffer buf = IntBuffer.allocate(10);
    //put还有很多重载方法
    buf.put(13);// position位置：0 - > 1
    buf.put(21);// position位置：1 - > 2
    buf.put(35);// position位置：2 - > 3
    //把位置复位为0，也就是position位置：3 - > 0
    buf.flip();
    System.out.println("使用flip复位：" + buf);//[pos=3 lim=10  cap=10]位置、限制、容量
    System.out.println("容量为: " + buf.capacity()); //容量一旦初始化后不允许改变（warp方法包裹数组除外）
    System.out.println("限制为: " + buf.limit());         //由于只装载了三个元素,所以可读取或者操作的元素为3 则limit=3
    System.out.println("获取下标为1的元素：" + buf.get(1));
    System.out.println("get(index)方法，position位置不改变：" + buf);
    buf.put(1, 4);//把下标为1的位置赋值4
    System.out.println("put(index, change)方法，position位置不变：" +  buf);;
    for (int i = 0; i < buf.limit(); i++) {
         //调用get方法会使其缓冲区位置（position）向后递增一位
         System.out.print(buf.get() + "\t");
    }
    System.out.println("buf对象遍历之后为: " + buf);

// 2 wrap方法使用
    //  wrap方法会包裹一个数组: 一般这种用法不会先初始化缓存对象的长度，因为没有意义，最后还会被wrap所包裹的数组覆盖掉。
    //  并且wrap方法修改缓冲区对象的时候，数组本身也会跟着发生变化。
    int[] arr = new int[]{1,2,5};
    IntBuffer buf1 = IntBuffer.wrap(arr);//[pos=0 lim=3 cap=3]
    System.out.println(buf1);
    IntBuffer buf2 = IntBuffer.wrap(arr, 0 , 2);//截取0-2[pos=0  lim=2 cap=3]
    //这样使用表示容量为数组arr的长度，但是可操作的元素只有实际进入缓存区的元素长度
    System.out.println(buf2);

// 3 其他方法
    IntBuffer buf1 = IntBuffer.allocate(10);
    int[] arr = new int[]{1,2,5};
    buf1.put(arr);
    System.out.println(buf1);
    //一种复制方法，完全复制
    IntBuffer buf3 = buf1.duplicate();
    System.out.println(buf3);
    //设置buf1的位置属性注意这俩的区别
    buf1.position(0);
    buf1.flip();
    System.out.println(buf1);
    System.out.println("可读数据为：" + buf1.remaining());
    int[] arr2 = new int[buf1.remaining()];
    //将缓冲区数据放入arr2数组中去
    buf1.get(arr2);
    for(int i : arr2){
         System.out.print(Integer.toString(i) + ",");
    }

# 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.ServerSocketChannel;
    import java.nio.channels.SocketChannel;
    import java.util.Iterator;
    public class Server implements Runnable{
         //1 多路复用器（管理所有的通道）
         private Selector seletor;
         //2 建立缓冲区
         private ByteBuffer readBuf = ByteBuffer.allocate(1024);
         //3
         private ByteBuffer writeBuf = ByteBuffer.allocate(1024);
         public Server(int port){
               try {
                    //1 打开路复用器
                    this.seletor = Selector.open();
                    //2 打开服务器通道
                    ServerSocketChannel ssc =  ServerSocketChannel.open();
                    //3 设置服务器通道为非阻塞模式
                    ssc.configureBlocking(false);
                    //4 绑定地址
                    ssc.bind(new InetSocketAddress(port));
                    //5 把服务器通道注册到多路复用器上，并且监听阻塞事件
                    ssc.register(this.seletor,  SelectionKey.OP_ACCEPT);
                    
                    System.out.println("Server start, port :" +  port);
                    
               } catch (IOException e) {
                    e.printStackTrace();
               }
         }
         @Override
         public void run() {
               while(true){
                    try {
                         //1 必须要让多路复用器开始监听
                         this.seletor.select();
                         //2 返回多路复用器已经选择的结果集
                         Iterator<SelectionKey> keys =  this.seletor.selectedKeys().iterator();
                         //3 进行遍历
                         while(keys.hasNext()){
                               //4 获取一个选择的元素
                               SelectionKey key = keys.next();
                               //5 直接从容器中移除就可以了
                               keys.remove();
                               //6 如果是有效的
                               if(key.isValid()){
                                    //7 如果为阻塞状态
                                    if(key.isValid() &&  key.isAcceptable()){
                                         this.accept(key);
                                    }
                                    //8 如果为可读状态
                                    if(key.isValid() &&  key.isReadable()){
                                         this.read(key);
                                    }
                                    //9 写数据
                                    if(key.isValid() &&  key.isWritable()){
                                         //this.write(key); //ssc
                                    }
                               }
                         }
                    } catch (IOException e) {
                         e.printStackTrace();
                    }
               }
         }
         
         private void write(SelectionKey key){
               //ServerSocketChannel ssc =  (ServerSocketChannel)  key.channel();
               //ssc.register(this.seletor, SelectionKey.OP_WRITE);
         }
         private void read(SelectionKey key) {
               try {
                    //1 清空缓冲区旧的数据
                    this.readBuf.clear();
                    //2 获取之前注册的socket通道对象
                    SocketChannel sc = (SocketChannel)  key.channel();
                    //3 读取数据
                    int count = sc.read(this.readBuf);
                    //4 如果没有数据
                    if(count == -1){
                         key.channel().close();
                         key.cancel();
                         return;
                    }
                    //5 有数据则进行读取 读取之前需要进行复位方法(把position 和limit进行复位)
                    this.readBuf.flip();
                    //6 根据缓冲区的数据长度创建相应大小的byte数组，接收缓冲区的数据
                    byte[] bytes = new  byte[this.readBuf.remaining()];
                    //7 接收缓冲区数据
                    this.readBuf.get(bytes);
                    //8 打印结果
                    String body = new String(bytes).trim();
                    System.out.println("Server : " + body);
                    
                    // 9..可以写回给客户端数据
                    // 回发数据，并关闭channel
                    sc.configureBlocking(false);
                    //注册到多路复用器上，并设置读取标识
                    sc.register(this.seletor, SelectionKey.OP_READ);
                    ByteBuffer sendBuffer = ByteBuffer.wrap(("我是"  + body).getBytes());
                    sc.write(sendBuffer);
                    //sc.close();
                    
               } catch (IOException e) {
                    //e.printStackTrace();
                    SocketChannel sc = (SocketChannel)  key.channel();
                    try {
                         System.out.println("关闭连接");
                         key.cancel();  
                         sc.socket().close();  
                         sc.close();
                    } catch (IOException e1) {
                         //e1.printStackTrace();
                    }
               }
               
         }
         private void accept(SelectionKey key) {
               try {
                    //1 获取服务通道
                    ServerSocketChannel ssc =  (ServerSocketChannel)  key.channel();
                    //2 执行阻塞方法
                    SocketChannel sc = ssc.accept();
                    //3 设置阻塞模式
                    sc.configureBlocking(false);
                    //4 注册到多路复用器上，并设置读取标识
                    sc.register(this.seletor, SelectionKey.OP_READ);
               } catch (IOException e) {
                    e.printStackTrace();
               }
         }
         
         public static void main(String[] args) {
               
               new Thread(new Server(8765)).start();;
         }
    }

import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.channels.SocketChannel;
    public class Client {
         //需要一个Selector
         public static void main(String[] args) {
               
               //创建连接的地址
               InetSocketAddress address = new  InetSocketAddress("127.0.0.1", 8765);
               
               //声明连接通道
               SocketChannel sc = null;
               
               //建立缓冲区
               ByteBuffer buf = ByteBuffer.allocate(1024);
               
               try {
                    //打开通道
                    sc = SocketChannel.open();
                    //进行连接
                    sc.connect(address);
                    
                    while(true){
                         //定义一个字节数组，然后使用系统录入功能：
                         byte[] bytes = new byte[1024];
                         System.in.read(bytes);
                         
                         //把数据放到缓冲区中
                         buf.put(bytes);
                         //对缓冲区进行复位
                         buf.flip();
                         //写出数据
                         sc.write(buf);
                         //清空缓冲区数据
                         buf.clear();
                         
                         //从服务端读取消息
                    int readLenth = sc.read(buf);
                    //读取模式
                    buf.flip();
                    byte[] bytesrtn = new byte[readLenth];
                    buf.get(bytesrtn);
                    System.out.println(new String(bytesrtn));
                    buf.clear();
                    }
               } catch (IOException e) {
                    e.printStackTrace();
               } finally {
                    if(sc != null){
                         try {
                               sc.close();
                         } catch (IOException e) {
                               e.printStackTrace();
                         }
                    }
               }
         }
    }

# AIO #

AIO编程，在NIO基础之上引入了异步通道的概念，并提供了异步文件和异步套接字通道的实现，从而在真正意义上实现了异步非阻塞，NIO只是非阻塞而并非异步。而AIO它不需要通过多路复用器对注册的通道进行轮询操作即可实现异步读写，从而简化了NIO编程模型。也可称之为NIO2.0，这种模式才真正的属于我们异步非阻塞的模型。

import java.io.UnsupportedEncodingException;
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.channels.AsynchronousSocketChannel;
    import java.util.concurrent.ExecutionException;
    public class Client implements Runnable{
         private AsynchronousSocketChannel asc ;
         
         public Client() throws Exception {
               asc = AsynchronousSocketChannel.open();
         }
         
         public void connect(){
               asc.connect(new InetSocketAddress("127.0.0.1",  8765));
         }
         
         public void write(String request){
               try {
                    asc.write(ByteBuffer.wrap(request.getBytes())).get();//.get()表示异步
                    read();
               } catch (Exception e) {
                    e.printStackTrace();
               }
         }
         private void read() {
               ByteBuffer buf = ByteBuffer.allocate(1024);
               try {
                    asc.read(buf).get();//.get()表示异步
                    buf.flip();
                    byte[] respByte = new byte[buf.remaining()];
                    buf.get(respByte);
                    System.out.println(new  String(respByte,"utf-8").trim());
               } catch (InterruptedException e) {
                    e.printStackTrace();
               } catch (ExecutionException e) {
                    e.printStackTrace();
               } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
               }
         }
         
         @Override
         public void run() {
               while(true){
                    //就是不让它停止
               }
         }
         
         public static void main(String[] args) throws Exception {
               Client c1 = new Client();
               c1.connect();
               
               Client c2 = new Client();
               c2.connect();
               
               Client c3 = new Client();
               c3.connect();
               
               new Thread(c1, "c1").start();
               new Thread(c2, "c2").start();
               new Thread(c3, "c3").start();
               
               Thread.sleep(1000);
               
               c1.write("c1 aaa");
               c2.write("c2 bbbb");
               c3.write("c3 ccccc");
         }
         
    }

import java.net.InetSocketAddress;
    import java.nio.channels.AsynchronousChannelGroup;
    import java.nio.channels.AsynchronousServerSocketChannel;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    public class Server {
         //线程池
         private ExecutorService executorService;
         //线程组
         private AsynchronousChannelGroup threadGroup;
         //服务器通道
         public AsynchronousServerSocketChannel assc;
         
         public Server(int port){
               try {
                    //创建一个缓存池
                    executorService =  Executors.newCachedThreadPool();
                    //创建线程组
                    threadGroup =  AsynchronousChannelGroup.withCachedThreadPool(executorService,  1);
                    //创建服务器通道
                    assc =  AsynchronousServerSocketChannel.open(threadGroup);
                    //进行绑定
                    assc.bind(new InetSocketAddress(port));
                    
                    System.out.println("server start , port : " +  port);
                    //进行阻塞(其实不是阻塞。)
                    assc.accept(this, new  ServerCompletionHandler());
                    //一直阻塞 不让服务器停止
                    Thread.sleep(Integer.MAX_VALUE);
                    
               } catch (Exception e) {
                    e.printStackTrace();
               }
         }
         
         public static void main(String[] args) {
               Server server = new Server(8765);
         }
    }
    
    
    import java.nio.ByteBuffer;
    import java.nio.channels.AsynchronousSocketChannel;
    import java.nio.channels.CompletionHandler;
    import java.util.concurrent.ExecutionException;
    public class ServerCompletionHandler implements  CompletionHandler<AsynchronousSocketChannel, Server> {
         @Override
         public void completed(AsynchronousSocketChannel asc,  Server attachment) {
               //当有下一个客户端接入的时候 直接调用Server的accept方法，这样反复执行下去，保证多个客户端都可以阻塞
               attachment.assc.accept(attachment, this);
               read(asc);
         }
         private void read(final AsynchronousSocketChannel asc) {
               //读取数据
               ByteBuffer buf = ByteBuffer.allocate(1024);
               asc.read(buf, buf, new CompletionHandler<Integer,  ByteBuffer>() {
                    @Override
                    public void completed(Integer resultSize,  ByteBuffer attachment) {
                         //进行读取之后,重置标识位
                         attachment.flip();
                         //获得读取的字节数
                         System.out.println("Server -> " + "收到客户端的数据长度为:" + resultSize);
                         //获取读取的数据
                         String resultData = new  String(attachment.array()).trim();
                         System.out.println("Server -> " + "收到客户端的数据信息为:" + resultData);
                         String response = "服务器响应, 收到了客户端发来的数据: " + resultData;
                         write(asc, response);
                    }
                    @Override
                    public void failed(Throwable exc, ByteBuffer  attachment) {
                         exc.printStackTrace();
                    }
               });
         }
         
         private void write(AsynchronousSocketChannel asc, String  response) {
               try {
                    ByteBuffer buf = ByteBuffer.allocate(1024);
                    buf.put(response.getBytes());
                    buf.flip();
                    asc.write(buf).get();//.get()表示异步
               } catch (InterruptedException e) {
                    e.printStackTrace();
               } catch (ExecutionException e) {
                    e.printStackTrace();
               }
         }
         
         @Override
         public void failed(Throwable exc, Server attachment) {
               exc.printStackTrace();
         }
    }

[20200729095957469.png]: /images/20230208/986a09713bb04cd8a616fbd6da951155.png