关于后端:一文彻底理解BIONIOAIO

22次阅读

共计 23374 个字符，预计需要花费 59 分钟才能阅读完成。

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 多平台公布

正文完
后端
发表至： 后端
2022-07-17
 0
单例模式有几种写法
关于后端:面试题百日百刷kafka篇五
关于后端:3-Caller-服务调用-dapr
关于后端:Spring源码解析-IOC默认标签解析下