BIO、伪异步 IO、AIO和NIO

摘要：采用通信模型的服务端通常由一个独立的线程负责监听客户端的连接它接收到客户端连接请求之后为每个客户端创建一个新的线程进行链路处理处理完成之后通过输出流返回应答给客户端线程销毁这就是典型的一请求一应答通信模型该模型最大的问题就是缺乏弹性伸缩能力

采用 BIO 通信模型的服务端, 通常由一个独立的 Acceptor 线程负责监听客户端的连接, 它接收到客户端连接请求之后为每个客户端创建一个新的线程进行链路处理, 处理完成之后, 通过输出流返回应答给客户端, 线程销毁. 这就是典型的一请求一应答通信模型.

该模型最大的问题就是缺乏弹性伸缩能力, 当客户端并发访问量增加后, 服务端的线程个数和客户端并发访问数是 1:1 的关系.

当线程数过多之后, 系统性能就会下降, 系统也会发生线程堆栈溢出、创建新线程失败等问题, 最终导致进程宕机或者僵死, 不能对外提供服务.

后端通过维护一个消息队列和 N 个活跃线程, 来处理多个客户端的请求接入, 当有新的客户端接入时, 将客户端的 Socket 封装成一个 Task (java.lang.Runnable 接口) 放入后端线的线程池进行处理.

由于线程池可以设置消息队列的大小和最大线程数, 因此它的资源占用是可控的, 无论多少个客户端并发访问, 都不会导致资源耗尽和宕机.

客户端个数 M, 线程池最大线程数 N 的比例关系, 其中 M 可以远远大于 N.

注意: 当对 Socket 的输入流进行读取操作的时候，它会一直阻塞辖区, 直到发生如下三种事件:

有数据可读.

可用数据已经读取完毕.

发生空指针或IO异常.

当对方发送请求或应答消息比较缓慢, 或者网络传输比较慢时, 读取输入流一方的通信线程将被长时间阻塞, 如果对方要 60s 才能将数据发送完成, 读取一方的 IO 线程也将会被同步阻塞 60s, 在此期间, 其它接入消息只能在消息队列中排队.

假如所有的可用线程都被故障服务器阻塞, 那后续所有的 IO 消息都将在队列中排队.

由于线程池采用阻塞队列实现, 当队列积满之后, 后续入队列的操作将被阻塞.

由于前端只有一个 Accptor 线程接收客户端接入, 它被阻塞在线程池的同步阻塞队列之后, 新的客户端请求消息将被拒绝, 客户端会发生大量的连接超时.

与 Socket 类和 ServerSocket 类相对应, NIO 也提供了 SocketChannel 和 ServerSocketChannel 两种不同的套接字通道实现. 这两种新增的通道都支持阻塞和非阻塞两种模式.

一般来说, 低负载、低并发的应用程序可以选择同步阻塞IO以降低编程复杂度; 对于高负载、高并发的网络模型应用, 需要使用NIO的非阻塞模式进行开发.

Buffer 是一个对象, 它包含一些要写入或要读出的数据. 在NIO库中, 所有数据都是用缓冲区处理的. 在读取数据时, 它是直接读取到缓冲区中的; 在写入数据时, 写入到缓冲区中. 任何时候访问NIO中的数据, 都是通过缓冲区进行操作的.

缓冲区实质上是一个数组. 通常它是一个字节数组, 也可以使用其他种类的数组. 但是一个缓冲区不仅仅是一个数组, 缓冲区提供了对数据的结构化访问以及维护读写位置等信息.

常用缓冲区是 ByteBuffer, 一个 ByteBuffer 提供了一种功能用于操作 byte 数组. 除了 ByteBuffer, 还有其他的一些缓冲区.

ByteBuffer: 字节缓冲区

CharBuffer: 字符缓冲区

ShortBuffer: 短整形缓冲区

IntBuffer: 整形缓冲区

LongBuffer: 长整形缓冲区

FloatBuffer: 浮点型缓冲区

DoubleBuffer: 双精度浮点型缓冲区

每一个 Buffer 类都是 Buffer 接口的一个子实例. 除了 ByteBuffer, 每个 Buffer 类都有完全一样的操作, 只是它们所处理的类型不一样.

Channel 是一个通道, 网络数据通过 Channel 读取和写入. 通道与流的不同之处在于通道是双向的, 流只是在一个方向上移动(一个流必须是 InputStream 或者 OutputStream), 而通道可以用于读、写或者二者同时进行.

FileChannel: 用于文件的数据读写

DatagramChannel: 用于UDP的数据读写

SocketChannel: 用于TCP的数据读写. 一般是客户端实现

ServerSocketChannel: 允许我们监听TCP链接请求, 每个请求会创建会一个SocketChannel. 一般是服务器实现

多路复用器提供选择已经就绪的任务的能力. 简单来讲, Selector 会不断的轮询注册在其上的 Channel, 如果某个 Channel 上面发生读或写事件, 这个 Channel 就处于就绪状态, 会被 Selector 轮询出来, 然后通过 SelectionKey 可以获取就绪 Channel 的集合, 进行后续的 IO 操作.

一个多路复用器 Selector 可以同时轮询多个 Channel, 由于 JDK 使用了 epoll() 代替传统的 select 实现, 所以它并没有最大连接句柄 1024/2048 的限制. 这也意味着只需要一个线程负责 Selector 的轮询, 就可以接入成千上万的客户端.

public class MultiplexerTimeServer implements Runnable {

    private Selector selector;

    private ServerSocketChannel servChannel;

    private volatile boolean stop;

    /**
     * 初始化多路复用器、绑定监听端口
     *
     * @param port
     */
    public MultiplexerTimeServer(int port) {
        try {
            // 创建多路复用器
            selector = Selector.open();
            
            // 打开 ServerSocketChannel 用来监听客户端的连接, 它是所有客户端连接的父管道.
            servChannel = ServerSocketChannel.open();
            
            // 设置 ServerSocketChannel 为异步非阻塞模式
            servChannel.configureBlocking(false);
            
            // 绑定地址和端口
            servChannel.socket().bind(new InetSocketAddress(port), 1024);
            
            // 将 ServerSocketChannel 注册到 Reactor 线程的多路复用器 Selector 上, 监听 ACCEPT 事件
            servChannel.register(selector, SelectionKey.OP_ACCEPT);
            System.out.println("The time server is start in port : " + port);
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(1);
        }
    }

    public void stop() {
        this.stop = true;
    }

    /*
     * (non-Javadoc)
     *
     * @see java.lang.Runnable#run()
     */
    @Override
    public void run() {
        while (!stop) {
            try {
                selector.select(1000);
                Set selectedKeys = selector.selectedKeys();
                Iterator it = selectedKeys.iterator();
                SelectionKey key = null;
                while (it.hasNext()) {
                    key = it.next();
                    it.remove();
                    try {
                        handleInput(key);
                    } catch (Exception e) {
                        if (key != null) {
                            key.cancel();
                            if (key.channel() != null) {
                                key.channel().close();
                            }
                        }
                    }
                }
            } catch (Throwable t) {
                t.printStackTrace();
            }
        }

        // 多路复用器关闭后，所有注册在上面的Channel和Pipe等资源都会被自动去注册并关闭，所以不需要重复释放资源
        if (selector != null) {
            try {
                selector.close();
            } catch (IOException e) {
                e.printStackTrace();
            } 
        }
            
    }

    private void handleInput(SelectionKey key) throws IOException {

        if (key.isValid()) {
            // 处理新接入的请求消息
            if (key.isAcceptable()) {
                // Accept the new connection
                ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
                SocketChannel sc = ssc.accept();
                sc.configureBlocking(false);
                // Add the new connection to the selector
                sc.register(selector, SelectionKey.OP_READ);
            }
            if (key.isReadable()) {
                // Read the data
                SocketChannel sc = (SocketChannel) key.channel();
                ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                int readBytes = sc.read(readBuffer);
                if (readBytes > 0) {
                    readBuffer.flip();
                    byte[] bytes = new byte[readBuffer.remaining()];
                    readBuffer.get(bytes);
                    String body = new String(bytes, "UTF-8");
                    System.out.println("The time server receive order : "
                    + body);
                    String currentTime = "QUERY TIME ORDER"
                .equalsIgnoreCase(body) ? new java.util.Date(
                            System.currentTimeMillis()).toString()
                            : "BAD ORDER";
                    doWrite(sc, currentTime);
                } else if (readBytes < 0) {
                    // 对端链路关闭
                    key.cancel();
                    sc.close();
                } else {
                    ; // 读到0字节，忽略
                }
                
            }
        }
    }

    private void doWrite(SocketChannel channel, String response)
            throws IOException {
        if (response != null && response.trim().length() > 0) {
            byte[] bytes = response.getBytes();
            ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
            writeBuffer.put(bytes);
            writeBuffer.flip();
            channel.write(writeBuffer);
        }
    }
}

selector.select(1000); 休眠时间为1S, 无论是否有读写等事件发生, selector 每隔 1S 都被唤醒一次, selector 也提供了一个无参的 select 方法. 当有处于就绪状态的 Channel 时, selector 将返回就绪状态的 Channel 的 SelectionKey 集合, 我们通过对就绪状态的 Channel 集合进行迭代, 就可以进行网络的异步读写操作.

key.isAcceptable() 来判断 当前 SelectionKey 的通道是否已准备好接受新的套接字连接(处理新接入的客户端请求消息). 通过 ServerSocketChannel 的 accept 接收客户端的连接请求并创建 SocketChannel 实例, 完成上述操作后, 相当于完成了TCP的三次握手, TCP物理链路正式建立. 注意，我们需要将新创建的 SocketChannel 设置为异步非阻塞, 同时也可以对其TCP参数进行设置, 例如TCP接收和发送缓冲区的大小等.

根据 SelectionKey 的操作位进行判断即可获知网络事件的类型, 如 isAcceptable() 表示为 OP_ACCEPT 

key.isAcceptable() 来判断 当前 SelectionKey 的通道是否已准备好进行读取(读取客户端的请求消息). 首先创建一个 ByteBuffer, 由于我们事先无法得知客户端发送的码流大小, 作为例程, 我们开辟一个1M的缓冲区. 然后调用 SocketChannel 的 read 方法读取请求码流, 注意, 由于我们已经将 SocketChannel 设置为异步非阻塞模式, 因此它的 read 是非阻塞的. 使用返回值进行判断, 看读取到的字节数, 返回值有三种可能的结果:

返回值大于0: 读到了字节, 对字节进行编解码;

返回值等于0: 没有读取到字节, 属于正常场景, 忽略;

返回值为-1: 链路已经关闭, 需要关闭SocketChannel, 释放资源.

当读取到码流以后, 我们进行解码, 首先对 readBuffer 进行 flip 操作, 它的作用是将缓冲区当前的limit 设置为 position, position 设置为0, 用于后续对缓冲区的读取操作.

然后根据缓冲区可读的字节个数创建字节数组, 调用 ByteBuffer 的 get 操作将缓冲区可读的字节数组拷贝到新创建的字节数组中, 最后调用字符串的构造函数创建请求消息体并打印. 如果请求指令是 ”QUERY TIME ORDER” 则把服务器的当前时间编码后返回给客户端, 下面我们看看如果异步发送应答消息给客户端.

doWrite 方法将消息异步发送给客户端, 首先将字符串编码成字节数组, 根据字节数组的长度创建 ByteBuffer, 调用 ByteBuffer 的 put 操作将字节数组拷贝到缓冲区中, 然后对缓冲区进行flip操作, 最后调用 SocketChannel 的 write 方法将缓冲区中的字节数组发送出去.

需要指出的是, 由于 SocketChannel 是异步非阻塞的, 它并不保证一次能够把需要发送的字节数组发送完, 此时会出现“写半包”问题, 我们需要注册写操作, 不断轮询 Selector 将没有发送完的 ByteBuffer 发送完毕, 可以通过 ByteBuffer 的 hasRemain() 方法判断消息是否发送完成.

NIO 2.0 引入了新的异步通道的概念, 并提供了异步文件通道和异步套接字通道的实现. 异步通道提供以下两种方式获取操作结果.

通过 java.util.concurrent.Future 类来表示异步操作的结果.

在执行异步操作的时候传入一个 java.nio.channels

CompletionHandler 接口的实现类作为操作完成的回调.

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