前段时间在看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
系列文章会继续讲解,欢迎关注、点赞、留言分享。