CompletableFuture 异步超时 和取消

2021-12-22  本文已影响0人  NazgulSun

异步调用超时 和 cancel 的一些问题

通常我们都是使用阻塞方法的时候等待超时,比如 Future.get();
有一个case,比如一个List<Task>,比如10个,想要整体的任务在2s内完成,否则就超时,
超时的时候,正确的做法最好是立马停止运行中的线程任务;
FutureTask 有一个 canclel(true/false)的方法;
参考: https://www.jianshu.com/p/9fc446c2c1be
Runnning的线程,在执行CPU是无法取消的,只有blocked/new/Runnable的线程,才可以尝试取消;
为什么说尝试,因为调用的是interrupt方法,只会对 sleep,wait,join等方法有效;
会设置interrupted标志位;所以想要一个可以cancell的task,需要更多的设计细节;

从目前来看,competableFuture 底层使用forkjoinPool,cancel方法是无效的;
所以想要interrupt线程,还是需要用futureTask;

另外,使用completableFuture,可以实现异步超时, jdk9,已经有原生的实现,但是在jdk8,需要自己做类似下面的实现,
需要利用applyToEigther的特性;

    static ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(
            Runtime.getRuntime().availableProcessors());


    public static <T> CompletableFuture<T> failAfter(Duration duration){
        /// need a schedular executor
        final CompletableFuture<T> timer = new CompletableFuture<>();
        scheduler.schedule(()->{
            TimeoutException timeoutException = new TimeoutException("time out after "+ duration.getSeconds());
            return timer.completeExceptionally(timeoutException);
        },duration.toMillis(), TimeUnit.MILLISECONDS);
        return timer;
    }

    public static <T> CompletableFuture<T> within(CompletableFuture<T> taskFuture, Duration duration){
        CompletableFuture<T> timeoutWatcher = failAfter(duration);
        return taskFuture.applyToEither(timeoutWatcher, Function.identity());
    }


    public static void run(){
        /// do logical here
        CompletableFuture<String> slowlyPrinter = CompletableFuture.supplyAsync(
                ()->{
                    String msg = "say hello!";
                    try {
                        Integer  sl = ThreadLocalRandom.current().nextInt(20);
                        Thread.sleep(sl * 1000);
                        System.out.println("finish slow printer after + "+ sl);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return msg;
                });

        CompletableFuture<String> chains = within(slowlyPrinter, Duration.ofSeconds(10));
        chains.thenAccept(System.out::println)
                .exceptionally((e)->{
                    System.out.println(e.getMessage());
                    return null;
                });

    }

    public static void main(String[] args){
        for(int i=0; i< 1; i++)
            run();
    }
上一篇下一篇

猜你喜欢

热点阅读