RPC框架入门篇（1）之【BIO，NIO，AIO】

---

前段时间在看dubbo的源码，看的差很少了也开始在写一个RPC框架，如今写的快一半了，才想起来怎么按部就班的经过文章的方式跟你们聊这个东西。因而思来想去，决定先从最基础的服务间网络通讯提及比较好，后面再慢慢的跟你们引出怎么去写一个RPC框架。

本篇主要跟你们聊下网咯I/O，主要是针对初学者的由浅入深系列。

传统的BIO通讯弊端

---

传统的BIO通讯当接收到客户端的请求时，为每个请求建立一个新的线程进行链路处理，处理完成以后，经过输出流返回给客户端，而后线程销毁。

这种模型的弊端就是当并发数上涨之后，server端的线程也跟着线性增加，会带来服务性能的急剧降低，而且可能会发生线程堆栈溢出，从而致使不能对外提供服务。

伪异步IO模型

---

在传统BIO模型的基础上，用线程池来处理客户端的请求，防止高并发致使的server端资源被耗尽问题。

伪异步IO的缺点

不管是BIO仍是伪异步本质上都是阻塞型I/O，都是基于Stream进行网络数据的读和写。首先咱们看下InputStream的read方法源码：

/**
     * Reads the next byte of data from the input stream. The value byte is
     * returned as an <code>int</code> in the range <code>0</code> to
     * <code>255</code>. If no byte is available because the end of the stream
     * has been reached, the value <code>-1</code> is returned. This method
     * blocks until input data is available, the end of the stream is detected,
     * or an exception is thrown.
     *
     * <p> A subclass must provide an implementation of this method.
     *
     * @return     the next byte of data, or <code>-1</code> if the end of the
     *             stream is reached.
     * @exception  IOException  if an I/O error occurs.
     */
    public abstract int read() throws IOException;
复制代码

经过注释能够知道，对Socket的输入流进行读取的时候，会一直发生阻塞，直到如下3种状况：

• 有数据可读
• 可用数据已经读取完毕
• 发生空指针或者异常

意味着当对方发送请求或者应答消息比较缓慢的时候，或者网络传输比较慢的时候，读取输入流一方的通讯线程将被长时间阻塞。在此期间，后面的请求都得排队。

在继续看下Outputtream的write方法：

/**
     * Writes the specified byte to this output stream. The general
     * contract for <code>write</code> is that one byte is written
     * to the output stream. The byte to be written is the eight
     * low-order bits of the argument <code>b</code>. The 24
     * high-order bits of <code>b</code> are ignored.
     * <p>
     * Subclasses of <code>OutputStream</code> must provide an
     * implementation for this method.
     *
     * @param      b   the <code>byte</code>.
     * @exception  IOException  if an I/O error occurs. In particular,
     *             an <code>IOException</code> may be thrown if the
     *             output stream has been closed.
     */
    public abstract void write(int b) throws IOException;
复制代码

当调用write写输出流的时候，会发生阻塞，直到全部要发送的字节所有写入完毕，或者发生异常。切换为从TCP/IP角度来理解，当消息的接收方处理比较缓慢，不能及时的从TCP缓冲区读取数据，这会致使发送方的TCP``window size不断缩小，直到为0，双方处于Keep-Alive状态，消息发送方就不能在继续像TCP缓冲区写入消息，若是采用的是同步阻塞I/O，write将会被无限期阻塞，直到window size大于0或者发生I/O异常。

所以使用阻塞I/O的Socket和ServerSocket在生产使用问题不少，所以NIO诞生了，对应的是SocketChannel和ServerSocketChannel两个类。

NIO编程介绍

---

NIO相关概念

Buffer

传统的BIO主要是面向流的，能够将数据直接写入或者读取到Stream对象中；而在NIO中，读取和写入数据都是在缓冲区中处理的，任什么时候候访问NIO中的数据，都是经过缓冲区进行的。最经常使用的缓冲区是ByteBuffer，经常使用的缓冲区还有下面几种：

关于Buffer的源码部分，因为篇幅关系再也不啰嗦。

Channel

Channel就是一个通道，网络数据经过Channel进行数据的读取。Stream只是在一个方向上流动，读和写分别在InputStream和OutputStream上进行，而Channel能够读和写同时进行。 实际上Channel能够分为两大类，用于网络数据读写的SelectableChannel和文件操做的FileChannel。NIO中的ServerSocketChannel和SocketChannel都是SelectableChannel的子类。

Selector

多路复用器Selector是NIO的基础，多路复用器能够不断地轮循注册在其上的Channel，若是某个Channel发生了读或者写事件，那么这个Channel就属于就绪状态，就会被Selector轮循出来，而后经过SelectionKey能够读取Channel的集合，进行后续的I/O操做。

JDK中的Selector使用了epoll()替代了传统的select，因此一个Selector能够同时注册大量的Channel，没有传统的链接句柄的限制。

NIO服务端和客户端解基本链路图

NIO服务端通讯的过程大体以下：

接下来看NIO客户端链路图：

这里就不贴server端和client端的代码了，由于这两部分的代码都比较冗长。

NIO对比BIO

1. 链接操做都是异步的，能够经过多路复用器Selector注册OP_CONNECT等待后续结果，不须要像以前客户端那样被同步阻塞；
2. SocketChannel的读写操做都是异步的，若是没有可读写的数据，直接同步返回，这样通讯IO线程就能够处理其余的请求，不会被阻塞。
3. 因为JDK的Selector在Linux等系统上都是经过epoll实现，他没有链接句柄的限制（上限是系统的最大句柄数或者对单个进程的句柄限制数），这意味着一个Selector能够处理成千上万个链接请求，并且性能方面也不会有明显的降低，所以，比较适合作高性能，高负载的服务器。

真正意义上的异步IO-AIO

JDK NIO2.0异步文件通道和异步套接字通道的实现，NIO2.0的异步套接字通道是真正意义上的异步非阻塞IO，熟悉UNIX的应该知道事件驱动I/O(AIO)，相比较NIO1.0，不须要经过到多路复用器就Selector对注册的通道Channel进行一个个的轮循就能够实现异步读写，所以实际编程中也比较简洁。 这里简单贴一下AIO实现一个基本的服务端代码实现。

服务器端代码：

public static class AioServerHandler implements Runnable {
        int port;
        CountDownLatch latch;
        AsynchronousServerSocketChannel ssc;
        public AioServerHandler(int port) {
            this.port = port;
            try {
                ssc = AsynchronousServerSocketChannel.open();
                ssc.bind(new InetSocketAddress(port));
                System.out.println("AioServer is started at port: " + port);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        @Override
        public void run() {
            latch = new CountDownLatch(1);
            // 读取请求消息
            doAccept();
            // 阻塞一下消息，防止线程退出
            try {
                latch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        public void doAccept() {
            // CompletionHandler
            ssc.accept(this, new AcceptCompletionHandler());
        }
    }
复制代码

// 接收链接
    public static class AcceptCompletionHandler implements CompletionHandler<AsynchronousSocketChannel, AioServerHandler> {
        // 读取客户端请求消息，而后将请求写回去
        @Override
        public void completed(AsynchronousSocketChannel result, AioServerHandler attachment) {
            // AsynchronousServerSocketChannel能够接成千上万的客户端，新的链接将继续调用complete方法
            attachment.ssc.accept(attachment, this); // 继续AsynchronousServerSocketChannel的accept方法，若是有新的客户端链接，将继续调用CompletionHandler的Complete方法
            // 读取消息
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            result.read(buffer, buffer, new ReadCompletionHandler(result));
        }
        @Override
        public void failed(Throwable exc, AioServerHandler attachment) {
            exc.printStackTrace();
            attachment.latch.countDown(); // 释放服务
        }
    }
复制代码

// 读取消息和返回消息给客户端
    public static class ReadCompletionHandler implements CompletionHandler<Integer, ByteBuffer> {
        AsynchronousSocketChannel channel;
        public ReadCompletionHandler(AsynchronousSocketChannel channel) {
            if (this.channel == null) {
                this.channel = channel;
            }
        }
        @Override
        public void completed(Integer result, ByteBuffer attachment) {
            attachment.flip();
            byte[] body = new byte[attachment.remaining()];
            attachment.get(body);

            try {
                String req = new String(body, "UTF-8");
                System.out.println("server接收到消息: " + req);
                doWrite(String.valueOf(System.currentTimeMillis()));
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            }
        }
        private void doWrite(String current) {
            byte[] bytes = current.getBytes();
            ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
            writeBuffer.put(bytes);
            writeBuffer.flip();
            channel.write(writeBuffer, writeBuffer, new CompletionHandler<Integer, ByteBuffer>() {
                @Override
                public void completed(Integer result, ByteBuffer attachment) {
                    // 入若是没有发送完，继续发送
                    if (attachment.hasRemaining()) {
                        channel.write(writeBuffer, writeBuffer, this);
                    }
                }
                @Override
                public void failed(Throwable exc, ByteBuffer attachment) {
                    try {
                        channel.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
        @Override
        public void failed(Throwable exc, ByteBuffer attachment) {
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

复制代码

AIO 服务端代码分析

AsynchronousServerSocketChannel做为一个异步的服务通道，而后绑定服务端口号。而后当AsynchronousServerSocketChannel.accept成功的接收请求了，再经过AcceptCompletionHandler对象来读取请求消息。CompletionHandler有两个方法：

1. public void completed(AsynchronousSocketChannel result, AioServerHandler attachment)
2. public void failed(Throwable exc, AioServerHandler attachment)

completed接口实现，读取attachment的AsynchronousServerSocketChannel，而后继续调用accept方法，这里接收客户端请求已经成功了，那为何还须要再次调用AsynchronousServerSocketChannel.accept方法呢？

由于对于AsynchronousServerSocketChannel.accept来讲，当有新的客户端请求的时候，系统将回调AcceptCompletionHandler.complete方法，表示新的客户端请求已经接收成功，因为AsynchronousServerSocketChannel能够链接成千上万的客户端，所以当一个客户端链接成功以后，继续调用accept方法以等待新的客户端来异步链接AsynchronousServerSocketChannel。

当新客户端和服务端的链接创建成功以后，则须要经过AsynchronousSocketChannel.read来异步读取客户端的请求消息。

@Override
    public final <A> void read(ByteBuffer dst,
                               A attachment,
                               CompletionHandler<Integer,? super A> handler)
    {
        read(dst, 0L, TimeUnit.MILLISECONDS, attachment, handler);
    }
复制代码

ByteBuffer dst接收缓冲区，用于从异步Channel中读取数据包，A attachment异步Channel绑定的附件，用于通知回调的时候做为入参使用，CompletionHandler<Integer,? super A> handler为异步回调接口handler。

继续看ReadCompletionHandler，将AsynchronousSocketChannel传给ReadCompletionHandler的构造方法，主要做为读取半包参数和应答客户端返回消息来用。关于半包读写这里再也不赘述，后续的RPC入门文章会继续说明。

这里主要针对AsynchronousSocketChannel.write方法进行说明：

@Override
    public final <A> void write(ByteBuffer src,
                                A attachment,
                                CompletionHandler<Integer,? super A> handler)

    {
        write(src, 0L, TimeUnit.MILLISECONDS, attachment, handler);
    }
复制代码

ByteBuffer src和A attachment与上面的read方法的参数意义同样，src做为AsynchronousSocketChannel的接收缓存；attachment做为Channel的绑定附件，回调的时候做为入参使用；这里直接实例化CompletionHandler做为实现write的异步回调，当能够写的时候会调用complete方法进行应答。

其实CompletionHandler的failed方法在实际的业务中须要注意下，须要对Throwable进行异常判断，若是是I/O异常，则须要关闭链路释放异常，若是是其余的异常则能够根据实际的业务须要进行处理。本例子中为了简单，就直接关闭链路。

这篇文章主要简单的介绍下相关的概念，关于客户端代码示例这里再也不叙述。后续的RPC系列文章会继续讲解，欢迎关注、点赞、留言分享。