Java NIO与Reactor模式

2020-04-12 本文已影响0人挪威的senlin

NIO(Non-blocking I/O)，非阻塞的I/O模型。自java1.4版本之后推出，为什么在已有I/O库之后Java还会推出NIO，以及NIO的原理是什么和基于NIO的应用是本文介绍的内容。

BIO与NIO

为什么JAVA要有了IO系统之后还要再推出一套NIO目的是什么？首先要先明白一个概念

同步/异步：同步就是下一步操作必须等待当前操作完成才能执行，而异步则无需等待当前操作是否完成，调用之后就可以进行下一步操作，通常通过事件、回调的形式来通知执行的结果。
阻塞/非阻塞：阻塞就是获取一个结果的时候如果没有完成将当前线程挂起等待，而非阻塞是如果结果没有返回也不会阻塞当前线程。

    public static void main(String[] args) throws  InterruptedException, ExecutionException {
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        //同步
        queryDB();
        //异步
        Future<?> future = executorService.submit(() -> queryDB());
        //判断是否执行完成 非阻塞
        future.isDone();
        //获取执行结果 阻塞
        future.get();
    }

    private static void  queryDB(){
        try {
            Thread.sleep(3000);
        }catch (InterruptedException e){
            e.printStackTrace();
        }
        System.out.println("查询数据库完毕");
    }

比如上面一段代码，queryDB()这一步是同步执行，当前线程会等待queryDB()返回，而将任务提交给线程池这一步是异步执行，当前线程无需等待任务的执行结果。future.isDone()这是一个非阻塞的操作，不管任务执行成功都会返回结果，不会阻塞当前线程。future.get()如果任务没有执行完成，会阻塞当前线程知道任务执行完成返回结果。

传统的java.io包和java.net提供的部分api比如Socket、ServerScoket下都是同步阻塞的被称为BIO,java.nio报下的Buffer、Selector、Channel等关键组件提供多路复用的、同步非阻塞IO程序被称为NIO。

BIO Server

使用传统I/O编程

public class IOServer {
    public static void main(String[] args) throws Exception {

        ServerSocket serverSocket = new ServerSocket(8000);

        // (1) 接收新连接线程
        new Thread(() -> {
            while (true) {
                try {
                    // (1) 阻塞方法获取新的连接
                    Socket socket = serverSocket.accept();

                    // (2) 每一个新的连接都创建一个线程，负责读取数据
                    new Thread(() -> {
                        try {
                            int len;
                            byte[] data = new byte[1024];
                            InputStream inputStream = socket.getInputStream();
                            // (3) 按字节流方式读取数据
                            while ((len = inputStream.read(data)) != -1) {
                                System.out.println(new String(data, 0, len));
                            }
                        } catch (IOException e) {
                        }
                    }).start();

                } catch (IOException e) {
                }

            }
        }).start();
    }
}

BIO模型

可以看出这种编程方式简单，易懂，但是上面的代码问题在于没一个客户端连接都需要新增加一个线程去处理。我们都知道线程的资源是很珍贵的，如果有大量的客户端连接服务器，线程资源是远远不够的，而且用于处理连接的线程在进行读和写操作的时候都是处于同步阻塞状态，是对资源的一种浪费。

NIO Server

使用NIO编写服务端

public class NIOServer {
    public static void main(String[] args) throws Exception {
        // 创建Selector和Channel
        try (Selector selector = Selector.open(); ServerSocketChannel serverSocket = ServerSocketChannel.open();) {
            serverSocket.bind(new InetSocketAddress(InetAddress.getLocalHost(), 8888));
            // 非阻塞注册到Selector
            serverSocket.configureBlocking(false);
            //监听连接事件
            serverSocket.register(selector, SelectionKey.OP_ACCEPT);
            while (true) {
                // 阻塞等待就绪的Channel
                selector.select();
                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                Iterator<SelectionKey> iter = selectedKeys.iterator();
                while (iter.hasNext()) {
                    SelectionKey key = iter.next();
                    ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
                    //数据处理
                    handleClient(serverSocketChannel);
                    iter.remove();
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private static void handleClient(ServerSocketChannel serverSocketChannel) throws Exception {
        SocketChannel clientChannel = serverSocketChannel.accept();
        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
        //面向 Buffer
        clientChannel.read(byteBuffer);
        byteBuffer.flip();
        System.out.println(Charset.defaultCharset().newDecoder().decode(byteBuffer)
                .toString());
    }
}

NIO模型

开启Selector，相当于调度员的角色。
创建一个 ServerSocketChannel，并且向 Selector 注册，通过指定 SelectionKey.OP_ACCEPT，告诉调度员，它关注的是新的连接请求。
Selector 阻塞在 select 操作，当有 Channel 发生接入请求，就会被唤醒。
基于Buffer处理数据。

上面两种模式对比，BIO是面向直接流，读取完之后无法再次读取。输入和输出是两个不同的定义。而NIO面向的是Buffer，可以随意的读取里面任意的位置的字节，而且输入输出可以对应同一个Buffer。BIO是同步阻塞的处理连接和读取数据，而NIO引入多路I/O复用的机制，可以非阻塞的监听多个连接，检查是否有连接事件、获取读取事件发生，是以事件为驱动。这种模型提高了服务器的处理性能，和节省不少资源。

NIO

NIO有三个关键角色，Selector、Channel、Buffer，Selector类似于调度员的角色，管理多个I/O事件，Channel是对I/O操作的一个抽象，可以对同一个Channel进行读取和写入操作。Buffer可以看做是数据的容器，可以对buffer的任意位置的数据进行操作。他们之间的关系如下图所示:

对应关系

Selector

介绍selector之前先了解几中创建的I/O模型。

I/O模型

selector是基于操作系统所提供的多路I/O复用机制所实现，比如linux的epoll。

Channel

Channel是对I/O操作事件的抽象，可以使文件的读写，网络的传输等。所有对Channel数据的读写都是通过Buffer来进行的。

Buffer

1. 原理
Buffer低层是一个数组存放数据，有几个关键的变量实现可以对Buffer数据的写入和读取。

capacity：buffer所能存放元素的最大数量。
limit：第一个不可以读或写的位置，大小不能超过capacity。
position ：下一个读或写的位置。

buffer

代码示例，这个几个变量在buffer初始化、写入和读取的变化。

    public static void main(String[] args) {
        IntBuffer intBuffer = IntBuffer.allocate(10);
        //初始化 capacity和limit都是10 position为0
        System.out.println("init capacity：" + intBuffer.capacity() + "\t limit：" + intBuffer.limit() + "\t position：" + intBuffer.position());
        intBuffer.put(1);
        intBuffer.put(2);
        //放入两个元素 capacity和limit都是10 position为2
        System.out.println("after put capacity：" + intBuffer.capacity() + "\t limit：" + intBuffer.limit() + "\t position：" + intBuffer.position());
        intBuffer.flip();
        //flip capacity为10 limit为2 position为0
        System.out.println("after flip capacity：" + intBuffer.capacity() + "\t limit：" + intBuffer.limit() + "\t position：" + intBuffer.position());
        while (intBuffer.hasRemaining()) {
            System.out.println("fetch data :" + intBuffer.get());
            //capacity为10 limit为2 position每读取一次增加一次
            System.out.println("after get capacity：" + intBuffer.capacity() + "\t limit：" + intBuffer.limit() + "\t position：" + intBuffer.position());
        }
    }

执行结果

init capacity：10     limit：10    position：0
after put capacity：10    limit：10    position：2
after flip capacity：10   limit：2     position：0
fetch data :1
after get capacity：10    limit：2     position：1
fetch data :2
after get capacity：10    limit：2     position：2

又上面的代码可以知道capacity在buffer初始化之后是不会改变的，position在每次读写操作之后都会增加1，当position和limit相等时，就不能对buffer进行读或写的操作了。limit在初始化的时候和capacity相等，每次读写操作也不会改变limit的数值，在调用flip方法之后limit和position值相等。在每次写入数据之后，都必须调用filp方法才能对buffer进行数据读取。这也是buffer能够同时实现对数据的读写的关键方法。下面来看它的实现：

    public final Buffer flip() {
        limit = position;
        position = 0;
        mark = -1;
        return this;
    }

非常简单，就是把limit指向position的位置，然后把position置为0，就可以从buffer中读取数据了。这也是可以对同一个Channel实现读写功能的关键。
2. DirectBuffer
Buffer还有一个及其重要的概念DirectBuffer（直接缓存区），java的对象绝大多数都是分配在堆上的，而DirectBuffer是分配在虚拟机之外的内存区域，直接和操作系统打交道。为什么需要使用DirectBuffer它和HeapBuffer的区别是什么？这里要先明白几个概念。

内核态与用户态
当一个任务（进程）执行系统调用而陷入内核代码中执行时，称进程处于内核运行态（内核态）。此时处理器处于特权级最高的（0级）内核代码中执行。当进程处于内核态时，执行的内核代码会使用当前进程的内核栈。每个进程都有自己的内核栈。
当进程在执行用户自己的代码时，则称其处于用户运行态（用户态）。此时处理器在特权级最低的（3级）用户代码中运行。当正在执行用户程序而突然被中断程序中断时，此时用户程序也可以象征性地称为处于进程的内核态。因为中断处理程序将使用当前进程的内核栈。
普通I/O操作

普通I/O操作

一次普通的文件读写操作经过了一下几个步骤：
• 用户向操作系统发出read()指令，从用户态切换到内核态。
• 系统请求磁盘文件，将数据复制到内核缓存中。
• 从内核态切换到用户态，将内核缓存中的数据复制用户缓存中。
•从用户态切换到内核态，将数据写入socket缓存中，向网络中发送数据。

这些步骤一共会有三次次的形态切换，和三次的数据拷贝。

零拷贝

零拷贝

使用零拷贝经过以下几个步骤：
• 用户向操作系统发出sendfile()命令，从用户态切换到内核态。
•系统请求磁盘文件，将数据复制到内核缓存中。
•将数据写入socket缓存中，向网络中发送数据。(linux提供scatter/gather的功能，也无需将内核缓存的数据全部复制到socket缓存中，只需要将文件描述符写入socket缓存即可完成数据的拷贝)

可以看见零拷贝只是一个相对于用户程序来说是零拷贝，而在操作系统层面还是存在数据拷贝的。零拷贝对比上面的普通I/O操作减少了操作系统形态的切换和数据拷贝的次数，提升了时间和减少了内存使用。
HeapBuffer就就普通的I/O的流程，需要将内存的的数据拷贝到java堆中，然后再进行数据操作，完成之后再将数据拷贝到内存之后。而DirectBuffer映射的就是内存的数据，无需进行数据的拷贝就可以操作内存中的数据。

Reactor

NIO提供的多路I/O复用的功能，能够基于这个机制做什么呢。这里我将基于大师Doug Lea的《Scalable IO in Java》分析，要想彻底的了解Reactor和NIO这边文章推荐大家一定要去看。
1. Classic Service Designs

image.png
传统的服务架构，和之前分析的BIO一样每个客户端对应一个线程处理。

2. Basic Reactor Design

image.png
由一个Reactor的监听IO事件，当检测到一个新的事件；新的事件包含连接建立就绪、读就绪、写就绪等，然后将监听到的事件分发出去给对应handler处理，基于事件驱动。

3. Worker Thread Pools

image.png
Reactor角色只关心是否有事件发生比如连接事件、读事件、写事件，然后将事件快速分发处理，让对应的handler去处理，将对数据的处理也让Reactor来做将会拖慢整个流程所以把对数据的处理交给线程池来做。

4. Using Multiple Reactors

image.png

多个Reactor角色，这里将职责划分的更为细致。mainReactor负责监听连接事件，subReactor来监听读写事件而具体对数据怎么处理则交给线程池。

应用

netty作为高性能的网络框架，很多框架都是要netty作为底层通讯框架比如dubbo、RocketMQ。netty底层就是基于NIO实现，并且看过netty的学习demo的，可以知道netty是标准的Reactor模式。服务器启动有两个事件循环，bossGroup负责连接事件，接受到新的连接之后将它转给workerGroup进行处理，模式大致和Multiple Reactors大致一样,bossGroup对应mainReactor,workerGroup对应subReactor。

public class NettyServer {

    public static void main(String[] args) {
        //负责连接
        NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
        //负责数据读写
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();

        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap
                .group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    protected void initChannel(NioSocketChannel ch) {
                    }
                });

        serverBootstrap.bind(8000);
    }
}

本文是自己学习netty时的一些自己的见解，有不正确的地方希望大家指出。