java8中CompletableFuture異步處理超時的方法
Java 8 的 CompletableFuture 並沒有 timeout 機制,雖然可以在 get 的時候指定 timeout,但是我們知道get 是一個同步堵塞的操作。怎樣讓 timeout 也是異步的呢?Java 8 內有內建的機制支持,一般的實現方案是啟動一個 ScheduledThreadpoolExecutor
線程在 timeout 時間后直接調用 CompletableFuture.completeExceptionally(new TimeoutException())
,然后用 acceptEither()
或者 applyToEither
看是先計算完成還是先超時。
在 java 9 引入了 orTimeout
和 completeOnTimeOut
兩個方法支持 異步 timeout 機制:
- public CompletableFuture orTimeout(long timeout, TimeUnit unit) : completes the CompletableFuture with a TimeoutException after the specified timeout has elapsed.
- public CompletableFuture completeOnTimeout(T value, long timeout, TimeUnit unit) : provides a default value in the case that the CompletableFuture pipeline times out.
內部實現上跟我們上面的實現方案是一模一樣的,只是現在不需要自己實現了。
實際上hystrix等熔斷的框架,其實現線程Timeout之后就關閉線程,也是基於同樣的道理,所以我們可以看到hystrix中會有一個Timer Thread。
超時工具類
package com.example.netty.async; import java.util.concurrent.*; import java.util.function.Function; /** * java8中CompletableFuture異步處理超時的方法 * * Java 8 的 CompletableFuture 並沒有 timeout 機制,雖然可以在 get 的時候指定 timeout,是一個同步堵塞的操作。怎樣讓 timeout 也是異步的呢?Java 8 內有內建的機 * 制支持,一般的實現方案是啟動一個 ScheduledThreadpoolExecutor 線程在 timeout 時間后直接調用 CompletableFuture.completeExceptionally(new TimeoutException()), * 然后用acceptEither() 或者 applyToEither 看是先計算完成還是先超時: * * 在 java 9 引入了 orTimeout 和 completeOnTimeOut 兩個方法支持 異步 timeout 機制: * * public CompletableFuture orTimeout(long timeout, TimeUnit unit) : completes the CompletableFuture with a TimeoutException after the specified timeout has elapsed. * public CompletableFuture completeOnTimeout(T value, long timeout, TimeUnit unit) : provides a default value in the case that the CompletableFuture pipeline times out. * 內部實現上跟我們上面的實現方案是一模一樣的,只是現在不需要自己實現了。 * * 實際上hystrix等熔斷的框架,其實現線程Timeout之后就關閉線程,也是基於同樣的道理,所以我們可以看到hystrix中會有一個Timer Thread * * * @author luliang * @date 2021-02-24 9:48 */ public class CompletableFutureTimeout { /** * Singleton delay scheduler, used only for starting and * cancelling tasks. */ static final class Delayer { static ScheduledFuture<?> delay(Runnable command, long delay, TimeUnit unit) { return delayer.schedule(command, delay, unit); } static final class DaemonThreadFactory implements ThreadFactory { @Override public Thread newThread(Runnable r) { Thread t = new Thread(r); t.setDaemon(true); t.setName("CompletableFutureDelayScheduler"); return t; } } static final ScheduledThreadPoolExecutor delayer; // 注意,這里使用一個線程就可以搞定 因為這個線程並不真的執行請求 而是僅僅拋出一個異常 static { (delayer = new ScheduledThreadPoolExecutor( 1, new CompletableFutureTimeout.Delayer.DaemonThreadFactory())). setRemoveOnCancelPolicy(true); } } public static <T> CompletableFuture<T> timeoutAfter(long timeout, TimeUnit unit) { CompletableFuture<T> result = new CompletableFuture<T>(); // timeout 時間后 拋出TimeoutException 類似於sentinel / watcher CompletableFutureTimeout.Delayer.delayer.schedule(() -> result.completeExceptionally(new TimeoutException()), timeout, unit); return result; } /** * 哪個先完成 就apply哪一個結果 這是一個關鍵的API,exceptionally出現異常后返回默認值 * * @param t * @param future * @param timeout * @param unit * @param <T> * @return */ public static <T> CompletableFuture<T> completeOnTimeout(T t, CompletableFuture<T> future, long timeout, TimeUnit unit) { final CompletableFuture<T> timeoutFuture = timeoutAfter(timeout, unit); return future.applyToEither(timeoutFuture, Function.identity()).exceptionally((throwable) -> t); } /** * 哪個先完成 就apply哪一個結果 這是一個關鍵的API,不設置默認值,超時后拋出異常 * * @param t * @param future * @param timeout * @param unit * @param <T> * @return */ public static <T> CompletableFuture<T> orTimeout(T t, CompletableFuture<T> future, long timeout, TimeUnit unit) { final CompletableFuture<T> timeoutFuture = timeoutAfter(timeout, unit); return future.applyToEither(timeoutFuture, Function.identity()).exceptionally((throwable) -> t); } }
測試類
package com.example.netty.async; import java.util.concurrent.*; /** * @author luliang * @date 2021-02-24 10:05 */ public class CompletableFutureTimeout2 { public static void main(String[] args) throws ExecutionException, InterruptedException { CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } return 10; }); CompletableFuture<Integer> within = CompletableFutureTimeout.completeOnTimeout(1, future, 1, TimeUnit.SECONDS); System.out.println(within.get()); CompletableFuture<String> futureStr = CompletableFuture.supplyAsync(() -> { try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } return "正常執行"; }); CompletableFuture<String> withinStr = CompletableFutureTimeout.completeOnTimeout("異常執行", futureStr, 1, TimeUnit.SECONDS); System.out.println(withinStr.get()); } }