天天看點

Java異步非阻塞程式設計的幾種方式

Java異步非阻塞程式設計的幾種方式

作者 | 塵定

來源 | 阿裡技術公衆号

一 從一個同步的Http調用說起

一個很簡單的業務邏輯,其他後端服務提供了一個接口,我們需要通過接口調用,擷取到響應的資料。

逆地理接口:通過經緯度擷取這個經緯度所在的省市區縣以及響應的code:

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

服務端執行,最簡單的同步調用方式:

Java異步非阻塞程式設計的幾種方式

服務端響應之前,IO會阻塞在:java.net.SocketInputStream#socketRead0 的native方法上:

Java異步非阻塞程式設計的幾種方式

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

線程模型示例:

Java異步非阻塞程式設計的幾種方式

同步最大的問題是在IO等待的過程中,線程資源沒有得到充分的利用,對于大量IO場景的業務吞吐量會有一定限制。

二 JDK NIO & Future

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

Java異步非阻塞程式設計的幾種方式
Java異步非阻塞程式設計的幾種方式

當然并不是所有的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)           
Java異步非阻塞程式設計的幾種方式

主線程在等待結果傳回過程中依然需要等待,沒有根本解決此問題。

三 使用Callback回調方式

第二節中,依然需要主線程等待,擷取結果,那麼可不可以在主線程完成發送請求後,再也不用關心這個邏輯,去執行其他的邏輯?那就可以使用Callback機制。

Java異步非阻塞程式設計的幾種方式

如此一來,主線程再也不需要關心發起IO後的業務邏輯,發送完請求後,就可以徹底去幹其他事情,或者回到線程池中再供排程。如果是HttpServer,那麼需要結合Servlet 3.1的異步Servlet。

Java異步非阻塞程式設計的幾種方式
Java異步非阻塞程式設計的幾種方式
異步Servelt參考資料 https://www.cnblogs.com/davenkin/p/async-servlet.html

使用Callback方式,從線程模型中看,發現線程資源已經得到了比較充分的利用,整個過程中已經沒有線程阻塞。

四 Callback hell

回調地獄,當Callback的線程還需要執行下一個IO調用的時候,這個時候進入回調地獄模式。

典型的應用場景如,通過經緯度擷取行政區域adcode(逆地理接口),然後再根據獲得的adcode,擷取當地的天氣資訊(天氣接口)。

在同步的程式設計模型中,幾乎不會涉及到此類問題。

Java異步非阻塞程式設計的幾種方式

Callback方式的核心缺陷

五 JDK 1.8 CompletableFuture

那麼有沒有辦法解決Callback Hell的問題?當然有,JDK 1.8中提供了CompletableFuture,先看看它是怎麼解決這個問題的。

将逆地理的Callback邏輯,封裝成一個獨立的CompletableFuture,當異步線程回調時,調用 future.complete(T) ,将結果封裝。

Java異步非阻塞程式設計的幾種方式

将天氣執行的Call邏輯,也封裝成為一個獨立的CompletableFuture ,完成之後,邏輯同上。

Java異步非阻塞程式設計的幾種方式

compose銜接,whenComplete輸出:

Java異步非阻塞程式設計的幾種方式

每一個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作為參數執行接下來的邏輯。

Java異步非阻塞程式設計的幾種方式

這樣一來,就完美解決回調地獄問題,在主的邏輯中,看起來像是在同步的進行編碼。

六 Vert.x Future

Info-Service中,大量使用的 Vert.x Future 也是類似的解決的方案,不過設計上使用Handler的概念。

Java異步非阻塞程式設計的幾種方式

核心執行的邏輯差不多:

Java異步非阻塞程式設計的幾種方式

這當然不是Vertx的全部,當然這是題外話了。

七 Reactive Streams

異步程式設計對吞吐量以及資源有好處,但是有沒有統一的抽象去解決此類問題内,答案是 Reactive Streams。

核心抽象:Publisher Subscriber Processor Subscription ,整個包裡面,隻有這四個接口,沒有實作類。

Java異步非阻塞程式設計的幾種方式

在JDK 9裡面,已經被作為一種規範封裝到 java.util.concurrent.Flow :

Java異步非阻塞程式設計的幾種方式
Java異步非阻塞程式設計的幾種方式
參考資料 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

一個簡單的例子:

Java異步非阻塞程式設計的幾種方式

八 Reactor & Spring 5 & Spring WebFlux

Flux & Mono

Java異步非阻塞程式設計的幾種方式
Java異步非阻塞程式設計的幾種方式
https://projectreactor.io/docs/core/3.1.0.M3/reference/index.html https://speakerdeck.com/simonbasle/projectreactor-dot-io-reactor3-intro