高性能网络IO框架研究一

IoT平台要求能够接入大规模的终端数据，因此对于底层的IO通信系统的性能和稳定性要求非常高。于是我对高性能IO框架进行了一些深入的研究。并将研究的内容总结出来，以供大家交流学习。

前面写了一篇关于高性能网络IO框架在，传统的Netty受到了非常大的挑战，不过总体来说，Netty在整个Java的生态体系中仍然还是最为重要的网络通信框架。

网络I/O的三种模式

BIO —— Block I/O 同步阻塞型IO

同步阻塞型IO是最简单的一种I/O模式。首先，阻塞与非阻塞的意思。阻塞IO指的是需要内核IO操作彻底完成后才返回到用户空间执行用户程序的操作指令。“阻塞”指的是用户程序（发起IO请求的进程或者线程）的执行状态。传统的IO模型都是阻塞IO模型，并且在Java中默认创建的socket都属于阻塞IO模型。在程序中的表现形式就是执行的时候会一直等待IO执行完毕，线程才会继续执行下去。

package BIO.demo;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;

public class Test1 {
    public static void main(String[] args) throws IOException {
        ServerSocket serverSocket = new ServerSocket(9000);
        while (true)
        {
            System.out.println("等待客户端的连接……");
            Socket accept = serverSocket.accept();
            System.out.println("客户端已经连接");
            //一旦客户端建立连接就将socket交给一个新的阻塞线程去处理。
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        handle(accept);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }

    public static void handle(Socket accept) throws IOException {
        byte[] aByte = new byte[1024];
        System.out.println("准备read客户端数据……");
        int read = accept.getInputStream().read(aByte);
        System.out.println(read);
        if (read!=-1)
        {
            String s = new String(aByte, 0, read);
            System.out.println("接收到客户端的数据"+s);
        }
    }
}

BIO比较简单，而且实现起来也比较容易，不容易出错。如果应用的并发连接数并不多的情况下(并发不过千)，或者服务器的并发处理性能可以满足要求的情况下，也可以采用这个模式。对性能和整体的可用性并不会有太大的影响。

NIO —— Non-Block I/O 非阻塞型IO

下面的代码是采用了Selector，每个一段时间去换取channel的selector中是否有网络IO事件，这里是捕获连接（accept）和读取（read）这两个事件。一旦读取到事件发生，则进行处理。这样就避免了BIO中线程阻塞的问题。同一个线程可以处理多个I/O访问。这样就使得整个系统能够处理更多的网络I/O。Netty就是基于Linux操作系统的select/epoll指令的NIO框架。当然Netty可以支持Reactive的模型进行处理，可以同时开多个线程接收并处理访问。

package NIO.demo;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;

public class NIOServer {

    public static void main(String[] args) throws IOException {
        ServerSocketChannel serverSocketChannel=ServerSocketChannel.open();
        //配置为非阻塞
        serverSocketChannel.configureBlocking(false);
        //绑定端口
        serverSocketChannel.bind(new InetSocketAddress("127.0.0.1",1234));
        //selector
        Selector selector = Selector.open();
        //serverSocketChannel 注册到selector中，并且监听连接时间
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        while (true){
            if(selector.select(1000)==0){
                System.out.println("未检测出的连接");
                continue;
            }
            //所有发生事件的通道 对应的SelectionKey
            Set<SelectionKey> selectionKeys = selector.selectedKeys();
            Iterator<SelectionKey> iterator = selectionKeys.iterator();
            while (iterator.hasNext()) {
                SelectionKey selectionKey = iterator.next();
                if (selectionKey.isAcceptable()) {
                    //获取新连接的客户端Socket
                    SocketChannel socketChannel = serverSocketChannel.accept();
                    //配置为非阻塞
                    socketChannel.configureBlocking(false);
                    //把客户端Socket 注册进selector 并且监听 读 事件，以及配置服务端缓存区
                    socketChannel.register(selector,SelectionKey.OP_READ, ByteBuffer.allocate(512));
                }else if(selectionKey.isReadable()){
                    //获取发生的读事件的客户端Socket
                    SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                    //获取缓存区
                    ByteBuffer buffer = (ByteBuffer) selectionKey.attachment();
                    buffer.clear();
                    socketChannel.read(buffer);
                    System.out.println(new String(buffer.array()));
                }
                //移除通道，避免重复处理
                iterator.remove();
            }
        }
    }
}

AIO —— Async I/O 异步IO

异步I/O是通过异步处理的方式进行网络I/O访问。这种模式是将用户空间线程作为被动接收者。当内核接收到I/O请求的时候，或者缓存区获取完I/O数据时，会通过一个回调的方式通知用户空间线程进行处理。真正的异步模式需要操作系统底层能够支持。异步IO要求服务线程获得accept请求后，就直接运行下去，知道有回调函数通知到线程以后，再继续处理。由于BSD-Unix和Linux系统并没有底层的异步IO接口可用于socket，因此，实际上都是通过epoll或者kqueue这样多路复用的方式来模拟异步。Linux2.6内核中引入了异步I/O的支持，称为Linux-AIO，不过不支持网络I/O这种方式。这种异步模式分为两步走：

用户通过io_submit()提交I/O请求
过一会再调用io_getevents()来检查events是否已经ready

通过这样的方式就可以写完全异步的I/O程序了。近期的Linux AIO已经可以支持epoll(),从而除了存储I/O,也可以支持网络I/O了。不过由于AIO先天设计上的缺陷，使得这个框架的扩展和演进都非常困难。

io_uring 是 2019 年 Linux 5.1 内核首次引入的高性能异步 I/O 框架。这个框架统一了网路和硬件的异步IO。

Netty NIO的三大组件

Channel & Buffer

Channel指的是一个输入输出的通道，在Netty中常见的有FileChannel、DatagramChannel、SocketChannel和ServerSocketChannel。

FileChannel：用于文件输入输出的通道

DatagramChannel：用于UDP网络编程时使用的通道

SocketChannel和ServerSocketChannel：用于TCP网络传输的通道,前者既可以用于服务端，也可以用于客户端；而后者只用于服务端。

Buffer则是用来缓冲读写数据，常见的buffer是ByteBuffer，以字节为单位缓冲数据。其他还有ShortBuffer，IntBuffer， LongBuffer，FloatBuffer，DoubleBuffer，不过不常用。

ByteBuffer只是一个抽象类，ByteBuffer分为：

MappedByteBuffer，DirectByteBuffer和HeapByteBuffer三类。

Selector

如上面的代码所示，最早BIO程序是一个线程来处理一个socket的数据处理和读写操作。这种方式涉及到系统线程的调度，内存占用高，线程上下文切换成本高，无法支持超高并发的系统。系统本身能够支持的并发线程数量是有限的，而且线程的切换（thread context switch）会有一定的系统开销。当网络并发数较多的时候，就会建立大量的线程。大量线程的创建，销毁和切换会造成非常大的额外系统开销，使得系统响应和处理性能受到严重影响。

改进上述问题的一个方法是预先建立一些线程，并限制线程的总数，这就是所谓的线程池的模式。下图是一个线程池模式的示意图，当有主线程接收到客户端请求时，会新建一个socket，并检查线程池中是否有空闲的线程，如果没有空闲的线程则加入队列中。如果队列也填满，就会执行拒绝策略。线程池适合短连接模式的网络访问，socket中的操作完成后，客户端立刻释放线程。如果客户端连接长期占用线程，则很快就会出现无法处理新连接的情况。这种模式比较适合Web程序的HTTP请求模式。

为了能够支持高并发，长连接的场景，Netty采用了Selector组件监听不同channel的IO事件，这样就能够复用同一个线程进行IO处理了。如下图所示，selector同时监控了多个channel，一旦接收到了某个channel的IO时间，select()方法就会解除阻塞，执行相关的操作。如果没有事件发生，则select()方法将会一直处于阻塞的状态。这样就大大提高了每个线程的利用率，提高了大连接，少量数据传输模式下IO应用的并发能力。这种select模式通常需要通过轮询每个channel的IO事件来实现，底层通过调用操作系统的epoll完成。

Linux上的新型异步I/O框架io_uring

由于Linux AIO的失败，于是再5.1版本后引入了新型的异步框架io_uring。io_uring的构思最初来之于Jens Axboe。io_uring最初只是对于异步模式的一种探索，后来称为了一个和AIO完全不同的异步接口。

io_uring的特点

首先，io_uring是一个真正的异步接口，只要设置了合适的flag，系统调用的时候仅仅将请求加入队列中，而不会涉及到其他的切换，确保了永远都不会阻塞。
支持不同类型的I/O，包括cached files，direct-access和blocking sockets。对于sockets编程更不需要poll+read/write这个步骤，只需提交阻塞读写（blocking read/write),提交完成后就会进入completion ring中。
很强的灵活型和可扩展性，甚至可以用来重写Linux中所有的系统调用。

原理和核心数据结构

每一个io_uring的实例都有两个环形队列（ring），通过mmap在内核层和用户层共享内存。这两个队列分别是：

提交队列：submission queue（SQ）
完成队列：completion queue（CQ）

这两个队列都是单消费者，单生产者的模式。采用无锁的接口，内部使用内存屏障做同步。

使用方式主要有一下几点：

请求
- 应用建立SQ entries（SQE），更新SQ tail；
- 内核消费SQE，更新SQ head。
完成
- 内核为了完成一个或多个请求创建CQ entries（CQE），更新CQ tail；
- 应用程序消费CQE，更新SQ head。
- 完成时间（completion events）可能以任意顺序到达，不过应该与特定的SQE相关联。
- 消费CQE过程无需切换到内核态

批处理I/O请求的提交

我们看到，通过io_uring这样的请求方式是通过批处理进行的，这一点其实和AIO是一样的。不过io_uring将批处理能力扩展到除了storage I/O以外的一些其他的系统调用。比如：

read
write
send
recv
accept
openat
stat
一些专用调用

io_uring实例的三种模式

中断模式（interrupt driven）这是默认的模式，可以通过io_uring_enter() 提交I/O请求，然后检测CQ状态判断是否完成。
轮询模式（polled）Busy-waiting for an I/O completion，而不是通过异步 IRQ（Interrupt Request）接收通知。

这种模式需要文件系统（如果有）和块设备（block device）支持轮询功能。相比中断驱动方式，这种方式延迟更低（都不需要系统调用），但可能会消耗更多 CPU 资源。不过目前只有开了O_DIRECT flag的才可以用这个模式
内核轮询模式（kernel polled）这种模式下会创建一个内核线程（Kernel thread）来执行SQ轮询的工作。使用这种模式的 io_uring 实例， 应用无需切到到内核态 就能触发（issue）I/O 操作。通过 SQ 来提交 SQE，以及监控 CQ 的完成状态，应用无需任何系统调用，就能提交和收割 I/O