Java异步非阻塞编程的几种方式

简介: Java异步非阻塞编程的几种方式

image.png

作者 | 尘定
来源 | 阿里技术公众号

一 从一个同步的Http调用说起

一个很简单的业务逻辑,其他后端服务提供了一个接口,我们需要通过接口调用,获取到响应的数据。

逆地理接口:通过经纬度获取这个经纬度所在的省市区县以及响应的code:

curl-i"http://xxx?latitude=31.08966221524924&channel=amap7a&near=false&longitude=105.13990312814713"
{"adcode":"510722"}

服务端执行,最简单的同步调用方式:

image.png

服务端响应之前,IO会阻塞在:java.net.SocketInputStream#socketRead0 的native方法上:

image.png

通过jstack日志,可以发现,此时这个Thread会一直在runable的状态:

"main"#1 prio=5 os_prio=31 tid=0x00007fed0c810000 nid=0x1003 runnable [0x000070000ce14000]   java.lang.Thread.State: RUNNABLE
        at java.net.SocketInputStream.socketRead0(Native Method)
        at java.net.SocketInputStream.socketRead(SocketInputStream.java:116)
        at java.net.SocketInputStream.read(SocketInputStream.java:171)
        at java.net.SocketInputStream.read(SocketInputStream.java:141)
        at org.apache.http.impl.conn.LoggingInputStream.read(LoggingInputStream.java:84)
        at org.apache.http.impl.io.SessionInputBufferImpl.streamRead(SessionInputBufferImpl.java:137)
        at org.apache.http.impl.io.SessionInputBufferImpl.fillBuffer(SessionInputBufferImpl.java:153)
        at org.apache.http.impl.io.SessionInputBufferImpl.readLine(SessionInputBufferImpl.java:282)
        at org.apache.http.impl.conn.DefaultHttpResponseParser.parseHead(DefaultHttpResponseParser.java:138)
        at org.apache.http.impl.conn.DefaultHttpResponseParser.parseHead(DefaultHttpResponseParser.java:56)
        at org.apache.http.impl.io.AbstractMessageParser.parse(AbstractMessageParser.java:259)
        at org.apache.http.impl.DefaultBHttpClientConnection.receiveResponseHeader(DefaultBHttpClientConnection.java:163)
        at org.apache.http.impl.conn.CPoolProxy.receiveResponseHeader(CPoolProxy.java:165)
        at org.apache.http.protocol.HttpRequestExecutor.doReceiveResponse(HttpRequestExecutor.java:273)
        at org.apache.http.protocol.HttpRequestExecutor.execute(HttpRequestExecutor.java:125)
        at org.apache.http.impl.execchain.MainClientExec.execute(MainClientExec.java:272)
        at org.apache.http.impl.execchain.ProtocolExec.execute(ProtocolExec.java:185)
        at org.apache.http.impl.execchain.RetryExec.execute(RetryExec.java:89)
        at org.apache.http.impl.execchain.RedirectExec.execute(RedirectExec.java:110)
        at org.apache.http.impl.client.InternalHttpClient.doExecute(InternalHttpClient.java:185)
        at org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:83)
        at org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:108)
        at com.amap.aos.async.AsyncIO.blockingIO(AsyncIO.java:207)
                .......

线程模型示例:

image.png

同步最大的问题是在IO等待的过程中,线程资源没有得到充分的利用,对于大量IO场景的业务吞吐量会有一定限制。

二 JDK NIO & Future

在JDK 1.5 中,JUC提供了Future抽象:

image.png

image.png

当然并不是所有的Future都是这样实现的,如 io.netty.util.concurrent.AbstractFuture 就是通过线程轮询去。

这样做的好处是,主线程可以不用等待IO响应,可以去做点其他的,比如说再发送一个IO请求,可以等到一起返回:

"main"#1 prio=5 os_prio=31 tid=0x00007fd7a500b000 nid=0xe03 waiting on condition [0x000070000a95d000]   java.lang.Thread.State: WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
- parking to wait for  <0x000000076ee2d768> (a java.util.concurrent.CountDownLatch$Sync)
        at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304)
        at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231)
        at org.asynchttpclient.netty.NettyResponseFuture.get(NettyResponseFuture.java:162)
        at com.amap.aos.async.AsyncIO.futureBlockingGet(AsyncIO.java:201)
        .....
"AsyncHttpClient-2-1"#11 prio=5 os_prio=31 tid=0x00007fd7a7247800 nid=0x340b runnable [0x000070000ba94000]   java.lang.Thread.State: RUNNABLE
        at sun.nio.ch.KQueueArrayWrapper.kevent0(Native Method)
        at sun.nio.ch.KQueueArrayWrapper.poll(KQueueArrayWrapper.java:198)
        at sun.nio.ch.KQueueSelectorImpl.doSelect(KQueueSelectorImpl.java:117)
        at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
- locked <0x000000076eb00ef0> (a io.netty.channel.nio.SelectedSelectionKeySet)
- locked <0x000000076eb00f10> (a java.util.Collections$UnmodifiableSet)
- locked <0x000000076eb00ea0> (a sun.nio.ch.KQueueSelectorImpl)
        at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
        at io.netty.channel.nio.NioEventLoop.select(NioEventLoop.java:693)
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:353)
        at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:140)
        at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:144)
        at java.lang.Thread.run(Thread.java:748)

image.png

主线程在等待结果返回过程中依然需要等待,没有根本解决此问题。

三 使用Callback回调方式

第二节中,依然需要主线程等待,获取结果,那么可不可以在主线程完成发送请求后,再也不用关心这个逻辑,去执行其他的逻辑?那就可以使用Callback机制。

image.png

如此一来,主线程再也不需要关心发起IO后的业务逻辑,发送完请求后,就可以彻底去干其他事情,或者回到线程池中再供调度。如果是HttpServer,那么需要结合Servlet 3.1的异步Servlet。

image.png
image.png

异步Servelt参考资料
https://www.cnblogs.com/davenkin/p/async-servlet.html

使用Callback方式,从线程模型中看,发现线程资源已经得到了比较充分的利用,整个过程中已经没有线程阻塞。

四 Callback hell

回调地狱,当Callback的线程还需要执行下一个IO调用的时候,这个时候进入回调地狱模式。

典型的应用场景如,通过经纬度获取行政区域adcode(逆地理接口),然后再根据获得的adcode,获取当地的天气信息(天气接口)。

在同步的编程模型中,几乎不会涉及到此类问题。

image.png

Callback方式的核心缺陷

五 JDK 1.8 CompletableFuture

那么有没有办法解决Callback Hell的问题?当然有,JDK 1.8中提供了CompletableFuture,先看看它是怎么解决这个问题的。

将逆地理的Callback逻辑,封装成一个独立的CompletableFuture,当异步线程回调时,调用 future.complete(T) ,将结果封装。

image.png

将天气执行的Call逻辑,也封装成为一个独立的CompletableFuture ,完成之后,逻辑同上。

image.png

compose衔接,whenComplete输出:

image.png

每一个IO操作,均可以封装为独立的CompletableFuture,从而避免回调地狱。

CompletableFuture,只有两个属性:

  • result:Future的执行结果 (Either the result or boxed AltResult)。
  • stack:操作栈,用于定义这个Future接下来操作的行为 (Top of Treiber stack of dependent actions)。

weatherFuture这个方法是如何被调用的呢?

通过堆栈可以发现,是在 reverseCodeFuture.complete(result) 的时候,并且也将获得的adcode作为参数执行接下来的逻辑。

image.png

这样一来,就完美解决回调地狱问题,在主的逻辑中,看起来像是在同步的进行编码。

六 Vert.x Future

Info-Service中,大量使用的 Vert.x Future 也是类似的解决的方案,不过设计上使用Handler的概念。

image.png

核心执行的逻辑差不多:

image.png

这当然不是Vertx的全部,当然这是题外话了。

七 Reactive Streams

异步编程对吞吐量以及资源有好处,但是有没有统一的抽象去解决此类问题内,答案是 Reactive Streams。

核心抽象:Publisher Subscriber Processor Subscription ,整个包里面,只有这四个接口,没有实现类。

image.png


在JDK 9里面,已经被作为一种规范封装到 java.util.concurrent.Flow :

image.png
image.png

参考资料
https://www.baeldung.com/java-9-reactive-streams
http://ypk1226.com/2019/07/01/reactive/reactive-streams/
https://www.reactivemanifesto.org/
https://projectreactor.io/learn

一个简单的例子:

image.png

八 Reactor & Spring 5 & Spring WebFlux

Flux & Mono
image.png
image.png

参考资料
https://projectreactor.io/docs/core/3.1.0.M3/reference/index.html
https://speakerdeck.com/simonbasle/projectreactor-dot-io-reactor3-intro
相关文章
|
15天前
|
安全 Java 开发者
深入理解Java并发编程:线程安全与性能优化
【4月更文挑战第9天】本文将深入探讨Java并发编程的核心概念,包括线程安全和性能优化。我们将详细解析Java中的同步机制,包括synchronized关键字、Lock接口以及并发集合等,并探讨它们如何影响程序的性能。此外,我们还将讨论Java内存模型,以及它如何影响并发程序的行为。最后,我们将提供一些实用的并发编程技巧和最佳实践,帮助开发者编写出既线程安全又高效的Java程序。
22 3
|
16天前
|
Java
Java 并发编程:深入理解线程池
【4月更文挑战第8天】本文将深入探讨 Java 中的线程池技术,包括其工作原理、优势以及如何使用。线程池是 Java 并发编程的重要工具,它可以有效地管理和控制线程的执行,提高系统性能。通过本文的学习,读者将对线程池有更深入的理解,并能在实际开发中灵活运用。
|
12天前
|
安全 算法 Java
深入理解Java并发编程:线程安全与性能优化
【4月更文挑战第11天】 在Java中,高效的并发编程是提升应用性能和响应能力的关键。本文将探讨Java并发的核心概念,包括线程安全、锁机制、线程池以及并发集合等,同时提供实用的编程技巧和最佳实践,帮助开发者在保证线程安全的前提下,优化程序性能。我们将通过分析常见的并发问题,如竞态条件、死锁,以及如何利用现代Java并发工具来避免这些问题,从而构建更加健壮和高效的多线程应用程序。
|
17天前
|
Java
Java并发编程:深入理解线程池
【4月更文挑战第7天】在现代软件开发中,多线程编程已经成为一种不可或缺的技术。为了提高程序性能和资源利用率,Java提供了线程池这一强大工具。本文将深入探讨Java线程池的原理、使用方法以及如何根据实际需求定制线程池,帮助读者更好地理解和应用线程池技术。
15 0
|
5天前
|
IDE Java 物联网
《Java 简易速速上手小册》第1章:Java 编程基础(2024 最新版)
《Java 简易速速上手小册》第1章:Java 编程基础(2024 最新版)
13 0
|
5天前
|
安全 Java 开发者
Java并发编程:深入理解Synchronized关键字
【4月更文挑战第19天】 在Java多线程编程中,为了确保数据的一致性和线程安全,我们经常需要使用到同步机制。其中,`synchronized`关键字是最为常见的一种方式,它能够保证在同一时刻只有一个线程可以访问某个对象的特定代码段。本文将深入探讨`synchronized`关键字的原理、用法以及性能影响,并通过具体示例来展示如何在Java程序中有效地应用这一技术。
|
6天前
|
安全 Java 调度
Java并发编程:深入理解线程与锁
【4月更文挑战第18天】本文探讨了Java中的线程和锁机制,包括线程的创建(通过Thread类、Runnable接口或Callable/Future)及其生命周期。Java提供多种锁机制,如`synchronized`关键字、ReentrantLock和ReadWriteLock,以确保并发访问共享资源的安全。此外,文章还介绍了高级并发工具,如Semaphore(控制并发线程数)、CountDownLatch(线程间等待)和CyclicBarrier(同步多个线程)。掌握这些知识对于编写高效、正确的并发程序至关重要。
|
7天前
|
缓存 分布式计算 监控
Java并发编程:深入理解线程池
【4月更文挑战第17天】在Java并发编程中,线程池是一种非常重要的技术,它可以有效地管理和控制线程的执行,提高系统的性能和稳定性。本文将深入探讨Java线程池的工作原理,使用方法以及在实际开发中的应用场景,帮助读者更好地理解和使用Java线程池。
|
7天前
|
Java API 数据库
深研Java异步编程:CompletableFuture与反应式编程范式的融合实践
【4月更文挑战第17天】本文探讨了Java中的CompletableFuture和反应式编程在提升异步编程体验上的作用。CompletableFuture作为Java 8引入的Future扩展,提供了一套流畅的链式API,简化异步操作,如示例所示的非阻塞数据库查询。反应式编程则关注数据流和变化传播,通过Reactor等框架实现高度响应的异步处理。两者结合,如将CompletableFuture转换为Mono或Flux,可以兼顾灵活性和资源管理,适应现代高并发环境的需求。开发者可按需选择和整合这两种技术,优化系统性能和响应能力。
|
7天前
|
存储 缓存 安全
Java并发基础之互斥同步、非阻塞同步、指令重排与volatile
在Java中,多线程编程常常涉及到共享数据的访问,这时候就需要考虑线程安全问题。Java提供了多种机制来实现线程安全,其中包括互斥同步(Mutex Synchronization)、非阻塞同步(Non-blocking Synchronization)、以及volatile关键字等。 互斥同步(Mutex Synchronization) 互斥同步是一种基本的同步手段,它要求在任何时刻,只有一个线程可以执行某个方法或某个代码块,其他线程必须等待。Java中的synchronized关键字就是实现互斥同步的常用手段。当一个线程进入一个synchronized方法或代码块时,它需要先获得锁,如果
24 0