Java进阶知识点:服务端高并发的基石 - NIO与Reactor AIO与Proactor

Posted 2020-12-29 爱养花的码农

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Java进阶知识点:服务端高并发的基石 - NIO与Reactor AIO与Proactor相关的知识，希望对你有一定的参考价值。

一、背景

要提升服务器的并发处理能力，通常有两大方向的思路。

1、系统架构层面。比如负载均衡、多级缓存、单元化部署等等。

2、单节点优化层面。比如修复代码级别的性能Bug、JVM参数调优、IO优化等等。

一般来说，系统架构的合理程度，决定了系统在整体性能上的伸缩性（高伸缩性，简而言之就是可以很任性，性能不行就加机器，加到性能足够为止）；而单节点在性能上的优化程度，决定了单个请求的时延，以及要达到期望的性能，所需集群规模的大小。两者双管齐下，才能快速构建出性能良好的系统。

今天，我们就聊聊在单节点优化层面最重要的IO优化。之所以IO优化最重要，是因为IO速度远低于CPU和内存，而不够良好的软件设计，常常导致CPU和内存被IO所拖累，如何摆脱IO的束缚，充分发挥CPU和内存的潜力，是性能优化的核心内容。

而CPU和内存又是如何被IO所拖累的呢？这就从Java中几种典型的IO操作模式说起。

二、Java中的典型IO操作模式

2.1 同步阻塞模式

Java中的BIO风格的API，都是该模式，例如：

Socket socket = getSocket();
socket.getInputStream().read(); //读不到数据誓不返回

该模式下，最直观的感受就是如果IO设备暂时没有数据可供读取，调用API就卡住了，如果数据一直不来就一直卡住。

2.2 同步非阻塞模式

Java中的NIO风格的API，都是该模式，例如：

SocketChannel socketChannel = getSocketChannel(); //获取non-blocking状态的Channel
socketChannel.read(ByteBuffer.allocate(4)); //读不到数据就算了，立即返回0告诉你没有读到

该模式下，通常需要不断调用API，直至读取到数据，不过好在函数调用不会卡住，我想继续尝试读取或者先去做点其他事情再来读取都可以。

2.3 异步非阻塞模式

Java中的AIO风格的API，都是该模式，例如：

AsynchronousSocketChannel asynchronousSocketChannel = getAsynchronousSocketChannel();
asynchronousSocketChannel.read(ByteBuffer.allocate(4), null, new CompletionHandler<Integer, Object>() {
    @Override
    public void completed(Integer result, Object attachment) {
        //读不到数据不会触发该回调来烦你，只有确实读取到数据，且把数据已经存在ByteBuffer中了，API才会通过此回调接口主动通知您
    }
    @Override
    public void failed(Throwable exc, Object attachment) {
    }
});

该模式服务最到位，除了会让编程变的相对复杂以外，几乎无可挑剔。

2.4 小结

对于IO操作而言，同步和异步的本质区别在于API是否会将IO就绪（比如有数据可读）的状态主动通知你。同步意味着想要知道IO是否就绪，必须发起一次询问，典型的一问一答，如果回答是没有就绪，那你还得自己不断询问，直到答案是就绪为止。异步意味着，IO就绪后，API将主动通知你，无需你不断发起询问，这通常要求调用API时传入通知的回调接口。

阻塞和非阻塞的本质区别在于IO操作因IO未就绪不能立即完成时，API是否会将当前线程挂起。阻塞意味着API会一直等待IO就绪后，完成本次IO操作才返回，在此之前调用该API的用户线程将一直挂起，无法进行其他计算处理。非阻塞意味着API会立即返回，而不是等待IO就绪，用户可以立即再次获得线程的控制权，可以使用该线程进行其他计算处理。

那有没有异步阻塞模式呢？如果API支持异步，相当于API说：“你玩去吧，我准备好了通知你”，但是你还是傻乎乎地不去玩，原地等待API做完后的通知。这通常是因为本次IO操作很重要，拿不到结果业务流程根本无法继续，所以为了编程上的简单起见，还是乖乖等吧。可见异步阻塞模式更多的是出于业务流程控制和简化编码难度的考虑，由业务代码自主形成的，Java语言不会特别为你准备异步阻塞IO的API。

三、分离快与慢

3.1 BIO的局限

CPU和内存是高速设备，磁盘、网络等IO设备是低速设备，在Java编程语言中，对CPU和内存的使用被抽象为对线程、栈、堆的使用，对IO设备的使用被抽象为IO相关的API调用。

显然，如果使用BIO风格的IO API，由于其同步阻塞特性，会导致IO设备未就绪时，线程挂起，该线程无法继续使用CPU和内存，直至IO就绪。由于IO设备的速度远低于CPU和内存，所以使用BIO风格的API时，有极大的概率会让当前线程长时间挂起，这就形成了CPU和内存资源被IO所拖累的情况。

作为服务端应用，会面临大量客户端向服务端发起连接请求的场景，每个连接对服务端而言，都意味着需要进行后续的网络IO读取，IO读取完成后，才能获得完整的请求内容，进而才能再进行一些列相关计算处理获得请求结果，最后还要将结果通过网络IO回写给客户端。使用BIO的编码风格，通常是同一个线程全程负责一个连接的IO读取、数据处理和IO回写，该线程绝大部分时间都可能在等待IO就绪，只有极少时间在真正利用CPU资源。

而此时服务器要想同时处理大量客户端连接，后端就同时开启与并发连接数量相应的线程。线程是操作系统的宝贵资源，而且每开启一个操作系统线程，Java还会消耗-Xss指定的线程堆栈大小的堆外内存，如果同时存在大量线程，操作系统调度线程的开销也会显著增加，导致服务器性能快速下降。所以此时服务器想要支持上万乃至几十万的高并发连接，可谓难上加难。

3.2 NIO的突破

3.2.1 突破思路

由于NIO的非阻塞特性，决定了IO未就绪时，线程可以不必挂起，继续处理其他事情。这就为分离快与慢提供了可能，高速的CPU和内存可以不必苦等IO交互，一个线程也不必局限于只为一个IO连接服务。这样，就让用少量的线程处理海量IO连接成为了可能。

3.2.2 思路落地

虽然我们看到了曙光，但是要将这个思路落地还需解决掉一些实际的问题。

a）当IO未就绪时，线程就释放出来，转而为其他连接服务，那谁去监控这个被抛弃IO的就绪事件呢？

b）IO就绪了，谁又去负责将这个IO分配给合适的线程继续处理呢？

为了解决第一个问题，操作系统提供了IO多路复用器（比如Linux下的select、poll和epoll），Java对这些多路复用器进行了封装（一般选用性能最好的epoll），也提供了相应的IO多路复用API。NIO的多路复用API典型编程模式如下：

// 开启一个ServerSocketChannel，在8080端口上监听
ServerSocketChannel server = ServerSocketChannel.open();
server.bind(new InetSocketAddress("0.0.0.0", 8080));
// 创建一个多路复用器
Selector selector = Selector.open();
// 将ServerSocketChannel注册到多路复用器上，并声明关注其ACCEPT就绪事件
server.register(selector, SelectionKey.OP_ACCEPT);
while (selector.select() != 0) {
    // 遍历所有就绪的Channel关联的SelectionKey
    Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
    while (iterator.hasNext()) {
        SelectionKey key = iterator.next();
        // 如果这个Channel是READ就绪
        if (key.isReadable()) {
            // 读取该Channel
            ((SocketChannel) key.channel()).read(ByteBuffer.allocate(10));
        }
        if (key.isWritable()) {
            //... ...
        }
        // 如果这个Channel是ACCEPT就绪
        if (key.isAcceptable()) {
            // 接收新的客户端连接
            SocketChannel accept = ((ServerSocketChannel) key.channel()).accept();
            // 将新的Channel注册到多路复用器上，并声明关注其READ/WRITE就绪事件
            accept.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);
        }
        // 删除已经处理过的SelectionKey
        iterator.remove();
    }
}

IO多路复用API可以实现用一个线程，去监控所有IO连接的IO就绪事件。

第二个问题在上面的代码中其实也得到了“解决”，但是上面的代码是使用监控IO就绪事件的线程来完成IO的具体操作，如果IO操作耗时较大（比如读操作就绪后，有大量数据需要读取），那么会导致监控线程长时间为某个具体的IO服务，从而导致整个系统长时间无法感知其他IO的就绪事件并分派IO处理任务。所以生产环境中，一般使用一个Boss线程专门用于监控IO就绪事件，一个Work线程池负责具体的IO读写处理。Boss线程检测到新的IO就绪事件后，根据事件类型，完成IO操作任务的分配，并将具体的操作交由Work线程处理。这其实就是Reactor模式的核心思想。

3.2.3 Reactor模式

如上所述，Reactor模式的核心理念在于：

a）依赖于非阻塞IO。

b）使用多路复用器监管海量IO的就绪事件。

c）使用Boss线程和Work线程池分离IO事件的监测与IO事件的处理。

Reactor模式中有如下三类角色：

a）Acceptor。用户处理客户端连接请求。Acceptor角色映射到Java代码中，即为SocketServerChannel。

b）Reactor。用于分派IO就绪事件的处理任务。Reactor角色映射到Java代码中，即为使用多路复用器的Boss线程。

c）Handler。用于处理具体的IO就绪事件。（比如读取并处理数据等）。Handler角色映射到Java代码中，即为Worker线程池中的每个线程。

Acceptor的连接就绪事件，也是交由Reactor监管的，有些地方为了分离连接的建立和对连接的处理，为将Reactor分离为一个主Reactor，专门用户监管连接相关事件（即SelectionKey.OP_ACCEPT），一个从Reactor，专门用户监管连接上的数据相关事件（即SelectionKey.OP_READ 和SelectionKey.OP_WRITE）。

关于Reactor的模型图，网上一搜一大把，我就不献丑了。相信理解了它的核心思想，图自然在心中。关于Reactor模式的应用，可以参见著名NIO编程框架Netty，其实有了Netty之后，一般都直接使用Netty框架进行服务端NIO编程。

3.3 AIO的更进一步

3.3.1 AIO得天独厚的优势

你很容易发现，如果使用AIO，NIO突破时所面临的落地问题似乎天然就不存在了。因为每一个IO操作都可以注册回调函数，天然就不需要专门有一个多路复用器去监听IO就绪事件，也不需要一个Boss线程去分配事件，所有IO操作只要一完成，就天然会通过回调进入自己的下一步处理。

而且，更让人惊喜的是，通过AIO，连NIO中Work线程去读写数据的操作都可以省略了，因为AIO是保证数据真正读取/写入完成后，才触发回调函数，用户都不必关注IO操作本身，只需关注拿到IO中的数据后，应该进行的业务逻辑。

简而言之，NIO的多路复用器，是通知你IO就绪事件，AIO的回调是通知你IO完成事件。AIO做的更加彻底一些。这样在某些平台上也会带来性能上的提升，因为AIO的IO读写操作可以交由操作系统内核完成，充分发挥内核潜能，减少了IO系统调用时用户态与内核态间的上下文转换，效率更高。

（不过遗憾的是，Linux内核的AIO实现有很多问题（不在本文讨论范畴），性能在某些场景下还不如NIO，连Linux上的Java都是用epoll来模拟AIO，所以Linux上使用Java的AIO API，只是能体验到异步IO的编程风格，但并不会比NIO高效。综上，Linux平台上的Java服务端编程，目前主流依然采用NIO模型。）

使用AIO API典型编程模式如下：

//创建一个Group，类似于一个线程池，用于处理IO完成事件
AsynchronousChannelGroup group = AsynchronousChannelGroup.withCachedThreadPool(Executors.newCachedThreadPool(), 32);
//开启一个AsynchronousServerSocketChannel，在8080端口上监听
AsynchronousServerSocketChannel server = AsynchronousServerSocketChannel.open(group);
server.bind(new InetSocketAddress("0.0.0.0", 8080));
//接收到新连接
server.accept(null, new CompletionHandler<AsynchronousSocketChannel, Object>() {
    //新连接就绪事件的处理函数
    @Override
    public void completed(AsynchronousSocketChannel result, Object attachment) {
        result.read(ByteBuffer.allocate(4), attachment, new CompletionHandler<Integer, Object>() {
            //读取完成事件的处理函数
            @Override
            public void completed(Integer result, Object attachment) {
            }

            @Override
            public void failed(Throwable exc, Object attachment) {
            }
        });
    }
    @Override
    public void failed(Throwable exc, Object attachment) {
    }
});