這篇文章主要講解了“Java中怎么處理異步超時的問題”,文中的講解內容簡單清晰,易于學習與理解,下面請大家跟著小編的思路慢慢深入,一起來研究和學習“Java中怎么處理異步超時的問題”吧!
專注于為中小企業(yè)提供成都網(wǎng)站建設、網(wǎng)站建設服務,電腦端+手機端+微信端的三站合一,更高效的管理,為中小企業(yè)行唐免費做網(wǎng)站提供優(yōu)質的服務。我們立足成都,凝聚了一批互聯(lián)網(wǎng)行業(yè)人才,有力地推動了千余家企業(yè)的穩(wěn)健成長,幫助中小企業(yè)通過網(wǎng)站建設實現(xiàn)規(guī)模擴充和轉變。
一天,我在改進多線程代碼時被Future.get()卡住了。
public void serve() throws InterruptedException, ExecutionException, TimeoutException { final FutureresponseFuture = asyncCode(); final Response response = responseFuture.get(1, SECONDS); send(response); } private void send(Response response) { //... }
這是用Java寫的一個Akka應用程序,使用了一個包含1000個線程的線程池(原來如此?。?mdash;—所有的線程都在阻塞在這個 get() 中。系統(tǒng)的處理速度跟不上并發(fā)請求的數(shù)量。重構以后,我們干掉了所有的這些線程僅保留了一個,極大的減少了內存的占用。我們簡單一點,通過一個Java 8的例子來演示。***步是使用CompletableFuture來替換簡單的Future(見:Tip 9)。
通過控制任務提交到ExecutorService的方式:只需用 CompletableFuture.supplyAsync(…, executorService) 來代替 executorService.submit(…) 即可
處理基于回調函數(shù)的API:使用promises
否則(如果你已經(jīng)使用了阻塞式的API或 Future
public void serve() throws InterruptedException, ExecutionException, TimeoutException { final CompletableFutureresponseFuture = asyncCode(); final Response response = responseFuture.get(1, SECONDS); send(response); }
很明顯,這不能解決任何問題,我們還必須利用新的風格來編程:
public void serve() { final CompletableFutureresponseFuture = asyncCode(); responseFuture.thenAccept(this::send); }
這個功能上是等同的,但是 serve() 只會運行一小段時間(不會阻塞或等待)。只需要記?。簍his::send 將會在完成 responseFuture 的同一個線程內執(zhí)行。如果你不想花費太大的代價來重載已經(jīng)存在的線程池或send()方法,可以考慮通過 thenAcceptAsync(this::send, sendPool) 好極了,但是我們失去了兩個重要屬性:異常傳播與超時。異常傳播很難實現(xiàn),因為我們改變了API。當serve()存在的時候,異步操作可能還沒有完成。 如果你關心異常,可以考慮返回 responseFutureor 或者其他可選的機制。至少,應該有異常的日志,否則該異常就會被吞噬了。
final CompletableFutureresponseFuture = asyncCode(); responseFuture.exceptionally(throwable -> { log.error("Unrecoverable error", throwable); return null; });
請小心上面的代碼:exceptionally() 試圖從失敗中恢復過來,返回一個可選的結果。這個地方雖可以正常的工作,但是如果對 exceptionally()和withthenAccept() 使用鏈式調用,即使失敗了也還是會調用 send() 方法,返回一個null參數(shù),或者任何其它從 exceptionally() 方法中返回的值。
responseFuture .exceptionally(throwable -> { log.error("Unrecoverable error", throwable); return null; }) .thenAccept(this::send); //probably not what you think
丟失一秒超時的問題非常巧妙。我們原始的代碼在Future完成之前最多等待(阻塞)1秒,否則就會拋出 TimeoutException。我們丟失了這個功能,更糟糕的是,單元測試超時的不是很方便,經(jīng)常會跳過這個環(huán)節(jié)。為了維持超時機制,而又不破壞事件 驅動的原則,我們需要建立一個額外的模塊:一個在給定時間后必定會失敗的 Future。
public staticCompletableFuture failAfter(Duration duration) { final CompletableFuture promise = new CompletableFuture<>(); scheduler.schedule(() -> { final TimeoutException ex = new TimeoutException("Timeout after " + duration); return promise.completeExceptionally(ex); }, duration.toMillis(), MILLISECONDS); return promise; } private static final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool( 1, new ThreadFactoryBuilder() .setDaemon(true) .setNameFormat("failAfter-%d") .build());
這個很簡單:我們創(chuàng)建一個promise(沒有后臺任務或線程池的 Future),然后在給定的 java.time.Duration 之后會拋出 TimeoutException 異常。如果在某個地方調用 get() 獲取這個 Future,阻塞的時間到達這個指定的時間后會拋出 TimeoutException。
實際上,它是一個包裝了 TimeoutException 的 ExecutionException,這個無需多說。注意,我使用了固定一個線程的線程池。這不僅僅是為了教學的目的:這是“1個線程應當能滿足任何人 的需求”的場景。failAfter() 本身沒多大的用處,但是如果和 ourresponseFuture 一起使用,我們就能解決這個問題了。
final CompletableFutureresponseFuture = asyncCode(); final CompletableFuture oneSecondTimeout = failAfter(Duration.ofSeconds(1)); responseFuture .acceptEither(oneSecondTimeout, this::send) .exceptionally(throwable -> { log.error("Problem", throwable); return null; });
這里還做了很多其他事情。在后臺的任務接收 responseFuture 時,我們也創(chuàng)建了一個“合成”的 oneSecondTimeout future,這在成功的時候永遠不會執(zhí)行,但是在1秒后就會導致任務失敗?,F(xiàn)在我們聯(lián)合這兩個叫做 acceptEither,這個操作將執(zhí)行先完成 Future 的代碼塊,而簡單的忽略 responseFuture 或 oneSecondTimeout 中運行比較慢的那個。如果 asyncCode() 代碼在1秒內執(zhí)行完成,this::send 就會被調用,而 oneSecondTimeout 異常就不會拋出。但是,如果 asyncCode() 執(zhí)行真的很慢,oneSecondTimeout 異常就先拋出。由于一個異常導致任務失敗,exceptionallyerror 處理器就會被調用,而不是 this::send 方法。你可以選擇執(zhí)行 send() 或者 exceptionally,但是不能兩個都執(zhí)行。當如,如果我們有兩個“普通”的 Future 正常執(zhí)行完成了,則***響應的那個將調用 send() 方法,后面的就會被丟棄。
這個不是最清晰的解決方案。更清晰的方案是包裝原始的 Future,然后保證它能在給定的時間內執(zhí)行。這種操作對 com.twitter.util.Future 是可行的(Scala叫做 within()),但是 scala.concurrent.Future 中沒有這個功能(據(jù)推測是為了鼓勵使用前面的方式)。我們暫時不討論Scala背后如何執(zhí)行的,先實現(xiàn)類似 CompletableFuture 的操作。它接受一個 Future 作為輸入,然后返回一個 Future,這個 Future 在后臺任務完成時候執(zhí)行完成。但是,如果底層的 Future 執(zhí)行的時間太長,就或拋出異常:
public staticCompletableFuture within(CompletableFuture future, Duration duration) { final CompletableFuture timeout = failAfter(duration); return future.applyToEither(timeout, Function.identity()); }
這引導我們實現(xiàn)最終的、清晰的、靈活的方法:
final CompletableFutureresponseFuture = within( asyncCode(), Duration.ofSeconds(1)); responseFuture .thenAccept(this::send) .exceptionally(throwable -> { log.error("Unrecoverable error", throwable); return null; });
感謝各位的閱讀,以上就是“Java中怎么處理異步超時的問題”的內容了,經(jīng)過本文的學習后,相信大家對Java中怎么處理異步超時的問題這一問題有了更深刻的體會,具體使用情況還需要大家實踐驗證。這里是創(chuàng)新互聯(lián),小編將為大家推送更多相關知識點的文章,歡迎關注!