从bio到nio到aio,再到响应式编程,你这么说我就懂了

发布于 2021-09-04 17:13 ,所属分类:软件编程学习资料

要问计算机系统里,有哪些概念比较折腾人,nio绝对能算上一个。配合着多是异的网络编程,nio加上多线程一般能够完成双杀。

Linux有5种常见的IO模型。其中,阻塞IO就是bio,IO复用就是nio,异步IO就是aio,我们本篇文章就聚焦于此。

  • 阻塞式IO (bio)
  • 非阻塞式IO
  • IO复用 (nio)
  • 信号驱动式IO
  • 异步IO(aio)

在网络编程中,Reactor模型是必须要了解的。现在,大多数与IO相关的组件,都会使用Reactor模型,比如Tomcat、Redis、Nginx等,可见Reactor应用的广泛性。

Reactor是NIO的基础。为什么NIO的性能就能够比传统的阻塞IO性能高呢?我们首先来看一下传统阻塞式IO的一些特点。

1.阻塞IO模型

如上图,是典型的BIO模型,每当有一个连接到来,经过协调器的处理,就开启一个对应的线程进行接管。如果连接有1000条,那就需要1000个线程。线程资源是非常昂贵的,除了占用大量的内存,还会占用非常多的CPU调度时间,所以BIO在连接非常多的情况下,效率会变得非常低。

下面的代码是使用ServerSocket实现的一个简单socket服务器,监听在8888端口。

publicclassBIO{
staticbooleanstop=false;
publicstaticvoidmain(String[]args)throwsException{
intconnectionNum=0;
intport=8888;
ExecutorServiceservice=Executors.newCachedThreadPool();
ServerSocketserverSocket=newServerSocket(port);
while(!stop){
if(10==connectionNum){
stop=true;
}
Socketsocket=serverSocket.accept();
service.execute(()->{
try{
Scannerscanner=newScanner(socket.getInputStream());
PrintStreamprintStream=newPrintStream(socket.getOutputStream());
while(!stop){
Strings=scanner.next().trim();
printStream.println("PONG:"+s);
}
}catch(Exceptionex){
ex.printStackTrace();
}
});
connectionNum++;
}
service.shutdown();
serverSocket.close();
}
}

启动之后,使用nc命令进行连接测试,结果如下。

$nc-vlocalhost8888
Connectiontolocalhostport8888[tcp/ddi-tcp-1]succeeded!
hello
PONG:hello
nice
PONG:nice

可以看到,BIO的读写操作是阻塞的,线程的整个生命周期和连接的生命周期是一样的,而且不能够被复用。

就单个阻塞IO来说,它的效率并不比NIO慢。但是当服务的连接增多,考虑到整个服务器的资源调度和资源利用率等因素,NIO就有了显著的效果,NIO非常适合高并发场景。

2.非阻塞IO模型

其实,在处理IO动作时,有大部分时间是在等待。比如,socket连接要花费很长时间进行连接操作,在完成连接的这段时间内,它并没有占用额外的系统资源,但它只能阻塞等待在线程中。这种情况下,系统资源并不能被合理的利用。

Java的NIO,在Linux上底层是使用epoll实现的。epoll是一个高性能的多路复用I/O工具,改进了select和poll等工具的一些功能。在网络编程中,对epoll概念的一些理解,几乎是面试中必问的问题。

epoll的数据结构是直接在内核上进行支持的。通过epoll_create和epoll_ctl等函数的操作,可以构造描述符(fd)相关的事件组合(event)。

这里有两个比较重要的概念:

  • fd 每条连接、每个文件,都对应着一个描述符,比如端口号。内核在定位到这些连接的时候,就是通过fd进行寻址的
  • event 当fd对应的资源,有状态或者数据变动,就会更新epoll_item结构。在没有事件变更的时候,epoll就阻塞等待,也不会占用系统资源;一旦有新的事件到来,epoll就会被激活,将事件通知到应用方

关于epoll还会有一个面试题:相对于select,epoll有哪些改进?这里直接给出答案:

  • epoll不再需要像select一样对fd集合进行轮询,也不需要在调用时将fd集合在用户态和内核态进行交换
  • 应用程序获得就绪fd的事件复杂度,epoll时O(1),select是O(n)
  • select最大支持约1024个fd,epoll支持65535个
  • select使用轮询模式检测就绪事件,epoll采用通知方式,更加高效

我们还是以Java中的NIO代码为例,来看一下NIO的具体概念。

publicclassNIO{
staticbooleanstop=false;
publicstaticvoidmain(String[]args)throwsException{
intconnectionNum=0;
intport=8888;
ExecutorServiceservice=Executors.newCachedThreadPool();
ServerSocketChannelssc=ServerSocketChannel.open();
ssc.configureBlocking(false);
ssc.socket().bind(newInetSocketAddress("localhost",port));
Selectorselector=Selector.open();
ssc.register(selector,ssc.validOps());
while(!stop){
if(10==connectionNum){
stop=true;
}
intnum=selector.select();
if(num==0){
continue;
}
Iterator<SelectionKey>events=selector.selectedKeys().iterator();
while(events.hasNext()){
SelectionKeyevent=events.next();

if(event.isAcceptable()){
SocketChannelsc=ssc.accept();
sc.configureBlocking(false);
sc.register(selector,SelectionKey.OP_READ);
connectionNum++;
}elseif(event.isReadable()){
try{
SocketChannelsc=(SocketChannel)event.channel();
ByteBufferbuf=ByteBuffer.allocate(1024);
intsize=sc.read(buf);
if(-1==size){
sc.close();
}
Stringresult=newString(buf.array()).trim();
ByteBufferwrap=ByteBuffer.wrap(("PONG:"+result).getBytes());
sc.write(wrap);
}catch(Exceptionex){
ex.printStackTrace();
}
}elseif(event.isWritable()){
SocketChannelsc=(SocketChannel)event.channel();
}

events.remove();
}
}
service.shutdown();
ssc.close();
}
}

上面这段代码比较长,是使用NIO实现的和BIO相同的功能。从它的API设计上,我们就能够看到epoll的一些影子。

首先,我们创建了一个服务端ssc,并开启一个新的事件选择器,监听它的OP_ACCEPT事件。

ServerSocketChannelssc=ServerSocketChannel.open();
Selectorselector=Selector.open();
ssc.register(selector,ssc.validOps());

共有4种事件类型。分别是新连接事件(OP_ACCEPT)、连接就绪事件(OP_CONNECT)、读就绪事件(OP_READ)、写就绪事件(OP_WRITE)。任何网络和文件操作,都可以抽象成这四个事件。

接下来,在while循环里,使用select函数,阻塞在主线程里。所谓阻塞,就是操作系统不再分配CPU事件片到当前线程中,所以select函数是几乎不占用任何系统资源的。

intnum=selector.select();

一旦有新的事件到达,比如有新的连接到来,主线程就能够被调度到,程序就能够向下执行。这时候,就能够根据订阅的事件通知,持续获取订阅的事件。

由于注册到selector的连接和事件可能会有多个,所以这些事件也会有多个。我们使用安全的迭代器循环进行处理,在处理完毕之后,将它删除。

如果事件不删除的话,或者漏掉了某个事件的处理,会怎么样呢?后果还是比较严重的,由于事件总是存在,我们的程序会陷入无休无止的循环之中。

Iterator<SelectionKey>events=selector.selectedKeys().iterator();
while(events.hasNext()){
SelectionKeyevent=events.next();
...
events.remove();
}
}

有新的连接到达时,我们订阅了更多的事件。对于我们的数据读取来说,对应的事件就是OP_READ。和BIO编程面向流的方式不同,NIO操作的对象是抽象的概念Channel,通过缓冲区进行数据交换。

SocketChannelsc=ssc.accept();
sc.configureBlocking(false);
sc.register(selector,SelectionKey.OP_READ);

值得注意的是:服务端和客户端的实现方式,可以是不同的。比如,服务端是NIO,客户端可以是BIO,它们并没有什么强制要求。

另外一个面试时候经常问到的事件就是OP_WRITE。我们上面提到过,这个事件是表示写就绪的,当底层的缓冲区有空闲,这个事件就会一直发生,浪费占用CPU资源。所以,我们一般是不注册OP_WRITE的。

这里还有一个细节,在读取数据的时候,并没有像BIO的方式一样使用循环来获取数据。如下面的代码,我们创建了一个1024字节的缓冲区,用于数据的读取。如果连接中的数据,大于1024字节怎么办?

SocketChannelsc=(SocketChannel)event.channel();
ByteBufferbuf=ByteBuffer.allocate(1024);
intsize=sc.read(buf);

这涉及到两种事件的通知机制。

  • 水平触发 (level-triggered) 称作LT模式。只要缓冲区有数据,事件就会一直发生
  • 边缘触发 (edge-triggered) 称作ET模式。缓冲区有数据,仅会触发一次。事件想要再次触发,必须先将fd中的数据读完才行

可以看到,Java的NIO采用的就是水平触发的方式。LT模式频繁环唤醒线程,效率相比较ET模式低,所以Netty使用JNI的方式,实现了ET模式,效率上更高一些。

3.Reactor模式

了解了BIO和NIO的一些使用方式,Reactor模式就呼之欲出了。

NIO是基于事件机制的,有一个叫做Selector的选择器,阻塞获取的事件列表。获取到事件列表后,可以通过分发器,进行真正的数据操作。

上图是Doug Lea在讲解NIO时候的一张图,指明了最简单的Reactor模型的基本元素。你可以对比这上面的NIO代码分析一下,里面有四个主要元素:

  • Acceptor 处理client的连接,并绑定具体的事件处理器
  • Event 具体发生的事件
  • Handler 执行具体事件的处理者。比如处理读写事件
  • Reactor 将具体的事件分配给Handler

我们可以对上面的模型进行近一步细化,下面这张图同样是Doug Lea的ppt中的。它把Reactor部分分为mainReactor和subReactor两部分。mainReactor负责监听处理新的连接,然后将后续的事件处理交给subReactor,subReactor对事件处理的方式,也由阻塞模式变成了多线程处理,引入了任务队列的模式。

熟悉Netty的同学可以看到,这个模型就是Netty设计的基础。在Netty中,Boss线程对应着对连接的处理和分派,相当于mainReactor;Work线程 对应着subReactor,使用多线程负责读写事件的分发和处理。

这种模式将每个组件的职责分的更细,耦合度也更低,能有效的解决C10k问题。

4.AIO

关于NIO的概念,误解还是比较多的。面试官可能会问你:为什么我在使用NIO的时候,使用Channel进行读写,socket的操作依然是阻塞的?NIO主要体现在哪里?

//这行代码是阻塞的
intsize=sc.read(buf);

答案就是,NIO只负责对发生在fd描述符上的事件进行通知。事件的获取和通知部分是非阻塞的,但收到通知之后的操作,却是阻塞的。即使使用多线程去处理这些事件,它依然是阻塞的。

AIO更近一步,将这些对事件的操作也变成非阻塞的。下面是一段典型的AIO代码,它通过注册CompletionHandler回调函数进行事件处理。这里的事件是隐藏的,比如read函数,它不仅仅代表Channel可读了,而且会把数据自动的读取到ByteBuffer中。等完成了读取,就会通过回调函数通知你,进行后续的操作。

publicclassAIO{
publicstaticvoidmain(String[]args)throwsException{
intport=8888;
AsynchronousServerSocketChannelssc=AsynchronousServerSocketChannel.open();
ssc.bind(newInetSocketAddress("localhost",port));
ssc.accept(null,newCompletionHandler<AsynchronousSocketChannel,Object>(){
voidjob(finalAsynchronousSocketChannelsc){
ByteBufferbuffer=ByteBuffer.allocate(1024);
sc.read(buffer,buffer,newCompletionHandler<Integer,ByteBuffer>(){
@Override
publicvoidcompleted(Integerresult,ByteBufferattachment){
Stringstr=newString(attachment.array()).trim();
ByteBufferwrap=ByteBuffer.wrap(("PONG:"+str).getBytes());
sc.write(wrap,null,newCompletionHandler<Integer,Object>(){
@Override
publicvoidcompleted(Integerresult,Objectattachment){
job(sc);
}
@Override
publicvoidfailed(Throwableexc,Objectattachment){
System.out.println("error");
}
});
}
@Override
publicvoidfailed(Throwableexc,ByteBufferattachment){
System.out.println("error");
}
});
}
@Override
publicvoidcompleted(AsynchronousSocketChannelsc,Objectattachment){
ssc.accept(null,this);
job(sc);
}
@Override
publicvoidfailed(Throwableexc,Objectattachment){
exc.printStackTrace();
System.out.println("error");
}
});
Thread.sleep(Integer.MAX_VALUE);
}
}

AIO是Java1.7加入的,理论上性能是会提升的,但它现在发展的不太好。那部分对数据进行自动读取的操作,总得有地方实现,不在框架里,就得在内核里。Netty的NIO模型加上多线程处理,在这方面已经做的很好,编程模式也非常简单。所以,市面上对AIO的实践并不多,在采用技术选型的时候,一定要谨慎。

5.响应式编程

你可能听说过Spring5的webflux,webflux是可以替代spring mvc的一套解决方案,可以编写响应式的应用,两者之间的关系可以看下图。它的底层使用的是netty,所以操作是异步非阻塞的。类似的组件还有vert.x、akka、rxjava等。

webflux是运行在project reactor之上的一个封装,其根本特性是由后者提供的。至于再底层的非阻塞模型,就是由Netty保证的了。

非阻塞的特性我们可以理解,响应式又是什么概念呢?

响应式编程是一种面向数据流和变化传播的编程范式。这意味着可以在编程语言中很方便地表达静态或动态的数据流,而相关的计算模型会自动将变化的值通过数据流进行传播。

这段话很晦涩,在编程方面,它表达的意思是:把生产者消费者模式,使用简单的API表示出来,并自动处理背压(backpressure)问题。

背压,指的是生产者与消费者之间的流量控制。通过将操作全面异步化,来减少无效的等待和资源消耗。

Java的lambda表达式可以承担简单这个职责,Java9更是引入了响应式流(Reactive Stream),方便了我们的操作。比如,下面是Spring Cloud GateWayFluent API写法,响应式编程的API都是类似的。

publicRouteLocatorcustomerRouteLocator(RouteLocatorBuilderbuilder){
returnbuilder.routes()
.route(r->r.path("/market/**")
.filters(f->f.filter(newRequestTimeFilter())
.addResponseHeader("X-Response-Default-Foo","Default-Bar"))
.uri("http://localhost:8080/market/list")
.order(0)
.id("customer_filter_router")
)
.build();
}

从传统的开发模式过渡到reactor的开发模式,是有一定成本的,不过它确实能够提高我们应用程序的性能。具体用不用,就要看在编程难度和性能之间的取舍了。

小结

从上面的描述,我们了解到,BIO的线程模型是一个连接对应一个线程的,非常的浪费资源;NIO通过对关键事件的监听,通过主动通知的方式完成非阻塞操作,但它对事件本身的处理依然是非阻塞的;AIO完全是异步非阻塞的,但现实中使用很少。

使用Netty的多Acceptor模式和多线程模式,我们能够方便的完成类似AIO这样的操作。Netty的事件触发机制使用了高效的ET模式,使得支持的连接更多,性能更高。

使用Netty,能够构建响应式编程的基础,加上类似lambda表达式这样的书写风格,能够完成类似WebFlux这样的响应式框架。响应式编程是一个趋势,现在有越来越多的框架和底层的数据库支持响应式编程,我们的应用响应也会更加迅速。

开箱即用的微服务权限解决方案,你打几分?

2021-08-25

写Java代码,顺带就看了Mysql的binlog,如何?

2021-08-23

简单的Postman,还能玩出花?花到起飞呗

2021-08-18


相关资源