RPC框架入门篇(1)之【BIO,NIO,AIO】

2,612 阅读9分钟

前段时间在看dubbo的源码,看的差不多了也开始在写一个RPC框架,现在写的快一半了,才想起来怎么循序渐进的通过文章的方式跟大家聊这个东西。于是思来想去,决定先从最基础的服务间网络通信说起比较好,后面再慢慢的跟大家引出怎么去写一个RPC框架。

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

传统的BIO通信弊端


传统的BIO通信当接收到客户端的请求时,为每一个请求创建一个新的线程进行链路处理,处理完成之后,通过输出流返回给客户端,然后线程销毁。

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

伪异步IO模型


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

伪异步IO的缺点

无论是BIO还是伪异步本质上都是阻塞型I/O,都是基于Stream进行网络数据的读和写。首先我们看下InputStreamread方法源码:

    /**
     * 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种情况:

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

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

在继续看下Outputtreamwrite方法:

    /**
     * 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的SocketServerSocket在生产使用问题很多,因此NIO诞生了,对应的是SocketChannelServerSocketChannel两个类。

NIO编程介绍


NIO相关概念

Buffer

传统的BIO主要是面向流的,可以将数据直接写入或者读取到Stream对象中;而在NIO中,读取和写入数据都是在缓冲区中处理的,任何时候访问NIO中的数据,都是通过缓冲区进行的。最常用的缓冲区是ByteBuffer,常用的缓冲区还有下面几种:

关于Buffer的源码部分,由于篇幅关系不再啰嗦。

Channel

Channel就是一个通道,网络数据通过Channel进行数据的读取。Stream只是在一个方向上流动,读和写分别在InputStreamOutputStream上进行,而Channel可以读和写同时进行。 实际上Channel可以分为两大类,用于网络数据读写的SelectableChannel和文件操作的FileChannelNIO中的ServerSocketChannelSocketChannel都是SelectableChannel的子类。

Selector

多路复用器SelectorNIO的基础,多路复用器可以不断地轮循注册在其上的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接口实现,读取attachmentAsynchronousServerSocketChannel,然后继续调用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 srcA attachment与上面的read方法的参数意义一样,src作为AsynchronousSocketChannel的接收缓存;attachment作为Channel的绑定附件,回调的时候作为入参使用;这里直接实例化CompletionHandler作为实现write的异步回调,当可以写的时候会调用complete方法进行应答。

其实CompletionHandlerfailed方法在实际的业务中需要注意下,需要对Throwable进行异常判断,如果是I/O异常,则需要关闭链路释放异常,如果是其他的异常则可以根据实际的业务需要进行处理。本例子中为了简单,就直接关闭链路。

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