美文网首页
一文彻底理解BIO、NIO、AIO

一文彻底理解BIO、NIO、AIO

作者: 醉鱼java | 来源:发表于2022-07-16 22:51 被阅读0次

    IO模型

    BIO

    同步并阻塞,服务器实现模式为一个链接一个线程,即客户端有连接请求时服务器端就需要启动一个线程进行处理,如果这个连接不做任何事情会造成不必要的线程开销

    NIO

    同步非阻塞,服务器实现模式为一个线程处理多个请求,即客户端发送的连接请求都会注册到多路复用器上,多路复用器轮询到连接有I/O请求就进行处理

    AIO(NIO 2)

    异步非阻塞,服务器实现模式为一个有效请求一个线程,客户端的I/O请求都是有OS先完成了再通知服务器应用去启动线程进行处理,一般适用于连接数较多且连接时间较长的应用

    BIO深入剖析

    JAVA BIO 基本介绍

    单客户端对单服务端,发送单条信息

    • Server
    public class BioServer {
        public static void main(String[] args) {
            try {
                ServerSocket socket = new ServerSocket(8888);
                Socket accept = socket.accept();
                InputStream inputStream = accept.getInputStream();
                BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream));
                String msg = "";
                while ((msg = bufferedReader.readLine())!=null){
                    System.out.println("服务端收到消息:"+msg);
                }
            }catch (Exception e){
                e.printStackTrace();
            }
    
    
        }
    }
    
    • Client
    public class BioClient {
        public static void main(String[] args){
            try {
                Socket socket = new Socket("127.0.0.1",8888);
                OutputStream outputStream = socket.getOutputStream();
                PrintStream p = new PrintStream(outputStream);
                p.println("hello server!");
                p.flush();
            }catch (Exception e){
                e.printStackTrace();
            }
    
        }
    }
    

    单客户端对单服务端,发送多条消息

    • Server
    public class BioClient {
        public static void main(String[] args){
            try {
                Socket socket = new Socket("127.0.0.1",8888);
                OutputStream outputStream = socket.getOutputStream();
                Scanner scanner = new Scanner(System.in);
                while (true){
                    System.out.println("请输入:");
                    String next = scanner.next();
                    PrintStream p = new PrintStream(outputStream);
                    p.println(next);
                    p.flush();
                }
            }catch (Exception e){
                e.printStackTrace();
            }
    
        }
    }
    
    
    • Client
    public class BioServer {
        public static void main(String[] args) {
            try {
                ServerSocket socket = new ServerSocket(8888);
                Socket accept = socket.accept();
                InputStream inputStream = accept.getInputStream();
                BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream));
                String msg = "";
                while ((msg = bufferedReader.readLine())!=null){
                    System.out.println("服务端收到消息:"+msg);
                }
            }catch (Exception e){
                e.printStackTrace();
            }
    
    
        }
    }
    

    多客户端对单服务端,发送多条消息

    • Server
    public class BioServer {
        public static void main(String[] args) {
            try {
                ServerSocket socket = new ServerSocket(8888);
                while (true){
                    Socket accept = socket.accept();
                    new ServerSocketThread(accept).start();
                }
            }catch (Exception e){
                e.printStackTrace();
            }
    
    
        }
    }
    
    • Client
    public class BioClient {
        public static void main(String[] args){
            try {
                Socket socket = new Socket("127.0.0.1",8888);
                OutputStream outputStream = socket.getOutputStream();
                Scanner scanner = new Scanner(System.in);
                while (true){
                    System.out.println("请输入:");
                    String next = scanner.next();
                    PrintStream p = new PrintStream(outputStream);
                    p.println(next);
                    p.flush();
                }
            }catch (Exception e){
                e.printStackTrace();
            }
    
        }
    }
    
    • Thread
    public class ServerSocketThread extends Thread{
        private Socket socket;
        public ServerSocketThread(Socket socket){
            this.socket = socket;
        }
        @Override
        public void run() {
            try {
                InputStream inputStream = socket.getInputStream();
                BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream));
                String msg = "";
                while ((msg = bufferedReader.readLine())!=null){
                    System.out.println(Thread.currentThread().getName()+":服务端收到消息:"+msg);
                }
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }
    
    

    小结

    • 每个Socket接收到,都会创建一个线程,线程的竞争,切换上下文影响性能
    • 每个线程都会占用栈空间和CPU资源
    • 并不是每个Socket都进行IO操作,无意义的线程处理
    • 客户端的并发访问增加时,服务器将呈现1:1的线程开销;访问量大,系统将发生线程栈溢出,线程创建失败,最终导致进程当即或者僵死,从而不能对外提供服务

    伪异步I/O编程

    在上面的例子中,客户端的并发访问增加时,系统服务端将呈现1:1的线程开销,访问量越大系统将发生线程栈溢出,线程创建失败,最终导致进程当即或者僵死,从而不能对外提供服务
    
    接下来我们采用一个伪异步的I/O的通信框架,采用线程池和任务队列实现,当客户端接入时,将客户端的Socket封装成一个Task交给后端的线程池进行处理。JDK的线程池维护一个消息队列和N个活跃的线程,对消息队列中Socket任务进行处理,由于线程池可以设置消息队列的大小和最大线程数,因此,它的资源占用是可控的,无论多少个客户端并发访问,都不会导致资源的耗尽和宕机。如下图所示
    

    客户端代码实现

    • Server

      public class Server {
          public static void main(String[] args) {
              try {
                  ServerSocket serverSocket = new ServerSocket(8888);
                  HandlerSocketServerPool pool = new HandlerSocketServerPool(3,10);
                  while (true){
                      Socket accept = serverSocket.accept();
                      Runnable runnable = new ServerRunnableTarget(accept);
                      pool.execute(runnable);
                  }
      
              }catch (Exception e){
                  e.printStackTrace();
              }
          }
      }
      
    • ServerRunnableTarget

      public class ServerRunnableTarget  implements Runnable{
          private Socket socket;
      
      
          public  ServerRunnableTarget(Socket socket){
              this.socket = socket;
          }
          @Override
          public void run() {
              try {
                  InputStream inputStream = socket.getInputStream();
                  BufferedReader br = new BufferedReader(new InputStreamReader(inputStream));
                  String msg ;
                  while ((msg = br.readLine())!=null){
                      System.out.println(Thread.currentThread().getName()+"收到消息:"+msg);
                  }
              }catch (Exception e){
                  e.printStackTrace();
              }
      
          }
      }
      
    • HandlerSocketServerPool

      public class HandlerSocketServerPool {
          // 1.创建一个线程池的成员变量用于存储一个线程池对象
          private ExecutorService executorService;
      
          /**
           * 创建这个类的对象的时候需要初始化线程池对象
           * @param maxThreadNum
           * @param queueSize
           */
          public HandlerSocketServerPool(int maxThreadNum,int queueSize){
              executorService = new ThreadPoolExecutor(3,maxThreadNum,120, TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(queueSize));
          }
      
          /**
           * 提供一个方法来提交任务到线程池的任务队列来暂存,等着线程执行
           * @param runnable
           */
          public void execute(Runnable runnable){
              executorService.execute(runnable);
          }
      }
      
    • Client

      public class Client {
          public static void main(String[] args) {
              try {
                  Socket socket = new Socket("127.0.0.1", 8888);
                  OutputStream outputStream = socket.getOutputStream();
                  Scanner scanner = new Scanner(System.in);
                  while (true) {
                      System.out.println("请输入:");
                      String next = scanner.next();
                      PrintStream p = new PrintStream(outputStream);
                      p.println(next);
                      p.flush();
                  }
              } catch (Exception e) {
                  e.printStackTrace();
              }
      
          }
      }
      
      

    小结

    • 伪异步IO采用线程池的实现,因此避免了为每个请求创建一个独立线程造成线程资源耗尽的问题,但由于底层依然是采用的同步阻塞模型,因此无法从根本上解决问题
    • 如果单个消息处理的缓慢,或者服务器线程池中的全部线程都被阻塞,那么后续Socket的IO消息都将在队列中排队。新的Socket请求都将被拒绝,客户端会发生大量连接超时

    NIO

    基本介绍

    • JAVA NIO(New IO)也有人称之为 Java non-blocking IO 是从Java 1.4版本开始引入的IO API,可以替代标准的JAVA IO API。NIO与原来的IO有同样的作用和目的,但是使用方式完全不同,NIO支持面向缓冲区的,基于通道的IO操作,NIO将以更加高效的方式进行文件的读写操作。NIO可以理解为非阻塞IO,传统的IO的read和write只能阻塞执行,线程在读写IO期间不能干其它事情,比如调用socket.read()时,如果服务器一直没有数据传输过来,线程就一直阻塞, 而NIO中可以配置socket为非阻塞模式
    • NIO 相关类都放在java.nio包及子包下,并且对原java.io包中的很多类进行改写
    • NIO 有三大核心部分,Channel(通道),Buffer(缓冲区),Selector(选择器)
    • Java NIO的非阻塞模式,使一个线程从某通道发送请求或者读取数据,但是它仅能得到目前可用的数据,如果目前没有数据可用时,就什么都不会获取,而不是保持线程阻塞,所以直至数据变得可以读取之前,该线程可以继续做其他的事情,非阻塞写也是如此,一个线程请求斜日一些数据到某通道,但不需要等待它完全写入,这个线程同时可以去做别的事情
    • 通俗理解,NIO是可以做到用一个线程来处理多个操作的,假设有1000个请求过来,根据实际情况,可以分配20或者80个线程来处理,不像之前的阻塞IO那样,非得分配1000个

    NIO和BIO的比较

    • BIO以流的方式处理数据,而NIO以块的方式处理数据,块IO的效率比流IO高很多
    • BIO是阻塞的,NIO是非阻塞的
    • BIO基于字节流和字符流进行操作,而NIO基于Channle(通道)和Buffer(缓冲区)进行操作,数据总是从通道读取到缓冲区中,或者从缓冲区写入到通道中,Selecter(选择器)用于监听多个通道的事件(比如: 连接请求,数据到达),因此使用单个线程就可以监听多个客户端通道
    NIO BIO
    面向缓冲区(Buffer) 面向流(Stream)
    非阻塞(Non Blocking IO) 阻塞IO(Blocking IO)
    选择器(Selectors)

    NIO 三大核心

    NIO有三大核心部分:Channel(通道),Buffer(缓冲区),Selector(选择器)

    Buffer 缓冲区

    缓冲区本质上是一块可以写入数据,然后可以从中读取数据的内存,这块内存被包装成NIO Buffer对象,并提供了一组方法,用来方便的访问该块内存。相比较直接对数组的操作,Buffer API更加容易操作和管理

    Channel 通道

    Java NIO的通道类似流,但又有些不同,既可以从通道中读取数据,又可以写数据到通道。但流的(input,output)读写通常是单向的。通道可以非阻塞读取和写入通道,通道可以支持读取或者写入缓冲区,也支持异步的读写

    Selector 选择器

    Selector是一个Java NIO的组件,能够检测出一个或者多个的NIO通道,并确定哪些通道已经准备好进行读取或者写入,这样的话,一个单独的线程就可以管理多个Channel,从而管理多个网络连接,提高效率

    • 每个Channel 都会对应一个Buffer
    • 一个线程对应Selector,一个Selector对应多个Channel连接
    • 程序切换到哪个Channel是由事件决定的
    • Selector会根据不同的事件,在各个通道上切换
    • Buffer就是一个内存块,底层是一个数组
    • 数据的读取写入是可以通过Buffer完成的,BIO中要么是输入流,或者是输出流,不能双向,但是NIO的Buffer是可以读也可以写
    • Java NIO系统的核心在于:通道Channel和缓冲区Buffer。通道表示打开到IO设备(例如:文件、套接字)的连接,若需要使用NIO系统,需要获取用于连接IO设备的通道以及用于容纳数据的缓冲区,然后操作缓冲区对数据进行处理。一句话就是Channel负责传输,Buffer负责存取数据

    NIO核心一:缓冲区Buffer

    缓冲区Buffer

    一个用于特定基本数据类型的容器,由java.nio包定义的,所有缓冲区都是Buffer抽象类的子类。Java NIO中的Buffer主要用于与NIO通道的交互,数据是从通道读入缓冲区,从缓冲区写入通道中的

    Buffer类及其子类

    Buffer就像一个数组,可以保存多个相同类型的数据。根据数据类型的不同,有以下Buffer常用子类

    • ByteBuffer
    • CharBuffer
    • ShortBuffer
    • IntBuffer
    • LongBuffer
    • FloatBuffer
    • DoubleBuffer

    上述Buffer类都是采用的相似的方法进行管理数据,只是各自管理的数据类型不同而已,他们都是通过如下方法获取一个Buffer对象

    static xxxBuffer allocate(int capacity);//创建一个容量为capacity的xxxBuffer对象
    

    缓冲区的基本属性

    Buffer中的重要概念

    • 容量(capacity):作为一个内存块,Buffer具有一定的固定大小,也称为容量,缓冲区容量不能为负,并且创建后不能更改

    • 限制(limit):表示缓冲区中可以操作数据的大小(limit后不能进行数据的读写)。缓冲区限制不能为负,并且不能大于其容量。写入模式,限制等于buffer的容量,读取模式下,limit等于写入的数据量

    • 位置(position):下一个要读取或写入的数据的索引。缓冲区的位置不能为负,并且不能大于其限制

    • 标记(mark)重置(reset):标记是一个索引,通过Buffer中的mark()方法指定Buffer中一个特定的position,之后可以通过调用reset方法恢复到这个position

      标记、位置、限制、容量遵守一下不等式:0<=mark<=position<=limit<=capacity

    • 图示如下:

      1、通过allocate(分配容量为10的缓冲区)

    2、调用put方法写入5个数据到缓冲区

    3、通过flip切换读数据模式

    Buffer常见方法

    Buffer clear();// 清空缓冲区并返回对缓冲区的引用
    Buffer flip();// 为将缓冲区的界限设置为当前位置,并将当前位置重制为0
    int capacity();// 返回Buffer的capacity大小
    boolean hasRemaining();// 判断缓冲区是否还有元素
    int limit();// 返回Buffer的界限(limit)的位置
    Buffer limit(int n);// 将设置缓冲区界限为n,并返回一个具有新limit的缓冲区对象
    Buffer mark();//对缓冲区设置标记
    int position();// 返回缓冲区的当前位置position
    Buffer position(int n);// 将设置缓冲区的当前位置为n,并返回修改后的Buffer对象
    int remaining();//返回position和limit之间的元素个数
    Buffer reset();//将位置position转到以前设置的mark所有的位置
    Buffer rewind();// 将位置设为0,取消设置的mark
    

    缓冲区的数据操作

    // Buffer 所有的子类都提供了两个用于数据操作的方法;get()和put()
    // 取数据
    get();// 读取单个字节
    get(byte[] dst);// 批量读取多个字节到dst中
    get(int index);// 读取指定索引位置的字节(不会移动position)
    // 存数据
    put();// 将给定的单个字节写入缓冲区的当前位置
    put(byte[] src);// 将src中的字节写入缓冲区的当前位置
    put(int index,byte b);// 将指定字节写入缓冲区的索引位置(不会移动position)
    
    

    使用Buffer读写数据一般遵循以下四个步骤

    • 写入数据到Buffer
    • 调用flip()方法,转换为读模式
    • 从Buffer中读取数据
    • 调用buffer.clear()方法或者buffer.compact()方法清除缓冲区

    代码测试

    public class BufferTest {
        public static void main(String[] args) {
            ByteBuffer buffer = ByteBuffer.allocate(10);
            System.out.println(buffer.position());//0
            System.out.println(buffer.limit());//10
            System.out.println(buffer.capacity());//10
            System.out.println("=============");
            buffer.put("zuiyu".getBytes());
            System.out.println(buffer.position());//5
            System.out.println(buffer.limit());//10
            System.out.println(buffer.capacity());//10
            System.out.println("=============");
    
            buffer.flip();
            System.out.println(buffer.position());//0
            System.out.println(buffer.limit());//5
            System.out.println(buffer.capacity());//10
            System.out.println("=============");
    
            char ch = (char)buffer.get();
            System.out.println(ch);
            System.out.println(buffer.position());//1
            System.out.println(buffer.limit());//5
            System.out.println(buffer.capacity());//10
            System.out.println("=============");
    
            buffer.clear();
            System.out.println(buffer.position());//0
            System.out.println(buffer.limit());//10
            System.out.println(buffer.capacity());//10
            System.out.println((char)buffer.get());//z
            System.out.println("=============");
    
            ByteBuffer buf = ByteBuffer.allocate(10);
            buf.put("zuiyu".getBytes());
            buf.flip();
            byte[] b = new byte[2];
            buf.get(b);
            String rs = new String(b);
            System.out.println(rs);
            System.out.println(buffer.position());//1
            System.out.println(buffer.limit());//10
            System.out.println(buffer.capacity());//10
            System.out.println("=============");
    
            buf.mark();
            byte[] b2 =new byte[3];
            buf.get(b2);
            System.out.println(new String(b2));
            System.out.println(buffer.position());//1
            System.out.println(buffer.limit());//10
            System.out.println(buffer.capacity());//10
            System.out.println("=============");
    
            buf.reset();
            if (buf.hasRemaining()){
                System.out.println(buf.remaining());//3
            }
        }
    }
    

    直接与非直接缓冲区

    什么是直接内存与非直接内存

    根据官方文档的描述:

    byte buffer可以是两种类型,一种是基于直接内存(也就是非堆内存);另一种是非直接内存(也就是堆内存)。对于直接内存来说,JVM将会在IO操作上具有更高的性能,因为它直接作用于本地系统的IO操作。而非直接内存,也就是堆内存中的数据如果要做IO操作,会先从本进程内存复制到直接内存,在利用本地IO处理。

    从数据流的角度,非直接内存是下面这样的作用连

    本地IO => 直接内存 => 非直接内存 => 直接内存 => 本地IO
    

    直接内存是

    本地IO => 直接内存 => 本地IO
    

    通过上面对比很明显可以看出,在做IO处理时,比如网络发送大量数据时,直接内存会具有更高的效率,直接内存使用allocateDirect创建,但是它比申请普通的堆内存需要耗费更高的性能。不过,这部分的数据是在JVM之外的,因此它不会占用应用的内存。所以呢,当你有很大的数据要缓存,并且它的生命周期又很长,那么就比较适合使用直接内存。只是一般来说,如果不是能带来明显的性能提升,还是推荐使用堆内存。字节缓冲区是直接缓冲区还是非直接缓冲区可以通过调用其isDirect()方法来确定

    NIO核心二:通道Channel

    通道Channel

    通道Channel:由java.nio.channels包定义的,Channel表示IO源与目标打开的连接。Channel类似于传统的流,只不过Channel本身不能直接访问数据,Channel只能于Buffer进行交互

    1、NIO的通道类似与流,但是还有区别的:

    • 通道可以同时进行读写,而流只能读或者写
    • 通道可以实现异步读写数据
    • 通道可以从缓冲区读数据们也可以写数据到缓冲

    2、BIO中的stream是单向的,例如FileInputStream对象只能进行读取数据的操作,而NIO的通道Channel是双向的,可以读也可以写

    3、Channel在NIO中是一个接口

    public interface Channel extends Closeable{}
    

    常用的Channel实现类

    • FileChannel:用于读取、写入、映射和操作文件的通道
    • DatagramChannel:通过UDP读写网络中的数据通道
    • SocketChannel:通过TCP读写网络中的数据
    • ServerSocketChannel:可以监听新进来的TCP连接,对每一个新进来的连接都会创建一个SocketChannel。【ServerSocketChannel类似ServerSocket,SocketChannel类似Socket】

    FileChannel 类

    获取通道的一种方式是对支持通道的对象调用getChannel()方法,支持通道的类如下:

    • FileInputStream

    • FileOutputStream

    • RandomAccessFile

    • DatagramSocket

    • Socket

    • ServerSocket

      获取通道的其他方式是使用Files类的静态方法newByteChannel()获取字节通道。或者通过通道的静态方法open()打开并返回指定通道

    FileChannel的常用方法

    int read(ByteBuffer dst);// 从Channel中读取数据到ByteBuffer
    long read(ByteBuffer[] dsts);//将Channel中的数据分散到ByteBuffer[]中
    int write(ByteBuffer src);// 将ByteBuffer中的数据写入到Channel
    long write(ByteBuffer[] srcs);// 将ByteBuffer[]中的数据聚集到Channel
    long position();//返回此通道的文件位置
    FileChannel position(long p);// 设置此通道的文件位置
    long size();// 返回此通道的文件的当前大小
    FileChannel truncate(long s);// 将此通道的文件截取为指定大小s
    void force(boolean metaData);//强制将所有对此通道的文件更新写入到存储设备中
    

    本地文件写入数据

    @Test
        public void write(){
            try {
                FileOutputStream fos = new FileOutputStream("test.txt");
                FileChannel channel = fos.getChannel();
                ByteBuffer buffer = ByteBuffer.allocate(1024);
                buffer.put("hello zuiyu!".getBytes());
                buffer.flip();
                channel.write(buffer);
            }catch (Exception e){
                e.printStackTrace();
            }
    
        }
    

    本地文件读取数据

        @Test
        public void read() throws Exception{
            FileInputStream is = new FileInputStream("test.txt");
            FileChannel channel = is.getChannel();
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            channel.read(buffer);
            buffer.flip();
            String rs = new String(buffer.array(),0,buffer.remaining());
            System.out.println(rs);
        }
    

    使用Buffer完成文件的复制

     @Test
        public void copy()throws Exception{
            File srcFile  =new File("test.txt");
            File destFile  =new File("test_copy.txt");
    
            FileInputStream fis = new FileInputStream(srcFile);
            FileOutputStream fos = new FileOutputStream(destFile);
            FileChannel fisChannel = fis.getChannel();
            FileChannel fosChannel = fos.getChannel();
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            while (true){
                // 必须清空缓冲区,然后在写入数据到缓冲区
                buffer.clear();
                // 开始读取一次数据
                int flag = fisChannel.read(buffer);
                if (flag == -1){
                    break;
                }
                // 已经读取数据,切换缓冲区模式为可读模式
                buffer.flip();
                fosChannel.write(buffer);
            }
            fisChannel.close();
            fosChannel.close();
            System.out.println("文件复制完成!");
        }
    

    基于分散(scatter)和聚集(gather)

    分散读取(scatter): 是指把Channel通道的数据读入到多个缓冲区中去

    聚集写入(gather): 是指将多个buffer中的数据聚集到Channel中

     @Test
        public void buffersTest() throws Exception{
            // 1、字节输入管道
            FileInputStream fis = new FileInputStream("test.txt");
            FileChannel fisChannel = fis.getChannel();
            // 2、字节输出管道
            FileOutputStream fos = new FileOutputStream("test02.txt");
            FileChannel fosChannel = fos.getChannel();
            // 3、定义多个缓冲区做数据分散
            ByteBuffer buffer1 = ByteBuffer.allocate(4);
            ByteBuffer buffer2 = ByteBuffer.allocate(1024);
            ByteBuffer[] buffers  = {buffer1,buffer2};
            // 4、从通道中读取数据分散到各个缓冲区
            fisChannel.read(buffers);
            for (ByteBuffer buffer : buffers) {
                // 切换读数据模式
                buffer.flip();
                System.out.println(new String(buffer.array(),0,buffer.remaining()));
            }
            // 聚集写入到通道
            fosChannel.write(buffers);
            fisChannel.close();
            fosChannel.close();
            System.out.println("文件复制完成");
        }
    

    transferFrom

    从目标通道中去复制原通道数据

    @Test
        public void transferFormTest() throws Exception{
            // transferFrom 从目标通道中去复制原通道数据
            // 1、字节输入管道
            FileInputStream fis = new FileInputStream("test.txt");
            FileChannel fisChannel = fis.getChannel();
            // 2、字节输出管道
            FileOutputStream fos = new FileOutputStream("test03.txt");
            FileChannel fosChannel = fos.getChannel();
            fosChannel.transferFrom(fisChannel,fisChannel.position(),fisChannel.size());
            fisChannel.close();
            fosChannel.close();
            System.out.println("文件复制完成:transferFrom");
        }
    

    transferTo

    从源通道数据复制到目标通道

     @Test
        public void transferToTest() throws Exception{
            // transferTo 从原数据通道复制到目标通道
            // 1、字节输入管道
            FileInputStream fis = new FileInputStream("test.txt");
            FileChannel fisChannel = fis.getChannel();
            // 2、字节输出管道
            FileOutputStream fos = new FileOutputStream("test04.txt");
            FileChannel fosChannel = fos.getChannel();
            fisChannel.transferTo(fisChannel.position(),fisChannel.size(),fosChannel);
            fisChannel.close();
            fosChannel.close();
            System.out.println("文件复制完成:transferTo");
        }
    

    NIO核心三:选择器Selector

    选择器Selector

    选择器(Selector)是SelectableChannel对象的多路复用器,Selector可以同时监控多个SelectableChannel的IO状况,利用Selector可使一个单独的线程管理多个Channel。Selector是非阻塞的IO的核心

    • Java的NIO,用非阻塞的IO方式,可以用一个线程,处理多个的客户端连接,就会使用到Selector(选择器)
    • Selector能够检测多个注册的通道上是否有事件发生(注意:多个Channel以事件的方式可以注册到同一个Selector),如果有事件发生,便获取事件然后针对每个事件进行相应的处理,这样就可以只用一个单线程去管理多个通道,也就是管理多个连接和请求
    • 只有在连接/通道 真正有读写事件发生时,才会进行读写,就大大的减少了系统开销,并且不必为每个连接都创建一个线程,不用去维护多个线程
    • 避免了多线程之间的上下文切换导致的开销

    选择器(Selector)的应用

    创建Selector:通过调用Selector.open()方法创建一个Selector

    Selector selector = Selector.open();
    

    向选择器注册通道:SekectableChannel.register(Selector sel,int ops)

    // 获取通道
    ServerSocketChannel ssChannel = ServerSocketChannel.open();
    // 切换非阻塞模式
    ssChannel.configureBlocking(false);
    // 绑定连接
    ssChannel.bind(new InetSocketAddress(8888));
    // 获取选择器
    Selector selector = Selector.open();
    // 将通道注册到选择器上,并且指定监听接收事件
    ssChannel.register(selector,SelectionKey.OP_ACCRPT);
    

    当调用register(Selector sel,int ops)将通道注册到选择器时,选择器对通道的监听事件,需要通过第二个参数ops指定,可以监听到事件类型(可以使用SelectionKey的四个常量表示)

    • 读:Selection.OP_READ(1)

    • 写:SelectionKey.OP_WRITE(4)

    • 连接:SelectionKey.OP_CONNECT(8)

    • 接收:SelectionKey.OP_ACCEPT(16)

    • 若注册时不止监听一个事件,则可以使用"位或"操作符连接

      int interestSet = SelectionKey.OP_READ|SelectionKey.OP_WRITE
      

    NIO 非阻塞式网络通信原理分析

    Selector示意图和特点说明

    Seector可以实现:一个IO线程可以并发处理N个客户端连接和读写操作,这从根本上解决了传统同步阻塞IO一连接一线程模型,架构的性能、弹性伸缩能力和可靠性都得到了极大的提升

    服务端流程

    • 当客户端连接服务器端时,服务端会通过ServerSocketChannel得到SocketChannel

      // 获取通道
      ServerSocketChannel ssChannel = ServerSocketChannel.open();
      
    • 切换非阻塞模式

      // 切换非阻塞模式
      ssChannel.configureBlocking(false);
      
    • 绑定连接

      // 绑定连接
      ssChannel.bind(new InetSocketAddress(8888));
      
    • 获取选择器

      // 获取选择器
      Selector selector = Selector.open();
      
    • 将通道注册到选择器上,并且指定监听接收事件

      // 将通道注册到选择器上,并且指定监听接收事件
      ssChannel.register(selector,SelectionKey.OP_ACCRPT);
      
    • 轮询式的获取选择器上已经准备就绪的事件

       while (iterator.hasNext()){
                      // 拿到当前事件
                      SelectionKey sk = iterator.next();
                      // 判断当前事件是什么类型
                      if (sk.isAcceptable()){
                          // 获取当前接入的客户端通道
                          SocketChannel socketChannel = serverSocketChannel.accept();
                          // 配置非阻塞模式
                          socketChannel.configureBlocking(false);
                          // 将本客户端通道注册到选择器
                          socketChannel.register(selector,SelectionKey.OP_READ);
                      }else if (sk.isReadable()){
                          // 获取当前选择器上 《读》 就绪事件
                          SocketChannel channel =(SocketChannel) sk.channel();
                          // 读取数据
                          ByteBuffer buffer = ByteBuffer.allocate(1024);
                          int len =0;
                          while ((len = channel.read(buffer))>0){
                              buffer.flip();
                              System.out.println(new String(buffer.array(),0,len));
                              // 清除之前的数据
                              buffer.clear();
                          }
                      }
                      iterator.remove();
                  }
      

    客户端流程

    • 获取通道

              SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress(8888));
      
      
    • 切换非阻塞模式

              socketChannel.configureBlocking(false);
      
    • 分配指定大小的缓冲区

      ByteBuffer buffer = ByteBuffer.allocate(1024);
      
    • 发送数据给服务端

       Scanner sc = new Scanner(System.in);
              while (true){
                  System.out.println("请说:");
                  String msg = sc.nextLine();
                  buffer.put(("zuiyu:"+msg).getBytes());
                  buffer.flip();
                  socketChannel.write(buffer);
                  buffer.clear();
              }
      

    服务端代码实现

    public class Server {
        public static void main(String[] args) throws  Exception{
            // 获取通道
            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
            // 配置非阻塞模式
            serverSocketChannel.configureBlocking(false);
            // 绑定连接端口
            serverSocketChannel.bind(new InetSocketAddress(8888));
            // 获取选择器
            Selector selector = Selector.open();
            // 将通道注册到选择器上,并且指定监听接收事件
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            // 使用Selector选择器轮询已经就绪的事件
            while (selector.select()>0){
                System.out.println("开始轮询获取事件");
                // 获取选择器中所有已经就绪好的事件
                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                // 开始遍历事件
                while (iterator.hasNext()){
                    // 拿到当前事件
                    SelectionKey sk = iterator.next();
                    // 判断当前事件是什么类型
                    if (sk.isAcceptable()){
                        // 获取当前接入的客户端通道
                        SocketChannel socketChannel = serverSocketChannel.accept();
                        // 配置非阻塞模式
                        socketChannel.configureBlocking(false);
                        // 将本客户端通道注册到选择器
                        socketChannel.register(selector,SelectionKey.OP_READ);
                    }else if (sk.isReadable()){
                        // 获取当前选择器上 《读》 就绪事件
                        SocketChannel channel =(SocketChannel) sk.channel();
                        // 读取数据
                        ByteBuffer buffer = ByteBuffer.allocate(1024);
                        int len =0;
                        while ((len = channel.read(buffer))>0){
                            buffer.flip();
                            System.out.println(new String(buffer.array(),0,len));
                            // 清除之前的数据
                            buffer.clear();
                        }
                    }
                    iterator.remove();
                }
            }
    
        }
    }
    

    客户端代码实现

    public class Client {
        public static void main(String[] args) throws Exception{
            // 获取通道
            SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress(8888));
            // 配置非阻塞模式
            socketChannel.configureBlocking(false);
            // 分配指定缓冲区大小
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            // 发送数据到服务端
            Scanner sc = new Scanner(System.in);
            while (true){
                System.out.println("请说:");
                String msg = sc.nextLine();
                buffer.put(("zuiyu:"+msg).getBytes());
                buffer.flip();
                socketChannel.write(buffer);
                buffer.clear();
            }
        }
    }
    
    

    网络编程应用实例-群聊系统

    目标

    需求:近一步理解NIO非阻塞网络编程机制,实现多人群聊

    • 编写一个NIO群聊系统,实现客户端与客户端的通信需求(非阻塞)
    • 服务器端:可以检测用户上线、下线、并实现消息转发功能
    • 客户端:通过channel可以无阻塞的发送消息到系统中的其他客户端用户,同时可以接收其他客户端用户通过服务端转发来的消息

    服务端代码实现

    public class Server {
        // 定义成员属性,选择器,服务端通道,端口
        private Selector selector;
        private ServerSocketChannel serverSocketChannel;
        private static final int PORT = 8888;
    
        public Server(){
            try {
                // 创建选择器对象
                selector = Selector.open();
                // 获取通道
                serverSocketChannel = ServerSocketChannel.open();
                // 绑定客户端连接的端口
                serverSocketChannel.bind(new InetSocketAddress(PORT));
                // 设置非阻塞模式
                serverSocketChannel.configureBlocking(false);
                // 把通道注册到选择器上去,并且开始指定接收连接事件
                serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
    
            }catch (Exception e){
                e.printStackTrace();
            }
    
        }
    
        public static void main(String[] args) {
            // 创建服务端对象
            Server server = new Server();
            // 开始监听客户端的各种消息事件,连接、群聊消息、离线消息
            server.listen();
        }
    
        /**
         * 开始监听
         */
        private void listen() {
            try {
                while (selector.select()>0){
                    // 获取选择器中所有注册通道的就绪事件
                    Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                    // 遍历事件
                    while (iterator.hasNext()) {
                        SelectionKey sk = iterator.next();
                        // 判断事件的类型
                        if (sk.isAcceptable()){
                            // 获取当前客户端通道
                            SocketChannel socketChannel = serverSocketChannel.accept();
                            socketChannel.configureBlocking(false);
                            socketChannel.register(selector,SelectionKey.OP_READ);
    
                        } else if (sk.isReadable()) {
                            readClientData(sk);
                        }
                        iterator.remove();
                    }
    
                }
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    
        private void readClientData(SelectionKey sk) {
            SocketChannel socketChannel = null;
            try {
                socketChannel = (SocketChannel) sk.channel();
                // 创建缓冲区对象开始接收客户端通道的数据
                ByteBuffer buffer = ByteBuffer.allocate(1024);
                int read = socketChannel.read(buffer);
                if (read>0){
                    buffer.flip();
                    String msg = new String(buffer.array(),0,buffer.remaining());
                    System.out.println("接收到客户端消息:"+msg);
                    // 把这个消息推送到全部客户端接收
                    sendMsgToAllClient(msg,socketChannel);
                }
            }catch (Exception e){
                try {
                    System.out.println("有人离线:"+socketChannel.getRemoteAddress());
                    // 当前客户端离线
                    sk.cancel();// 取消注册
                    socketChannel.close();
                }catch (Exception e1){
                    e.printStackTrace();
                }
    
            }
        }
    
        /**
         * 把当前客户端的消息数据推送到当前全部在线注册的channel
         * @param msg
         * @param socketChannel
         */
        private void sendMsgToAllClient(String msg, SocketChannel socketChannel) throws Exception{
            System.out.println("服务端开始转发消息:"+Thread.currentThread().getName());
            for (SelectionKey key : selector.keys()) {
                SelectableChannel channel = key.channel();
                if (channel instanceof SocketChannel && !(channel==socketChannel)){
                    ByteBuffer byteBuffer = ByteBuffer.wrap(msg.getBytes());
                    ((SocketChannel)channel).write(byteBuffer);
                }
            }
        }
    
    }
    
    

    客户端代码实现

    public class Client {
        // 定义客户端相关参数
        private Selector selector;
        private static int PORT = 8888;
        private SocketChannel socketChannel;
    
        // 初始化客户端
        public Client(){
            try {
                // 创建选择器
                selector = Selector.open();
                // 连接服务端
                socketChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1",PORT));
                socketChannel.configureBlocking(false);
                socketChannel.register(selector, SelectionKey.OP_READ);
                System.out.println("客户端准备完成:"+Thread.currentThread().getName());
            }catch (Exception e){
                e.printStackTrace();
            }
    
    
        }
        public static void main(String[] args) {
            Client client = new Client();
            // 设置一个线程专门负责监听服务端发送过来的消息事件
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        client.readInfo();
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                }
            }).start();
            Scanner sc = new Scanner(System.in);
            while (sc.hasNextLine()){
                String s = sc.nextLine();
                client.sendToServer(s);
            }
        }
    
        private void sendToServer(String s){
            try {
                socketChannel.write(ByteBuffer.wrap(("zuiyu say:"+socketChannel.getLocalAddress()+":"+s).getBytes()));
            }catch (Exception e){
                e.printStackTrace();
            }
    
        }
    
        private void readInfo() throws Exception{
            if (selector.select()>0){
                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while (iterator.hasNext()) {
                    SelectionKey selectionKey = iterator.next();
                    if (selectionKey.isReadable()){
                        SocketChannel channel = (SocketChannel) selectionKey.channel();
                        ByteBuffer buffer = ByteBuffer.allocate(1024);
                        channel.read(buffer);
                        System.out.println("客户端消息:"+new String(buffer.array()).trim());
                    }
                    iterator.remove();
                }
            }
        }
    }
    

    AIO

    • JAVA AIO(NIO 2):异步非阻塞,服务器实现模式为一个有效请求一个线程,客户端的IO请求都是OS先完成了在通知服务器应用去启动线程进行处理

      BIO NIO AIO
      Socket SocketChannel AsynchronousSocketChannel
      ServerSocket ServerSockerChannel AsynchronousServerSocketChannel

      与NIO不同,当进行读写操作时,只需直接调用API的read或者write方法即可,这两种方法均为异步的,对于读操作而言,当有流可读取时,操作系统会将可读取的流传入read方法的缓冲区,对于写操作而言,当操作系统将write方法传递的流写入完毕后,操作系统主动通知应用程序

      即可以理解为read/write方法都是异步的,完成后会主动调用回调函数,在JDK1.7中,这部分内容被称为NIO2,主要是在java.nio.channels包下增加了下面异步通道

      AsynchronousSocketChannel
      AsynchronousServerSocketChannel
      AsynchronousFileChannel
      

    总结

    BIO、NIO、AIO

    • Java BIO:同步并阻塞,服务器实现模式为一个链接一个线程,即客户端有连接请求时服务器端就需要启动一个线程进行处理,如果这个链接不做任何事情会造成不必要的线程开销,当然可以通过线程池机制改善
    • Java NIO:同步非阻塞,服务器实现模式为一个请求一个线程,即客户端发送的链接请求都会注册到多路复用选择器上,多路复用器轮询到链接IO请求时才启动一个线程进行处理
    • Java AIO(NIO 2):异步非阻塞,服务器实现模式为一个有效请求一个线程,客户端的IO请求都是有OS先完成了在通知服务器应用去启动线程进行处理

    BIO、NIO、AIO 适用场景分析

    1、BIO方式适用于连接数目比较小且固定的架构,这种方式对服务器资源要求比较高,并发局限于应用中,JDK1.4以前的唯一选择,但程序简单易理解

    2、NIO方式适于连接数目多且连接比较短(轻操作)的架构,比如聊天服务器,弹幕系统,服务器间通讯等,编程比较复杂,JDK1.4开始支持

    3、AIO方式使用与连接数目多且连接比较长(重操作)的架构,比较相册服务器,充分调用OS参与并发操作,编程比较复杂,JDK7开始支持

    本文由mdnice多平台发布

    相关文章

      网友评论

          本文标题:一文彻底理解BIO、NIO、AIO

          本文链接:https://www.haomeiwen.com/subject/aocrirtx.html