Unix I/O 模型

有了上面介绍的铺垫，我们可以就可以来聊聊unix 的I/O模型了

在Linux(UNIX)操作系统中，共有五种IO模型，分别是：阻塞IO模型、非阻塞IO模型、IO复用模型、信号驱动IO模型以及异步IO模型。

阻塞I/O模型

套接字的默认状态是阻塞的，这就意味着当发出一个不能立即完成的套接字调用时，其进程被投入睡眠，等待相应操作完成才会进行唤醒。

如上图所示，应用程序通过系统调用（将recvfrom视为系统调用）recvfrom，其系统调用知道数据报大大且被复制到应用进程的缓冲区中或发生错误才返回。最常见的错误是系统调用被信号中断，我们说进程在从调用recvfrom开始到它返回的整段时间内是被阻塞的（进程无法做其他事情）。recvfrom成功返回后，应用程序进程开始处理数据报。

非阻塞I/O模型

进程把一个套接字设置成非阻塞再通知内核：当前所请求的I/O操作无需将本进程进行阻塞，而是返回一个错误。如上图前三次的调用recvfrom时没有数据可返回，因此内核转而立即返回一个EWOULDBLOCK错误。第四次调用recvfrom时已经有一个数据报准备好，它被复制到应用进程缓冲区，雨水recvfrom成功返回。接着进程继续处理数据。

当一个应用进程像这样对一个非阻寨描述符循环调用recvfrom时，我们称之为轮询（polling）。应用进程持续轮询内核，以查看某个操作是否就绪。这么做往往耗费大量CPU时间，不过这种模型偶尔也会遇到，通常是在专门提供菜一种功能的系统中才有。

I/O复用模型

Linux 2.6之前是select、poll，2.6之后是epoll，Windows是IOCP

I/O多路复用（I/O multiplexing）模型下可以调用selec、poll或epoll，阻塞在这两个系统调用中的某一个上，而不是阻塞真正的I/O系统调用上。我们阻塞在select调用，等待数据报套接字变为可读。当select返回套接字可读这一条件时，我们调用recvfrom把所读数据报复制到应用进程缓冲区。

I/O复用并不显得有什么优势，事实上由于使用select需要两个而不是单个系统调用，I/O复用还稍有劣势。实际上使用select的优势在于我们可以等待多个描述符就绪。

与I/O多路复用密切相关的另一种I/O模型是在多线程中使用阻塞式I/O，这种模型与上述模型极为相似，但它没有使用sclect阻寒在多个文件描述符上，而是使用多个线程（每个文件描迷符一个线程），这样每个线程都可以自由地调用诸如recvfrom之类的阻寒式I/O系统调用

Linux IO复用的底层实现参考：https://mikeygithub.github.io/2020/12/03/yuque/xzqx4t/

信号驱动I/O模型

我们也可以用信号，让内核在描述符就绪时发送SIGIO信号通知我们。我们称这种模型为信号驱动式I/O（signal-driven I/O），IO多路复用常用的方法有：select、poll以及epoll三种。IO多路复用的好处就在于单个process就可以同时处理多个网络连接的IO。

我们首先开启套接字的信号驱动式I/O功能，并通过sigaction系统调用安装一个型号处理函数。该系统调用将立即返回，我们的进程继续工作，也就是说它没有被阻塞。当数据准备好取读时，内核就为该进程产生一个SIGIO信号。我们随后既可以在型号处理函数中调用recvfrom读取数据报，并通知主循环数据已准备好带出来，也可以理解通知主循环，让它读取数据报。

无论如何处理SIGIO信号，这种模型的优势在于等待数据报到达期间进程不会被阻塞。主循环可以继续执行，只有等待来自信号处理函数的通知；既可以是数据已准备好被处理，也可以是数据报已准备好被读取。

异步I/O模型

异步I/O（asynchronous I/O）由POSIX规范定义。

这些函数的工作机制是：告知内核启动某个操作，并让内核在整个操作（包括将数据冲内核复制到我们自己的缓冲区）完成后通知我们。

这种模型和信号驱动式模型的区别在于：信号驱动式I/O是由内核通知我们核实可以启动一个I/O操作，而异步I/O模型是由内核通知我们I/O操作何时完成。

我们调用aio_read 函数（POSIX异步I/O函数以aio_或lio_开头），给内核传递描述符、缓冲区指针、缓冲区大小（与read相同的三个参数）和文件偏移（与lseek类似），并高手内核当整个操作完成时如何通知我们。该系统调用立即返回，并且在等待IO完成期间，我们的进程不被阻塞。

本例子中我们假设要求内核在操作完成时产生某个信号。该信号直到数据己复制到应用进程缓冲区才产生，这一点不同于信号驱动式VO模型，本书编写至此的时候，支持POSIX异步I0模型的系统仍较罕见。我们不能确定这样的系统是否支持套接宇上的这种模型。这儿我们只是用它作为一个与信号驱动式I/O模型相比照的例子

各种I/O模型的比较

根据上述定义，我们的前4种模型。阻塞式I/O模型、非阻塞式I/O模型、I/O复用模型和信号驱动式I/O模型都是同步I/O模型，因为其中真正的I/O操作 (recvfrom）将阻塞进程。只有异步I/O模型与POSIX定义的异步I/O相匹配。

I/O 模型概念

在了解I/O模型我们先要了解一下几个关键词，阻塞/非阻塞、同步/异步

同步/异步

同步（synchronous communication）和异步（asynchronous communication）关注的是两端（调用端和接收端）之间的消息通信机制。（描述两个模块之间的关系）

同步

所谓同步，就是在发出一个调用时，在没有得到结果之前，该调用就不返回。但是一旦调用返回，就得到返回值了。换句话说，就是由调用者主动等待这个调用的结果。

异步

异步则与同步是相反，调用在发出之后，这个调用就直接返回了，所以没有返回结果。换句话说，当一个异步过程调用发出后，调用者不会立刻得到结果。而是在调用发出后，被调用者通过状态、通知来通知调用者，或通过回调函数处理这个调用。

阻塞/非阻塞

阻塞（Blocking）和非阻塞（Non Blocking）指的是调用者（程序）在等待返回结果（或输入）时的状态。（描述一个模块的情况）

阻塞

阻塞时，在调用结果返回前，当前线程会被挂起，并在得到结果之后返回。

非阻塞

非阻塞时，如果不能立刻得到结果，则该调用者不会阻塞当前线程。因此对应非阻塞的情况，调用者需要定时轮询查看处理状态。

两者的关系

先上结论

同步/异步是描述两个模块之间的关系。
阻塞/非阻塞是描述一个模块的情况。

看看知乎上的大佬这么说的

怎样理解阻塞非阻塞与同步异步的区别？

一个网络包从应用程序A发到另一台电脑上的应用程序B，需要经历：

从A的业务代码到A的软件框架
从A的软件框架到计算机的操作系统内核
从A所在计算机的内核到网卡
从网卡经过网线发到交换机等设备，层层转发，到达B所在计算机的网卡
从B所在计算机的网卡到B所在计算机的内核
从B所在计算机的内核到B的程序的用户空间
从B的软件框架到B的业务代码

这个层级关系就像是过程调用一样，前一级调用后一级的功能，后一级返回一个结果给前一级（比如：成功，或者失败）。只有在单独一级的调用上，可以说同步还是异步的问题。所谓同步，是指调用协议中结果在调用完成时返回，这样调用的过程中参与双方都处于一个状态同步的过程。而异步，是指调用方发出请求就立即返回，请求甚至可能还没到达接收方，比如说放到了某个缓冲区中，等待对方取走或者第三方转交；而结果，则通过接收方主动推送，或调用方轮询来得到。

从这个定义中，我们看，首先1和7，这取決于软件框架的设计，如果软件框架可以beginxxX，然后立即返回，这就是一种异步调用，再比如javascript当中的异步HTTP调用，传入参数时提供一个回调函数，回调函数在完成时调用，再比如协程模型，调用接口后马上切换到其他协程继续执行，在完成时由框架切换回到协程中，这都是典型的异步接口设计。

而2和6，其实都需要调用方自己把数据在内核和用户空间里搬来搬去，其实都是同步接口，除非是10CP这样的专门的异步传输接口，所以这一级其实是同步的，阻塞与非阻塞的区别其实是影响调用接口的结果（在特定条件下是否提前返回结果），而不是调用方式。（就是说从接口形式上看其实都是同步的，都会在调用结束时返回结果，但是在暂时没有新数据的情况下，阻塞会等待到有新数据时再返回（接收端处理未完成情况），而非阻塞会立即返回“没有新数据”，其实是这个差别）。

3和5，内核一般通过缓冲区，使用DMI来传输数据，所以这一步又是异步的。

4，以太网是个同步时序逻辑，随信号传输时钟，必须两边设备同时就绪了才能开始传输数据，这叉是同步的。

总结来说，讨论究竟是异步还是同步，一定要严格说明说的是哪一部分。说非阻塞是同步而不是异步，这毫无疑问是正确的，然而说某个框架是异步I/O的框架，这也是正确的，因为说的其实是框架提供给业务代码的接口是异步的，不管是回调还是协程，比如说我们可以说某个库是异步的HTTPClient，并没有什么问题，因为说的是给业务代码的接口。由于通常异步的框架都需要在2中使用非阻塞的接口，的确会有很多人把非阻塞和异步混为一谈。

“同步/异步”、“阻塞/非阻塞”可以从字面上理解。所谓“同步”，是指“相同的步调”，既然是相同，必然涉及比较，那么就是关于两者（调用方与被调用方）的。也即是说，“同步/异步”描述的是两个模块之间的关系。所谓“阻塞”，是指“无法前进”了，“卡住了”，等待在那里而不能做其他事情了。也即是说，“阻塞/非阻塞”描述的是一个模块自身的运行状态。“同步/异步”、“阻塞/非阻塞”一个是描述两者的关系，另一个是描述一者的状态，所以其实是讲的不同的事情。

Java中对BIO/NIO/AIO的支持

在Linux 2.6以后，Java中NIO和AIO都是通过epoll来实现的，而在Windows上，AIO是通过IOCP来实现的。

可以把Java中的BIO、NIO和AIO理解为是Java语言对操作系统的各种IO模型的封装。程序员在使用这些API的时候，不需要关心操作系统层面的知识，也不需要根据不同操作系统编写不同的代码。只需要使用Java的API就可以了。

Java中的IO都是依赖操作系统内核进行的，我们程序中的IO读写其实调用的是操作系统内核中的read&write两大系统调用。

BIO

BIO（Blocking IO）阻塞式IO。同步阻塞I/O模式，数据的读取写入必须阻塞在一个线程内等待其完成。

采用 BIO 通信模型的服务端，通常有一个独立的 Acceptor 线程负责监听客户端的连接，它接收到客户端的连接请求之后，为每个客户端创建一个新的线程进行链路处理，处理完之后，通过输出流返回应答客户端，线程销毁。这就是典型的一请求一应答通信模型。这个是在多线程情况下执行的。当在单线程环境条件下时，在 while 循环中服务端会调用 accept 方法等待接收客户端的连接请求，一旦收到这个连接请求，就可以建立 socket，并在 socket 上进行读写操作，此时不能再接收其他客户端的连接请求，只能等待同当前服务端连接的客户端的操作完成或者连接断开。

该模型最大的问题就是缺乏弹性伸缩能力，当客户端并发访问量增加后，服务端的线程个数和客户端并发访问数呈 1:1 的正比关系，由于线程是 Java 虚拟机非常宝贵的系统资源，当线程数膨胀之后，系统的性能将急剧下降，随着并发访问量的继续增大，系统会发生线程堆栈溢出、创建新线程失败等问题，并最终导致进程宕机或者僵死，不能对外提供服务。

适用场景

BIO 适用于连接数目比较小且固定的结构。它对服务器资源要求比较高，并发局限于应用中，JDK1.4之前唯一选择，但程序直观简单易理解，如之前在 Apache 中使用。

优点缺点

优点

模型简单，编码效率高

缺点

原始BIO负载能力低，因为读写方法都是阻塞，当前只能处理单一任务，即便是开了多线程或线程池，当海量任务时其可处理能力依然低下。
每一个客户端建立连接后都需要创建独立的线程与客户端进行数据的读写，业务处理
当并发数较大时，会创建大量的流程来处理连接，系统资源会出现很大的开销
连接建立后，如果服务该客户端的线程没有数据可读时，线程则会阻塞在Read操作上，等待有数据后才读取，造成线程资源的浪费

采用 BIO 通信模型的服务端，通常由一个独立的 Acceptor 线程负责监听客户端的连接。我们一般通过在 while(true)循环中服务端会调用 accept() 方法等待接收客户端的连接的方式监听请求，请求一旦接收到一个连接请求，就可以建立通信套接字在这个通信套接字上进行读写操作，此时不能再接收其他客户端连接请求，只能等待同当前连接的客户端的操作执行完成，不过可以通过多线程来支持多个客户端的连接。

如果要让 BIO 通信模型 能够同时处理多个客户端请求，就必须使用多线程（主要原因是 socket.accept()、 socket.read()、 socket.write() 涉及的三个主要函数都是同步阻塞的），也就是说它在接收到客户端连接请求之后为每个客户端创建一个新的线程进行链路处理，处理完成之后，通过输出流返回应答给客户端，线程销毁。这就是典型的 一请求一应答通信模型 。我们可以设想一下如果这个连接不做任何事情的话就会造成不必要的线程开销，不过可以通过 线程池机制 改善，线程池还可以让线程的创建和回收成本相对较低。使用FixedThreadPool 可以有效的控制了线程的最大数量，保证了系统有限的资源的控制，实现了N(客户端请求数量):M(处理客户端请求的线程数量)的伪异步I/O模型（N 可以远远大于 M）

伪异步 IO，为了解决同步阻塞I/O面临的一个链路需要一个线程处理的问题，后来有人对它的线程模型进行了优化一一一后端通过一个线程池来处理多个客户端的请求接入，形成客户端个数M：线程池最大线程数N的比例关系，其中M可以远远大于N.通过线程池可以灵活地调配线程资源，设置线程的最大值，防止由于海量并发接入导致线程耗尽。

伪异步I/O通信框架采用了线程池实现，因此避免了为每个请求都创建一个独立线程造成的线程资源耗尽问题。不过因为它的底层任然是同步阻塞的BIO模型，因此无法从根本上解决问题。

实现案例

package com.example.demo.io.bio;

import java.io.*;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.*;

import static java.nio.charset.StandardCharsets.UTF_8;

public class BioServer {
    /**
     * 原始BIO
     *
     * @param args
     * @throws IOException
     */
    public static void main0(String[] args) throws IOException {
        ServerSocket serverSocket = new ServerSocket();
        serverSocket.bind(new InetSocketAddress("0.0.0.0", 8888), 50);
        Socket socket;
        while ((socket = serverSocket.accept()) != null) {
            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            String line;
            try {
                while ((line = bufferedReader.readLine()) != null) {
                    System.out.println(line);
                    OutputStream out = socket.getOutputStream();
                    out.write(line.getBytes(UTF_8));
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            socket.close();
        }
    }

    /**
     * 同时处理多个客户端请求
     * @param args
     * @throws IOException
     */
    public static void main1(String[] args) throws IOException {
        ServerSocket serverSocket = new ServerSocket();
        serverSocket.bind(new InetSocketAddress("0.0.0.0", 8888), 50);
        Socket acceptSocket;
        while ((acceptSocket = serverSocket.accept()) != null) {
            Socket finalAcceptSocket = acceptSocket;
            new Thread(() -> {
                Socket socket = finalAcceptSocket;
                try {
                    BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
                    String line;
                    while ((line = bufferedReader.readLine()) != null) {
                        System.out.println(line);
                        OutputStream out = socket.getOutputStream();
                        out.write(line.getBytes(UTF_8));
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    try {
                        socket.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }

    /**
     * 伪异步 IO
     * @param args
     * @throws IOException
     */
    public static void main(String[] args) throws IOException {
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        ServerSocket serverSocket = new ServerSocket();
        serverSocket.bind(new InetSocketAddress("0.0.0.0", 8888), 50);
        Socket acceptSocket;
        while ((acceptSocket = serverSocket.accept()) != null) {
            Socket finalAcceptSocket = acceptSocket;
            FutureTask<String> task = new FutureTask<>(() -> {
                Socket socket = finalAcceptSocket;
                try {
                    BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
                    String line;
                    while ((line = bufferedReader.readLine()) != null) {
                        System.out.println(line);
                        OutputStream out = socket.getOutputStream();
                        out.write(line.getBytes(UTF_8));
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    try {
                        socket.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
                return null;
            });
            executorService.submit(task);
        }
    }
}

public class BioClient {
    public static void main(String[] args) throws IOException, InterruptedException {
        //模拟五个client
        for (int i = 0; i < 5; i++) {
            new Thread(() -> {
                try {
                    Socket socket = new Socket("127.0.0.1", 8888);
                    for (int j = 0; j < 10; j++) {
                        try {
                            socket.getOutputStream().write(String.format("%s %s \r\n",Thread.currentThread().getName(),DateUtils.getDate()).getBytes(UTF_8));
                            Thread.sleep(10);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

NIO

Not Blocking IO 非阻塞式IO，NIO采用基于channel（管道）面向Buffer（缓冲区），采用Selector（多路复用器）的。

Buffer（缓冲区）

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

Buffer类型：我们最常用的就是ByteBuffer，实际上每一种java基木类型都对于了一种缓存区（除了Boolean类型）,如下所示：

ByteBuffer：字节缓冲区
CharBuffer：字符缓冲区
ShortBuffer：短整型缓冲区
IntBuffer：整形缓冲区
LongBuffer：长整形缓冲区
FloatBuffer：浮点型缓冲区
DoubleBuffer：双精度浮点型缓冲区

缓冲区的类图继承关系如下所示：
Package nio.png

Channel (管道、通道）

通道(Channel)，它就像自来水管道一样，网络数据通过Channel读取和写入，通道与流不同之处在于通道是双向的，而流只是一个方向上移动（一个流必须是Inputstream或OutputStream的子类），而通道可以用于读、写或者二者同时进行，最关键的是可以与多路复用器结合起来，有多种的状态位，方便多路复用器去识别。

事实上通道分为两大类，一类是网络读写的(SelectableChanneI)，一类是用于文件操作的(FileChannel)，我们使用MJSocketChanneI和ServerSockerChannel都是SelectableChannel的子类。

Selector（选择器、多路复用器）

多路复用器(Selector)，他是NIO编程的基础，非常重要。多路复用器提供选择己经就绪的任务的能力。简单说，就是Selctor会不断地轮询注册在其上的通道(Channel)，如果某个通道发生了读写操作，这个通道就处于就绪状态，会被Selector轮询出来，然后通过SelectionKey可以取得就绪的Channel集合，从而进行后续的IO操作。

一个多路复用器(Selector)可以负责成千上万Channel通道，没有上限，这也是JDK使用了epoll代了传统的select实现，获得连接句柄没有限制。这也就意味着我们只要一个线程负责轮询，就可以接入成千上万个客户端，这是JDK NIO库的巨大进步。

Selector线程就类似一个管理者(Master)，管理了成千上万个管道，然后轮询那个管道的数据己经准备好，通知cpu执行IO的读取或写入操作。

Selector模式：
当IO事件（管道）注册到选择器以后，Selector会分配给和个管道一个key值，相当于标签。Selector选择器是以轮询的方式进行查找注册的所有IO事件（管道），当我们的IO事件（管道）准备就绪后，select就会识别，会通过key值来找到相应的管道，进行相关的数据处理操作（从管道里读或写数据，写到我们的数据缓冲区中）。

每个管道都会对选择器进行注册不同的事件状态，以便选择器查找。

SelectionKey.OP_CONNECT//socket连接
SelectionKey.OP_ACCEPT//socket接收
SelectionKey.OP_READ//读取
SelectionKey.OP_WRlTE//写入

适用场景

NIO 适用于连接数目多且连接比较短的架构，比如聊天服务器，并发局限于应用中，变成比较复杂。JDK1.4开始支持，如在 Nginx、Netty 中使用。

优点缺点

优点

客户端发起的连接操作是异步的，可以通过在多路复用器注册OP_CONNECT等待后续结果，不需要像之前的客户端那样被同步阻塞；
SocketChannel的读写操作都是异步的，如果没有可读写的数据它不会同步等待，直接返回，这样IO通信线程就可以处理其它的链路，不需要同步等待这个链路可用；
线程模型的优化：由于JDK的Selector在Linux等主流操作系统上通过epoll实现，它没有连接句柄数的限制（只受限于操作系统的最大句柄数或者对单个进程的句柄限制），这意味着一个Selector线程可以同时处理成千上万个客户端连接，而且性能不会随着客户端的增加而线性下降，因此，它非常适合做高性能、高负载的网络服务器。

缺点

编程复杂、编程模型难、维护成本高

为什么大家都不愿意用 JDK 原生 NIO 进行开发呢？从上面的代码中大家都可以看出来，是真的难用！除了编程复杂、编程模型难之外，它还有以下让人诟病的问题：

JDK 的 NIO 底层由 epoll 实现，该实现饱受诟病的空轮询 bug 会导致 cpu 飙升 100%
项目庞大之后，自行实现的 NIO 很容易出现各类 bug，维护成本较高，Netty 的出现很大程度上改善了 JDK 原生 NIO 所存在的一些让人难以忍受的问题。

实现案例

package com.example.demo.io.nio;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;

public class NioReadFile {
    public static void main(String args[]) throws IOException {
        RandomAccessFile file = new RandomAccessFile("/Users/biaoyang/IdeaProjects/demo/src/main/java/com/example/demo/io/nio/NioClient.java", "r");
        FileChannel fileChannel = file.getChannel();
        ByteBuffer byteBuffer = ByteBuffer.allocate(512);
        while (fileChannel.read(byteBuffer) > 0) {
            // 翻转缓冲区以准备get操作
            byteBuffer.flip();
            while (byteBuffer.hasRemaining()) {
                System.out.print((char) byteBuffer.get());
            }
        }
        file.close();
    }
}

package com.example.demo.io.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Date;
import java.util.Iterator;
import java.util.Set;

public class NioServer {
    //缓冲区大小
    private int BLOCK = 4096;
    //接收数据缓冲区
    private ByteBuffer sendBuffer = ByteBuffer.allocate(BLOCK);
    //发送数据缓冲区
    private ByteBuffer receiveBuffer = ByteBuffer.allocate(BLOCK);
    //io复用器
    private Selector selector;

    public NioServer(int port) throws IOException{
        /**
         * 以下的所有说明均已linux系统底层进行说明：
         *      nio 的底层实现是 epoll 模式，采用多路复用技术，对nio的代码进行深入分析，结合epoll的底层实现
         * 进行详细的说明
         *      1.linux网络编程是两个进程之间的通信，跨集群合网络
         *      2.开启一个socket线程，在linux系统上任何操作均以文件句柄数表示，默认情况下
         *        一个线程可以打开1024个句柄，也就说最多同时支持1024个网络连接请求。阿里云默认打开65535个文件
         *        句柄，通常情况下，1G内存最多可以打开10w个句柄数
         */
        //打开服务器套接字通道
        //底层；在linux上面开启socket服务，启动一个线程，绑定ip地址和端口号
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        //服务器配置为非阻塞
        serverSocketChannel.configureBlocking(false);
        //检索与此通道关联的服务器套接字
        ServerSocket serverSocket = serverSocketChannel.socket();
        //进行服务绑定
        serverSocket.bind(new InetSocketAddress(port));
        //通过open()方法找到Selector
        //底层：开启epoll，为当前socket服务创建epoll服务，epoll_create
        selector = Selector.open();
        //注册到selector
        /**
         * 底层：
         *      1.将当前的epoll,服务器地址，端口号绑定,如果有连接请求，直接添加到epoll中，epoll的底层是红黑树，
         *  可以快速的实现连接的查找和状态更新。如果有新的连接过来，直接存放到epoll中。如果有连接过期，中断，
         *  会从epoll中删除。
         *      2.通过epoll_ctl添加到epoll的同时，会注册一个回调函数给内核，当网卡有数据来的时候，会通知内核，内核
         *      调用回调函数，将当前内核数据的事件状态添加到list链表中
         */
        serverSocketChannel.register(selector,SelectionKey.OP_ACCEPT);
        System.out.println("服务器启动 端口:"+port);
    }

    //监听
    private void listen() throws IOException{
        while (true){
            //选择一组键，并且相应得通道已经打开
            /**
             * epoll底层维护一个链表，rdlist，基于事件驱动模式，当网卡有数据请求过来，会发起硬件中断，通知内核已经有来了。内核调用
             * 回调函数，将当前的事件添加到rdlist中，将当前可用的rdlist列表发送给用户态，用户去遍历rdlist中的事件，进行处理
             */
            int readyChannels = selector.select();
            if(readyChannels == 0) continue;
            //返回此选择器得已选择键集
            Set<SelectionKey> selectionKeys = selector.selectedKeys();
            Iterator<SelectionKey> iterator = selectionKeys.iterator();
            while (iterator.hasNext()){
                SelectionKey selectionKey = iterator.next();
                //获取当前epoll的rdlist复制到用户态，遍历，同时删除当前rdlist中的事件
                iterator.remove();
                handleKEY(selectionKey);
            }
        }
    }

    //处理请求
    private void handleKEY(SelectionKey selectionKey) throws IOException {
        //接受请求
        ServerSocketChannel server = null;
        SocketChannel client = null;
        String receiveText;
        String sendText;
        int count = 0;
        if(selectionKey.isAcceptable()){
            System.out.println("当前通道已准备好接受新的套接字连接");
            //返回为止创建此键的通道
            server = (ServerSocketChannel) selectionKey.channel();
            //接受次通道套接字的连接
            //此方法返回的套接字通道（如果有）将处于阻塞模式
            client = server.accept();
            //配置为非阻塞
            client.configureBlocking(false);
            //注册到selector，等待连接
            client.register(selector,SelectionKey.OP_READ);
        }else if(selectionKey.isReadable()){
            //返回为之创建此键的通道
            client = (SocketChannel) selectionKey.channel();
            //将缓冲区清空以备下次读取
            receiveBuffer.clear();
            //读取服务器发送来的数据到缓冲区
            try {
                count = client.read(receiveBuffer);
            }catch (Exception e){}

            if(count>0){
                receiveText = new String(receiveBuffer.array(),0,count);
                System.out.println("服务器端接受客户端数据  :"+receiveText);
                client.register(selector,SelectionKey.OP_WRITE);
            }
        }else if(selectionKey.isWritable()){
            //将缓冲区清空以备下次写入
            sendBuffer.clear();
            //返回为之创建此键的通道
            client = (SocketChannel) selectionKey.channel();
            sendText = "message form server  " + new Date();
            //向缓冲区中输入数据
            sendBuffer.put(sendText.getBytes());
            //将缓冲区个标志复位，因为李米娜put了数据标志被改变要想从中读取数据发向服务器，就要复位
            sendBuffer.flip();
            //输出到通道
            client.write(sendBuffer);
            System.out.println("服务器端向客服端发送数据： "+sendText);
            client.register(selector,SelectionKey.OP_READ);
        }
    }

    public static void main(String[] args) throws IOException {
        int port = 8888;
        NioServer server = new NioServer(port);
        server.listen();
    }
}

package com.example.demo.io.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Date;
import java.util.Iterator;
import java.util.Set;

public class NioClient {

    private static boolean isEnding = false;
    //缓冲区大小
    private static int BLOCK = 4096;
    //接受数据缓冲区
    private static ByteBuffer sendBuffer = ByteBuffer.allocate(BLOCK);
    //发送数据缓冲区
    private static ByteBuffer receiveBuffer = ByteBuffer.allocate(BLOCK);
    //服务器地址
    private final static InetSocketAddress SERVER_ADDRESS = new InetSocketAddress("localhost",8888);

    public static void main(String[] args) throws IOException {
        //打开socker通道
        SocketChannel socketChannel = SocketChannel.open();
        //设置为非阻塞模式
        socketChannel.configureBlocking(false);
        //打开选择器
        Selector selector = Selector.open();
        //注册连接服务端socket动作
        socketChannel.register(selector, SelectionKey.OP_CONNECT);
        //连接
        socketChannel.connect(SERVER_ADDRESS);
        //分配缓冲区内存大小
        Set<SelectionKey> selectionKeys;
        Iterator<SelectionKey> iterator;
        SelectionKey selectionKey;
        SocketChannel client;
        String receiveText;
        String sendText;
        int count =  0;
        while (!isEnding){
            //选择一组键，其对应的通道已为 I/O 操作准备就绪
            //此方法执行处于阻塞模式的选择操作
            selector.select();
            //返回此选择器的已选择的键集
            selectionKeys = selector.selectedKeys();
            iterator = selectionKeys.iterator();
            while (iterator.hasNext()){
                selectionKey = iterator.next();
                if(selectionKey.isConnectable()){
                    System.out.println("client connect");
                    client = (SocketChannel) selectionKey.channel();
                    //判断此通道伤是否正在进行连接操作
                    //完成套接字通道的连接过程
                    if(client.isConnectionPending()){
                        client.finishConnect();
                        System.out.println("完成连接！");
                        sendBuffer.clear();
                        sendBuffer.put("Hello,Server".getBytes());
                        sendBuffer.flip();
                        client.write(sendBuffer);
                    }
                    client.register(selector,SelectionKey.OP_READ);
                }else if(selectionKey.isReadable()){
                    client = (SocketChannel) selectionKey.channel();
                    receiveBuffer.clear();
                    count = client.read(receiveBuffer);
                    if(count >0){
                        receiveText = new String(receiveBuffer.array(),0,count);
                        System.out.println("客户端接受服务器端的数据： "+receiveText);
                        client.register(selector,SelectionKey.OP_WRITE);
                    }
                }else if (selectionKey.isWritable()) {
                    sendBuffer.clear();
                    client = (SocketChannel) selectionKey.channel();
                    sendText = "message from client " + new Date();
                    sendBuffer.put(sendText.getBytes());
                    //将缓冲区各标志复位,因为向里面put了数据标志被改变要想从中读取数据发向服务器,就要复位
                    sendBuffer.flip();
                    client.write(sendBuffer);
                    System.out.println("客户端向服务器端发送数据： "+sendText);
                    client.register(selector, SelectionKey.OP_READ);
                    //isEnding = true;
                }
            }
            selectionKeys.clear();
        }
        selector.close();
        socketChannel.close();
    }
}

AIO

JDK1.7 升级了 NIO 类库，升级后的 NIO 类库被称为NIO2.0。也就是我们要介绍的 AIO。NIO2.0 引入了新的异步通道的概念，并提供了异步文件通道和异步套接字通道的实现。

异步通道提供两种方式获取操作结果。

通过 Java.util.concurrent.Future 类来表示异步操作的结果；
在执行异步操作的时候传入一个Java.nio.channels.CompletionHandler接口的实现类作为操作完成的回调。

NIO2.0 的异步套接字通道是真正的异步非阻塞 IO，它对应 UNIX 网络编程中的事件驱动 IO(AIO)，它不需要通过多路复用器(Selector)对注册的通道进行轮询操作即可实现异步读写，从而简化了 NIO 的编程模型。

Package channels.png

结论：异步 Socket Channel是被动执行对象，我们不需要想NIO编程那样创建一个独立的IO线程来处理读写操作。对于AsynchronousServerSocketChannel和AsynchronousSocketChannel，它们都由 JDK 底层的线程池负责回调并驱动读写操作。正因为如此，基于 NIO2.0 新的异步非阻塞 Channel 进行编程比 NIO 编程更为简单。

适用场景

AIO 适用于连接数目多且连接比较长（重操作）的架构，比如相册服务器，充分调用 OS 参与并发操作，编程比较复杂，JDK7 开始支持，在成长中，Netty 曾经使用过，后来放弃。

优点缺点

优点

支持连接数目多

缺点

编程比较复杂

实现案例

package com.example.demo.io.aio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.Scanner;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class AioServer implements Runnable {

    Charset charset = Charset.forName("UTF-8");

    public static void main(String[] args) throws InterruptedException {
        int port = 7890;
        new Thread(new AioServer(port)).start();
        TimeUnit.MINUTES.sleep(60);
    }

    int port;
    AsynchronousChannelGroup group;
    AsynchronousServerSocketChannel serverSocketChannel;

    public AioServer(int port) {
        this.port = port;
        init();
    }

    public void init() {
        try {
            // 创建处理线程池
            group = AsynchronousChannelGroup.withCachedThreadPool(Executors.newCachedThreadPool(), 5);
            // 创建服务channel
            serverSocketChannel = AsynchronousServerSocketChannel.open(group);
            // 绑定端口
            serverSocketChannel.bind(new InetSocketAddress(port));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void run() {
        // 接收请求
        // accept的第一个参数附件，第二个参数是收到请求后的接收处理器
        // 接收处理器AcceptHandler泛型的第一个参数的处理结果，这里是AsynchronousSocketChannel，即接收到的请求的channel
        // 第二个参数是附件，这里是AioServer，即其实例
        serverSocketChannel.accept(this, new AcceptHandler());
    }


    /**
     * 接收请求处理器
     * completionHandler泛型的第一个参数的处理结果，这里是AsynchronousSocketChannel，即接收到的请求的channel，
     * 第二个参数是附件，这里是AioServer，即其实例
     */
    class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel, AioServer> {

        @Override
        public void completed(AsynchronousSocketChannel result, AioServer attachment) {
            // 继续接收下一个请求，构成循环调用
            attachment.serverSocketChannel.accept(attachment, this);
            try {
                System.out.println("接收到连接请求：" + result.getRemoteAddress().toString());
                // 定义数据读取缓存
                ByteBuffer buffer = ByteBuffer.wrap(new byte[1024]);
                // 读取数据，并传入数据到达时的处理器
                // read的第一个参数数据读取到目标缓存，第二个参数是附件，第三个传输的读取结束后的处理器
                // 读取处理器泛型的第一个参数是读取的字节数，第二个参数输附件对象
                result.read(buffer, buffer, new ReadHandler(result));

                // 新开新城发送数据
                new Thread(new WriteThread(result)).start();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        @Override
        public void failed(Throwable exc, AioServer attachment) {

        }
    }

    /**
     * 读取数据处理器
     * completionHandler第一个参数是读取的字节数，第二个参数输附件对象
     */
    class ReadHandler implements CompletionHandler<Integer, ByteBuffer> {

        AsynchronousSocketChannel socketChannel;

        public ReadHandler(AsynchronousSocketChannel socketChannel) {
            this.socketChannel = socketChannel;
        }

        @Override
        public void completed(Integer result, ByteBuffer attachment) {
            if (result == -1) {
                attachment.clear();
                try {
                    socketChannel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
                return;
            }
            attachment.flip();
            String readMsg = charset.decode(attachment).toString();
            System.out.println("服务端接收到的数据：" + readMsg);
            attachment.compact();

            // 继续接收数据，构成循环
            socketChannel.read(attachment, attachment, this);
        }

        @Override
        public void failed(Throwable exc, ByteBuffer attachment) {

        }
    }

    /**
     * 写出数据处理器
     */
    class WriteHandler implements CompletionHandler<Integer, ByteBuffer> {
        AsynchronousSocketChannel socketChannel;
        Scanner scanner;

        public WriteHandler(AsynchronousSocketChannel socketChannel, Scanner scanner) {
            this.socketChannel = socketChannel;
            this.scanner = scanner;
        }


        @Override
        public void completed(Integer result, ByteBuffer attachment) {
            attachment.compact();
            String msg = scanner.nextLine();

            System.out.println("服务端即将发送的数据：" + msg);
            attachment.put(charset.encode(msg));
            attachment.flip();

            // 继续写数据，构成循环
            socketChannel.write(attachment, attachment, this);
        }

        @Override
        public void failed(Throwable exc, ByteBuffer attachment) {

        }
    }

    class WriteThread implements Runnable {

        private AsynchronousSocketChannel channel;

        public WriteThread(AsynchronousSocketChannel channel) {
            this.channel = channel;
        }

        @Override
        public void run() {
            // 第一缓冲区
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            Scanner scanner = new Scanner(System.in);
            String msg = scanner.nextLine();
            System.out.println("服务端输入数据：" + msg);
            buffer.put(charset.encode(msg + System.lineSeparator()));
            buffer.flip();
            // 写入数据，并有写数据时的处理器
            // write的第一个参数是数据写入的缓存，第二个参数是附件，第三个参数写结束后的处理器
            // 读取处理器泛型的第一个参数是写入的字节数，第二个是附件类型
            channel.write(buffer, buffer, new WriteHandler(channel, scanner));
        }
    }
}

package com.example.demo.io.aio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.nio.charset.Charset;
import java.util.Scanner;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class AioClient {

    static Charset charset = Charset.forName("UTF-8");

    public static void main(String[] args) throws InterruptedException {
        int port = 7890;
        String host  = "127.0.0.1";
        // 启动客户端
        new Thread(new AIOClient(port, host)).start();
        TimeUnit.MINUTES.sleep(100);
    }

    static class AIOClient implements Runnable {

        int port;
        String host;
        AsynchronousChannelGroup group;
        AsynchronousSocketChannel channel;
        InetSocketAddress address;

        public AIOClient(int port, String host) {
            this.port = port;
            this.host = host;
            // 初始化
            init();
        }

        private void init() {
            try {
                // 创建处理线程组
                group = AsynchronousChannelGroup.withCachedThreadPool(Executors.newCachedThreadPool(), 5);
                // 创建客户端channel
                channel = AsynchronousSocketChannel.open(group);
                address = new InetSocketAddress(host, port);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        @Override
        public void run() {
            // 接收请求，并传入收到请求后的处理器
            // connect 方法的第一二个参数是目标地址，第二个参数是附件对象，第三个参数是连接处理器
            // 连接处理器的泛型的第一个参数为空(即Void)，第二个参数为附件
            channel.connect(address, channel, new ConnectHandler());
        }
    }

    /**
     * 连接处理器
     */
    static class ConnectHandler implements CompletionHandler<Void, AsynchronousSocketChannel> {

        @Override
        public void completed(Void result, AsynchronousSocketChannel attachment) {
            try {
                System.out.println("connect server: " + attachment.getRemoteAddress().toString());
                // 定义数据读取缓存
                ByteBuffer buffer = ByteBuffer.allocate(1024);
                // 读取数据，并传入到数据到达时的处理器
                attachment.read(buffer, buffer, new ReadHandler(attachment));
                // 新开线程，发送数据
                new WriteThread(attachment).start();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        @Override
        public void failed(Throwable exc, AsynchronousSocketChannel attachment) {

        }
    }

    /**
     * 读处理器
     */
    static class ReadHandler implements CompletionHandler<Integer, ByteBuffer> {
        AsynchronousSocketChannel channel;
        public ReadHandler(AsynchronousSocketChannel channel) {
            this.channel = channel;
        }

        @Override
        public void completed(Integer result, ByteBuffer attachment) {
            attachment.flip();
            String readMsg = charset.decode(attachment).toString();
            System.out.println("client receive msg: " + readMsg);
            attachment.compact();
            // 继续接收数据，构成循坏
            channel.read(attachment, attachment, this);
        }
        @Override
        public void failed(Throwable exc, ByteBuffer attachment) {
        }
    }

    /**
     * 写处理器
     */
    static class WriteHandler implements CompletionHandler<Integer, ByteBuffer> {
        AsynchronousSocketChannel channel;
        Scanner scanner;

        public WriteHandler(AsynchronousSocketChannel channel, Scanner scanner) {
            this.channel = channel;
            this.scanner = scanner;
        }

        @Override
        public void completed(Integer result, ByteBuffer attachment) {
            attachment.compact();
            System.out.print("client input data: ");
            String msg = scanner.nextLine();
            System.out.println("clinet will send msg:" + msg);
            attachment.put(charset.encode(msg));
            attachment.flip();
            // 继续写入数据，构成循环
            channel.write(attachment, attachment, this);
        }

        @Override
        public void failed(Throwable exc, ByteBuffer attachment) {

        }
    }


    /**
     * 写处理独立创建线程
     */
    static class WriteThread extends Thread {

        private AsynchronousSocketChannel channel;

        public WriteThread(AsynchronousSocketChannel channel) {
            this.channel = channel;
        }

        @Override
        public void run() {
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            Scanner scanner = new Scanner(System.in);
            System.out.print("client input data:");
            String msg = scanner.nextLine();
            System.out.println("client send msg:" + msg);
            buffer.put(charset.encode(msg));
            buffer.flip();
            channel.write(buffer, buffer, new WriteHandler(channel, scanner));
        }
    }
}

学习笔记-I/O模型详解

相关背景

什么是I/O？

内核kernel

系统中断

用户空间和内核空间

缓存 I/O

Unix I/O 模型

阻塞I/O模型

非阻塞I/O模型

I/O复用模型

信号驱动I/O模型

异步I/O模型

各种I/O模型的比较

I/O 模型概念

同步/异步

同步

异步

阻塞/非阻塞

阻塞

非阻塞

两者的关系

Java中对BIO/NIO/AIO的支持

BIO

适用场景

优点缺点

实现案例

NIO

Buffer（缓冲区）

Channel (管道、通道）

Selector（选择器、多路复用器）

适用场景

优点缺点

优点

缺点

实现案例

AIO

适用场景

优点缺点

优点

缺点

实现案例

相关资料