Java中的CompletableFuture超时使用

我喜欢Java 8的CompletableFuture，但它有它的缺点： 惯用的超时处理就是其中之一。

JAVA 8我们只能收集异常信息，再次执行什么的（以下是JAVA8解决超时的方式，获取结果后你该做什么做什么）：

//我们让list里传入方法的参数1号报错和5号超时
public class Test4 {

    private static final ScheduledExecutorService scheduler =
            Executors.newScheduledThreadPool(
                    1,
                    r -> {
                        Thread thread = new Thread(r);
                        thread.setName("failAfter-%d");
                        thread.setDaemon(true);
                        return thread;
                    });

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        System.out.println("-------开始------");
        final CompletableFuture<Object> oneSecondTimeout = failAfter(Duration.ofSeconds(2))
                .exceptionally(xxx -> "超时");
        List<Object> collect = Stream.of("1", "2", "3", "4", "5", "6", "7")
                .map(x -> CompletableFuture.anyOf(createTaskSupplier(x)
                        , oneSecondTimeout))
                .collect(Collectors.toList())
                .stream()
                .map(CompletableFuture::join)
                .collect(Collectors.toList());
        System.out.println("-------结束------");
        System.out.println(collect.toString());
    }

    public static CompletableFuture<String> createTaskSupplier(String x) {
        return CompletableFuture.supplyAsync(getStringSupplier(x))
                .exceptionally(xx -> xx.getMessage());
    }


    public static Supplier<String> getStringSupplier(String text) {
        return () -> {
            System.out.println("开始 " + text);
            if ("1".equals(text)) {
                throw new RuntimeException("运行时错误");
            }
            try {
                if ("5".equals(text))
                    setSleepTime(20);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("停止 " + text);
            return text+"号";
        };
    }

    static void  setSleepTime(int SleepTime) throws InterruptedException {
        TimeUnit.SECONDS.sleep(SleepTime);
    }

    public static <T> CompletableFuture<T> failAfter(Duration duration) {
        final CompletableFuture<T> promise = new CompletableFuture<>();
        scheduler.schedule(() -> {
            final TimeoutException ex = new TimeoutException("Timeout after " + duration);
            return promise.completeExceptionally(ex);
        }, duration.toMillis(), TimeUnit.MILLISECONDS);
        return promise;
    }

}



//执行结果：
-------开始------
开始 1
开始 2
停止 2
开始 3
停止 3
开始 4
停止 4
开始 5
开始 6
停止 6
开始 7
停止 7
-------结束------
[java.lang.RuntimeException: 运行时错误, 2号, 3号, 4号, 超时, 6号, 7号]

 

幸运的是，JDK 9带来了两种新方法，可以为每个人提供渴望的功能 - 这对于确保在使用异步处理时的正确弹性至关重要。

 

在这篇超短篇文章中，尝试帮助大家对这个新API方法的认识。

 

CompletableFuture＃orTimeout

简单地说，在调用上述方法之后，如果未在指定的超时内完成，将来会抛出ExecutionException。

 

一个简单的例子：

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(this::computeEndlessly)
  .orTimeout(1, TimeUnit.SECONDS);

future.get(); // java.util.concurrent.ExecutionException after waiting for 1 second

由于设置了timeout为1秒，那么在get那里等待1秒后抛错

CompletableFuture#completeOnTimeout

在这种情况下，我们可以在达到超时后返回默认值：

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(this::computeEndlessly)
  .completeOnTimeout(42, 1, TimeUnit.SECONDS);

Integer result = future.get(); // 42

超时1秒后不是报错，而是返回了预设的42这个值，前提条件是你必须预设默认值。

 

就这么简单，虽然我不一定喜欢我们总是被迫预先计算默认值的事实。