乐趣区

关于java:Java异步非阻塞编程的几种方式

jiezi

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

一个很简略的业务逻辑，其余后端服务提供了一个接口，咱们须要通过接口调用，获取到响应的数据。

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

curl-i”http://xxx?latitude=31.08966221524924&channel=amap7a&near=false&longitude=105.13990312814713″

{“adcode”:”510722″}

服务端执行，最简略的同步调用形式：

服务端响应之前，IO 会阻塞在：java.net.SocketInputStream#socketRead0 的 native 办法上：

通过 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)
            .......

线程模型示例：

同步最大的问题是在 IO 期待的过程中，线程资源没有失去充沛的利用，对于大量 IO 场景的业务吞吐量会有肯定限度。

二 JDK NIO & Future

在 JDK 1.5 中，JUC 提供了 Future 形象：

!

当然并不是所有的 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)

主线程在期待后果返回过程中仍然须要期待，没有基本解决此问题。

三 应用 Callback 回调形式

第二节中，仍然须要主线程期待，获取后果，那么可不可以在主线程实现发送申请后，再也不必关怀这个逻辑，去执行其余的逻辑？那就能够应用 Callback 机制。

如此一来，主线程再也不须要关怀发动 IO 后的业务逻辑，发送完申请后，就能够彻底去干其余事件，或者回到线程池中再供调度。如果是 HttpServer，那么须要联合 Servlet 3.1 的异步 Servlet。

异步 Servelt 参考资料

https://www.cnblogs.com/daven…

应用 Callback 形式，从线程模型中看，发现线程资源曾经失去了比拟充沛的利用，整个过程中曾经没有线程阻塞。

四 Callback hell

回调天堂，当 Callback 的线程还须要执行下一个 IO 调用的时候，这个时候进入回调天堂模式。

典型的利用场景如，通过经纬度获取行政区域 adcode（逆天文接口），而后再依据取得的 adcode，获取当地的天气信息（天气接口）。

在同步的编程模型中，简直不会波及到此类问题。

!

Callback 形式的外围缺点

五 JDK 1.8 CompletableFuture

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

将逆天文的 Callback 逻辑，封装成一个独立的 CompletableFuture，当异步线程回调时，调用 future.complete(T)，将后果封装。

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

compose 连接，whenComplete 输入：

每一个 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 作为参数执行接下来的逻辑。

这样一来，就完满解决回调天堂问题，在主的逻辑中，看起来像是在同步的进行编码。

六 Vert.x Future

Info-Service 中，大量应用的 Vert.x Future 也是相似的解决的计划，不过设计上应用

外围执行的逻辑差不多：

这当然不是 Vertx 的全副，当然这是题外话了。

七 Reactive Streams

异步编程对吞吐量以及资源有益处，然而有没有对立的形象去解决此类问题内，答案是 Reactive Streams。

外围形象：Publisher Subscriber Processor Subscription，整个包外面，只有这四个接口，没有实现类。

在 JDK 9 外面，曾经被作为一种标准封装到 java.util.concurrent.Flow：

参考资料

https://www.baeldung.com/java…
http://ypk1226.com/2019/07/01…
https://www.reactivemanifesto…
https://projectreactor.io/learn

一个简略的例子：

八 Reactor & Spring 5 & Spring WebFlux

Flux & Mono

参考资料

https://projectreactor.io/doc…
https://speakerdeck.com/simon…

退出移动版