实现rpc的几个要点（RPC框架入门篇BIONIO）

审美累劳 2023-01-01 09:57:27

推荐学习

完美！白嫖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
连接操作都是异步的，可以通过多路复用器Selector注册OP_CONNECT等待后续结果，不需要像之前客户端那样被同步阻塞；
SocketChannel的读写操作都是异步的，如果没有可读写的数据，直接同步返回，这样通信IO线程就可以处理其他的请求，不会被阻塞。
由于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有两个方法：
public void completed(AsynchronousSocketChannel result, AioServerHandler attachment)
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
,