什么是NIO

NIO又叫New/Non-blocking IO，这个概念基本人人都听过，但是不一定每个人都懂他它的运行的原理。

这里我们来探讨这个问题，先用一个例子解释一下BIO到底阻塞了哪里。

/** * 这是一个单线程BIOServer * @author endless * @create 2020-03-23 */public class BioServerDemo {  public static void main(String[] args) throws IOException {	// 创建ServerSocket，并绑定端口    ServerSocket serverSocket = new ServerSocket(9999);    System.out.println("服务启动成功");    while (true) {      Socket socket = serverSocket.accept();      System.out.println("连接成功");      System.out.println("准备接收数据");      byte[] bytes = new byte[1024];      socket.getInputStream().read(bytes);      System.out.println("接收到了数据：" + new String(bytes));    }  }}

/** * BIO client * * @author endless * @create 2020-03-23 */public class BioClientDemo {  public static void main(String[] args) throws IOException {    // 连接Server    Socket socket = new Socket("127.0.0.1", 9999);    System.out.println("连接成功");    Scanner scanner = new Scanner(System.in);    // 循环等待输入消息    while (true) {      String str = scanner.next();      // 约定退出口令      if ("exit".equalsIgnoreCase(str)) {        socket.close();        System.exit(0);      }      socket.getOutputStream().write(str.getBytes());      socket.getOutputStream().flush();    }  }}

先运行Server

命令行打印服务启动成功，此时并无客户端连接，所以连接成功并未打印，说明程序被阻塞在了serverSocket.accept()方法

此时运行Client，Server打印日志连接成功和准备接收数据，此时Client尚未发送数据，Server被阻塞在socket.getInputStream().read(bytes)上，因此其他客户端无法进行连接。

在Client输入Hello回车，此时Server打印接收到了数据：Hello，说明客户端的连接发送过来数据了，此时服务端线程才解阻塞，在这个情况下，这个Server没有办法处理并发，同时期只能处理一个连接。

那么BIO是如何实现并发呢？答案也很明显，就是使用多线程，我们对Server进行一些小改动。

/** * 这是一个BIOServer * @author endless * @create 2020-03-23 */public class BioServerDemo {  public static void main(String[] args) throws IOException {    ServerSocket serverSocket = new ServerSocket(9999);    System.out.println("服务启动成功");    while (true) {      Socket socket = serverSocket.accept();      new Thread(()->{        System.out.println("连接成功");        System.out.println("准备接收数据");      	byte[] bytes = new byte[1024];        try {          socket.getInputStream().read(bytes);        } catch (IOException e) {          e.printStackTrace();        }        System.out.println("接收到了数据：" + new String(bytes));      }).start();    }  }}

使用子线程来对接收到的Socket进行处理，这样每个连接都被阻塞在单独的线程上，就可以实现并发访问Server。

总结：BIO的阻塞有两个地方：accept()和read()，并且BIO的并发只能通过多线程。

但是这里会有一个问题，就是如果绝大部分的连接都没有进行数据传输，只是建立了连接，这样就会产生很多无效的线程，而线程又是非常宝贵的稀缺资源，这样就会白白损失很多的性能，这也是BIO最大的性能瓶颈。

那能不能只用一个线程就能实现并发并且处理全部的连接呢？是否能设计一个Api，让accept和read不再阻塞，使用一个线程就能处理并发连接呢？答案是肯定的，这里就要用到我们的今天的主角NIO了。

NIO在JDK中被封装在了一个新的类中，我们先来写一个例子，这个例子实现了使用单线程来处理多连接。

/** * NIO Server Demo * * @author endless * @create 2020-03-23 */public class NioServerDemo {  // 保存客户端连接  static List<SocketChannel> channelList = new ArrayList<>();  public static void main(String[] args) throws IOException, InterruptedException {    // 创建NIO ServerSocketChannel    ServerSocketChannel serverSocket = ServerSocketChannel.open();    serverSocket.bind(new InetSocketAddress(9998));    // 设置ServerSocketChannel为非阻塞    serverSocket.configureBlocking(false);    System.out.println("服务启动成功");    while (true) {      SocketChannel socketChannel = serverSocket.accept();      if (socketChannel != null) { // 如果有客户端进行连接        System.out.println("连接成功");        // 设置SocketChannel为非阻塞        socketChannel.configureBlocking(false);        // 保存客户端连接在List中        channelList.add(socketChannel);      }      // 遍历连接进行数据读取      Iterator<SocketChannel> iterator = channelList.iterator();      while (iterator.hasNext()) {        SocketChannel o = iterator.next();        ByteBuffer byteBuffer = ByteBuffer.allocate(128);        int read = o.read(byteBuffer);        // 如果有数据，把数据打印出来        if (read > 0) {          System.out.println("接收到消息：" + new String(byteBuffer.array()));        } else if (read == -1) { // 如果客户端断开，把socket从集合中去掉          iterator.remove();          System.out.println("客户端断开连接");        }      }    }  }}

客户端可以复用之前的BIO客户端

运行NIOServer，Server启动完毕后运行两个Client，各发送一条消息进行测试

控制台显示两个连接成功，并且接收到了来自两个客户端的消息，表明Server可以使用单线程处理并发连接，这个Api的原理是什么呢？我们来进一步探究一下。

我们沿着源码一路往下找，会找到一个无法下载源码的文件

这里可以看得出，在Windows系统中，编译后的代码显示直接返回了WindowsSelectorProvider对象，很显然，这个对象是在和windows系统核心中的select方法交互，但是Linux中是不是也是这样呢，我们需要下载一个Linux版本的OpenJDK源码来探究一下。下载OpenJDK源码

下载后解压，Linux源码分布在\openjdk\jdk\src\share和\openjdk\jdk\src\solaris目录中。

在上图中可以看出，JDK在不同的系统中采用了不同的实现方案，这里使用的是EPollSelectorProvider，说明在Linux中，使用的是EPoll来实现的。

我们先来看看serverSocket.configureBlocking(false);到底是如何工作的，沿着源码往下找，发现一个本地方法。

这个本地方法的源码可以在OpenJDK中找到，如下图

上图红框中的函数就是这个本地方法调用的底层C语言的方法，前面的前缀是根据JNI调用规则添加的，我们知道，在C当中是可以直接调用操作系统的Api的，这个方法调用了fcntl这个命令，把传进来的blocking参数设置到了文件描述符上（文件描述符可以看作是一个对象，Linux中一切皆文件，类似于高级语言中的一切皆对象，任何的数据流转都要通过一个文件描述符来操作）。

接着看看serverSocket.accept()是如何实现

上图可以看到，accept方法调用了一个native方法accept0

这个accept0方法的描述简单翻译一下，就是接受一个新的连接，把给定的文件描述符引用设置为新的Socket，并将isaa[0]设置为套接字的远程地址，成功返回1，不成功返回IOStatus.UNAVAILABLE or IOStatus.INTERRUPTED。

继续探究一下本地方法的源码

这里调用了操作系统的accept方法，想知道这个方法文档的同学可以在Linux环境中使用man命令来查看

man命令可以查看Linux详尽的文档，2表示第二章：系统调用指令，最后加上你想查的指令方法即可

这里主要看一下返回值，accept返回一个非负数作为socket的描述符，如果失败返回-1。通过这个文件描述符，Java就可以通过native方法调用操作系统的API来在Socket中进行数据的传输。我们的代码用一个List保存了接收到的Socket，相当于保存了Socket的文件描述符，通过一个线程轮询这些文件描述符即可实现数据通信和处理新连接，这样就节约了大量的线程资源，但是大家想一想这种模型还有什么缺陷。

……

是的，如果连接数太多的话，还是会有大量的无效遍历，例如10000个连接中只有1000个连接时有数据的，但是由于其他9000个连接并没有断开，我们还是要每次轮询都遍历一万次，相当于有十分之九的遍历都是无效的，这显然不是一个让人很满意的状态。

总结：NIO的非阻塞是由操作系统来完成的，SocketChannel与文件描述符一一对应，通过遍历文件描述符来读取数据。

什么是多路复用器

上面的例子还不是Java NIO的完全体，仅仅是将原来的同步阻塞IO优化成了同步非阻塞IO，既然还是同步的，就意味着我们每次遍历，还是需要对每个Socket进行一次read操作来检查是不是有数据过来，都会调用系统内核的read指令，只不过是把阻塞变成了非阻塞，如果无用连接很多的话，那么绝大部分的read指令都是无意义的，这就会占用很多的CPU时间。

Linux有select、poll和epoll三个解决方案来实现多路复用，其中的select和poll，有点类似于上面的NIOServerDemo程序，他会把所有的文件描述符记录在一个数组中，通过在系统内核中遍历来筛选出有数据过来的Socket，只不过是从应用程序中遍历改成了在内核中遍历，本质还是一样的。

Epoll则使用了事件机制，在复用器中注册了一个回调事件，当Socket中有数据过来的时候调用，通知用户处理信息，这样就不需要对全部的文件描述符进行轮训了，这就是Epoll对NIO进行的改进。

我们来探究一下在Java中是如何用Epoll实现NIO事件机制的，先对上面的NIOServerDemo再进行改进。

/** * NIO Selector Server Demo * * @author  endless * @create 2020-03-23 */public class NioSelectorServerDemo {  public static void main(String[] args) throws IOException, InterruptedException {    // 创建NIO ServerSocketChannel    ServerSocketChannel serverSocket = ServerSocketChannel.open();    serverSocket.socket().bind(new InetSocketAddress(9998));    // 设置ServerSocketChannel为非阻塞    serverSocket.configureBlocking(false);    // 打开Selector处理Channel，即创建epoll    Selector selector = Selector.open();    // 将ServerSocket注册到selector用来接收连接    serverSocket.register(selector, SelectionKey.OP_ACCEPT);    System.out.println("服务启动成功");    while (true) {      // 阻塞等待需要处理的事件发生      selector.select();      // 获取selector中注册的全部事件的 SelectionKey 实例      Set<SelectionKey> selectionKeys = selector.selectedKeys();      Iterator<SelectionKey> iterator = selectionKeys.iterator();      // 遍历SelectionKey对事件进行处理      while (iterator.hasNext()) {        SelectionKey key = iterator.next();        iterator.remove();        // 如果是OP_ACCEPT事件，则进行连接获取和事件注册        if (key.isAcceptable()) {          ServerSocketChannel server = (ServerSocketChannel) key.channel();          SocketChannel socketChannel = server.accept();          socketChannel.configureBlocking(false);          // 这里只注册了读事件，如果需要给客户端发送数据可以注册写事件          socketChannel.register(selector, SelectionKey.OP_READ);          System.out.println("客户端连接成功");        }        // 如果是OP_READ事件，则进行读取和打印        if (key.isReadable()) {          SocketChannel socketChannel = (SocketChannel) key.channel();          ByteBuffer byteBuffer = ByteBuffer.allocate(128);          int read = socketChannel.read(byteBuffer);          // 如果有数据，把数据打印出来          if (read > 0) {            System.out.println("接收到消息：" + new String(byteBuffer.array()));          } else if (read == -1) { // 如果客户端断开连接，关闭Socket            System.out.println("客户端断开连接");            socketChannel.close();          }        }      }    }  }}

这段代码需要关注的点有以下几个方法：

Selector.open()socketChannel.register()selector.select()

接下来就来看看这三个方法究竟做了什么。

Select.open()

首先调用了SelectorProvider的openSelector()方法，这个方法返回一个EPollSelectorImpl实例

EPollSelectorProvider.java

    public AbstractSelector openSelector() throws IOException {        return new EPollSelectorImpl(this);    }

EPollSelectorImpl的构造方法中new了一个EPollArrayWrapper实例

EPollSelectorImpl.java

    EPollSelectorImpl(SelectorProvider sp) throws IOException {        super(sp);        // 创建管道，用long类型返回管道的两个文件描述符。读端以高32位的形式返回，写端以低32位的形式返回。        long pipeFds = IOUtil.makePipe(false);        fd0 = (int) (pipeFds >>> 32);        fd1 = (int) pipeFds;        // 创建Epoll文件描述符，并且创建映射数组记录事件        pollWrapper = new EPollArrayWrapper();        // 初始化中断文件描述符，把新创建的Epoll注册到管道的读文件描述符上        pollWrapper.initInterrupt(fd0, fd1);        fdToKey = new HashMap<>();    }

EPollArrayWrapper的构造方法中创建了Epoll实例

EPollArrayWrapper.java

    EPollArrayWrapper() throws IOException {        // 创建了Epoll实例，并将它的事件数组地址记录下来方便操作        epfd = epollCreate();        // the epoll_event array passed to epoll_wait        int allocationSize = NUM_EPOLLEVENTS * SIZE_EPOLLEVENT;        pollArray = new AllocatedNativeObject(allocationSize, true);        pollArrayAddress = pollArray.address();        // eventHigh needed when using file descriptors > 64k        if (OPEN_MAX > MAX_UPDATE_ARRAY_SIZE)            eventsHigh = new HashMap<>();    }    	...	    private native int epollCreate();    

Native方法epollCreate()的源码，调用了内核的epoll_create指令，创建并获取了一个文件描述符。

epoll有三个指令，epoll_create、epoll_ctl、epoll_wait，都可以在Linux环境中使用man命令来查看详细文档。

EPollArrayWrapper.c

JNIEXPORT jint JNICALLJava_sun_nio_ch_EPollArrayWrapper_epollCreate(JNIEnv *env, jobject this){    /*     * epoll_create expects a size as a hint to the kernel about how to     * dimension internal structures. We can't predict the size in advance.     */    int epfd = epoll_create(256);    if (epfd < 0) {       JNU_ThrowIOExceptionWithLastError(env, "epoll_create failed");    }    return epfd;}

注册事件时其实并没有对Epoll进行事件添加，而是在只是把它加入了待添加的容器。

public final SelectionKey register(Selector sel, int ops,                                       Object att)        throws ClosedChannelException    {        synchronized (regLock) {            if (!isOpen())                throw new ClosedChannelException();            if ((ops & ~validOps()) != 0)                throw new IllegalArgumentException();            if (blocking)                throw new IllegalBlockingModeException();            // 判断是否存在事件的key，存在的话就更新，不存在就新建            // 最终都会走到 EPollArrayWrapper#setUpdateEvents方法            SelectionKey k = findKey(sel);            if (k != null) {                k.interestOps(ops);                k.attach(att);            }            if (k == null) {                // New registration                synchronized (keyLock) {                    if (!isOpen())                        throw new ClosedChannelException();                    k = ((AbstractSelector)sel).register(this, ops, att);                    addKey(k);                }            }            return k;        }    }

AbstractSelector.register调用实现的子类方法implRegister

EPollSelectorImpl.java

    protected void implRegister(SelectionKeyImpl ski) {        if (closed)            throw new ClosedSelectorException();        SelChImpl ch = ski.channel;        int fd = Integer.valueOf(ch.getFDVal());        fdToKey.put(fd, ski);        pollWrapper.add(fd);        keys.add(ski);    }

调用setUpdateEvents写入待更新事件容器

EPollArrayWrapper.java

    /**     * Add a file descriptor     */    void add(int fd) {        // 强制初始update events为0，因为他可能会被之前注册kill掉        synchronized (updateLock) {            assert !registered.get(fd);            setUpdateEvents(fd, (byte)0, true);        }    }    /**     * 设置文件描述符的待更新事件到 eventsHigh 中。     */    private void setUpdateEvents(int fd, byte events, boolean force) {        if (fd < MAX_UPDATE_ARRAY_SIZE) {            if ((eventsLow[fd] != KILLED) || force) {                eventsLow[fd] = events;            }        } else {            Integer key = Integer.valueOf(fd);            if (!isEventsHighKilled(key) || force) {                eventsHigh.put(key, Byte.valueOf(events));            }        }    }

调用selector.select()时，会将刚才注册的待更新事件绑定到文件描述符上，然后进入阻塞状态等待事件回调。

EPollSelectorImpl.java

    protected int doSelect(long timeout) throws IOException {        if (closed)            throw new ClosedSelectorException();        processDeregisterQueue();        try {            begin();            pollWrapper.poll(timeout);        } finally {            end();        }        processDeregisterQueue();        int numKeysUpdated = updateSelectedKeys();        if (pollWrapper.interrupted()) {            // Clear the wakeup pipe            pollWrapper.putEventOps(pollWrapper.interruptedIndex(), 0);            synchronized (interruptLock) {                pollWrapper.clearInterrupted();                IOUtil.drain(fd0);                interruptTriggered = false;            }        }        return numKeysUpdated;    }

EPollArrayWrapper.java

    int poll(long timeout) throws IOException {    	// 将上一步待更新的时间进行注册        updateRegistrations();        // 进入阻塞状态，等待事件发生        updated = epollWait(pollArrayAddress, NUM_EPOLLEVENTS, timeout, epfd);        for (int i=0; i<updated; i++) {            if (getDescriptor(i) == incomingInterruptFD) {                interruptedIndex = i;                interrupted = true;                break;            }        }        return updated;    }    /**     * Update the pending registrations.     */    private void updateRegistrations() {        synchronized (updateLock) {            int j = 0;            while (j < updateCount) {                int fd = updateDescriptors[j];                short events = getUpdateEvents(fd);                boolean isRegistered = registered.get(fd);                int opcode = 0;                if (events != KILLED) {                    if (isRegistered) {                        opcode = (events != 0) ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;                    } else {                        opcode = (events != 0) ? EPOLL_CTL_ADD : 0;                    }                    if (opcode != 0) {                    	// 调用native方法进行事件绑定                        epollCtl(epfd, opcode, fd, events);                        if (opcode == EPOLL_CTL_ADD) {                            registered.set(fd);                        } else if (opcode == EPOLL_CTL_DEL) {                            registered.clear(fd);                        }                    }                }                j++;            }            updateCount = 0;        }    }    

总结：到此为止，我们已经粗略的把整个NIO调用流程都梳理了一遍，Java调用了操作系统的Api来创建Socket，获取到Socket的文件描述符，再创建一个Selector对象，对应操作系统的EPoll描述符，将获取到的Socket连接的文件描述符的事件绑定到Selector对应的EPoll文件描述符上，进行事件的异步通知，这样就实现了使用一条线程，并且不需要太多的无效的遍历，将事件处理交给了操作系统内核，大大提高了效率。

EPoll指令详解epoll_create

int epoll_create(int size);

创建一个epoll实例，并返回一个非负数作为文件描述符，用于对epoll接口的所有后续调用。参数size代表可能会容纳size个描述符，但size不是一个最大值，只是提示操作系统它的数量级，现在这个参数基本上已经弃用了。

epoll_ctl

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

使用文件描述符epfd引用的epoll实例，对目标文件描述符fd执行op操作。

参数epfd表示epoll对应的文件描述符，参数fd表示socket对应的文件描述符。

参数op有以下几个值：EPOLL_CTL_ADD：注册新的fd到epfd中，并关联时间eventEPOLL_CTL_MOD：修改已经注册的fd的监听事件；EPOLL_CTL_DEL：从epfd中移除fd，并且忽略掉绑定的event，这时event可以为null；

参数event是一个结构体。

	struct epoll_event {	    __uint32_t   events;      /* Epoll events */	    epoll_data_t data;        /* User data variable */	};		typedef union epoll_data {	    void        *ptr;	    int          fd;	    __uint32_t   u32;	    __uint64_t   u64;	} epoll_data_t;

events有很多可选值，这里只举例最常见的几个：

EPOLLIN ：表示对应的文件描述符是可读的（close也会发送消息）；EPOLLOUT：表示对应的文件描述符是可写的；EPOLLERR：表示对应的文件描述符发生了错误；

成功则返回0，失败返回-1

epoll_wait

int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);

等待文件描述符epfd上的事件。

epfd是Epoll对应的文件描述符，events表示调用者所有可用事件的集合，maxevents表示最多等到多少个事件就返回，timeout是超时时间。

完

操作系统的IO还涉及到零拷贝和直接内存两部分的知识，也是操作系统提高性能的利器，将在以后的文章中进行探讨。