推荐学习

• 完美！白嫖4份满分级“并发编程”宝典，不得不献出我的膝盖呀
  
• 牛掰！“基础-中级-高级”Java程序员面试集结，看完献出我的膝盖
  
• 最新Java岗面试清单：分布式+Dubbo+线程+Redis+数据库+JVM+并发
  

前段时间在看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系列文章会继续讲解。

作者：pdm_peng
链接：https://juejin.im/post/5eee126551882565ba1d5ee2