如何使用 java future 包装供应商，该未来在超时时返回值，但在后台继续运行（具有取消选项）

Question

例如，我有一个供应商可能需要时间运行：

Supplier<Integer> numLoader = sneaky(() -> {
    Thread.sleep(10000);
    System.out.println("5 Calculated!");
    return 5;
});

* sneaky 只是一个转换为运行时异常的实用程序。

我希望能够做这样的事情：

Future<Integer> future = createFutureValueOnTimeout(-1, numLoader);
// numLoader takes 10 seconds to complete so -1 is returned.
int num = future.get(1000, TimeUnit.MILLISECONDS);
if (resourcesAreLow()) {
    future.cancel(true);
}
doSomethingWithTheValue(num);

我还有一个createFutureValueOnTimeout的部分实现：

private static <V> Future<V> createFutureValueOnTimeout(V v, Supplier<V> supplier) {
    CompletableFuture<V> completableFuture = CompletableFuture.supplyAsync(supplier);
    return new FutureDecorator<V>(completableFuture) {
        @Override
        public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException {
            return completableFuture.completeOnTimeout(v, timeout, unit).get();
        }
    };
}

问题在于，调用取消时，睡眠不会中断。

1. 我怎样才能让取消生效
2. 有没有更简单的方法可以在超时时返回值

完整测试：

public class TimeoutTest {
    @SneakyThrows
    @Test
    public void testTimeout() {
        int loadTimeMillis = 10000;
        Supplier<Integer> numLoader = () -> {
            try {
                // Simulate long operation
                Thread.sleep(loadTimeMillis);
            } catch (InterruptedException e) {
                System.out.println("Interrupted! message: " + e.getMessage());
                throw Lombok.sneakyThrow(e);
            }
            System.out.println("5 Calculated!");
            return 5;
        };

        Future<Integer> future = createFutureValueOnTimeout(-1, numLoader);
        
        long start = System.currentTimeMillis();

        // numLoader takes 10 seconds to complete so -1 is returned.
        int num = future.get(1000, TimeUnit.MILLISECONDS);

        System.out.println("Got: num: " + num + ". time: " + (System.currentTimeMillis() - start));

        if (resourcesAreLow()) {
            future.cancel(true);
        }
        // Don't stop the test. Give time for the cancel to kick in.
        Thread.sleep(loadTimeMillis);
        System.out.println("Finished. Time: " + (System.currentTimeMillis() - start));
    }

    private boolean resourcesAreLow() {
        return true;
    }

    private static <V> Future<V> createFutureValueOnTimeout(V v, Supplier<V> supplier) {
        CompletableFuture<V> completableFuture = CompletableFuture.supplyAsync(supplier);
        return new FutureDecorator<V>(completableFuture) {
            @Override
            public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException {
                return completableFuture.completeOnTimeout(v, timeout, unit).get();
            }
        };
    }

    private static class FutureDecorator<V> implements Future<V> {
        private final Future<V> inner;

        private FutureDecorator(Future<V> inner) {this.inner = inner;}

        @Override
        public boolean cancel(boolean mayInterruptIfRunning) {
            return inner.cancel(mayInterruptIfRunning);
        }

        @Override
        public boolean isCancelled() {
            return inner.isCancelled();
        }

        @Override
        public boolean isDone() {
            return inner.isDone();
        }

        @Override
        public V get() throws InterruptedException, ExecutionException {
            return inner.get();
        }

        @Override
        public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
            return inner.get(timeout, unit);
        }
    }
}

输出：（请注意缺少“已中断！”消息）

Got: num: -1. time: 1007
5 Calculated!
Finished. Time: 11021

Answer 1

您可以将支持取消的执行器/未来API与完全未来 组合使用：

public static <R> CompletableFuture<R> withInterruptionSupport(Callable<R> c) {
    CompletableFuture<R> cf = new CompletableFuture<>();
    FutureTask<R> ft = new FutureTask<R>(c) {
        @Override
        protected void set(R v) {
            super.set(v);
            cf.complete(v);
        }
        @Override
        protected void setException(Throwable t) {
            super.setException(t);
            cf.completeExceptionally(t);
        }
    };
    cf.defaultExecutor().execute(ft);
    cf.whenComplete((x,y) -> ft.cancel(true));
    return cf;
}

由于在实际函数中支持中断通常意味着处理中断异常，因此使用可调用而不是供应商，非常方便，因此允许抛出此异常。

只要CompletableFuture完成，支持中断取消的Future就会无条件取消，因为只要完成源于任务本身，未来已经完成，后续取消将被忽略。

这意味着，我们不需要在这里区分不同的完成可能性。< code>completeOnTimeout不仅起作用，您还可以对< code>CompletableFuture调用< code>cancel(…)，它将中断< code>Callable的计算(尽管< code>boolean参数仍然是不相关的)。即使在没有等待超时的情况下调用带有替代结果的< code>complete,也会中断现在已经过时的评估。

所以，下面的作品:

for(int timeout: new int[] { 5, 15 }) {
    System.out.println("with timeout of " + timeout);
    Integer i = withInterruptionSupport(() -> {
            Thread.sleep(10000);
            System.out.println("5 Calculated!");
            return 5;
        })
        .completeOnTimeout(42, timeout, TimeUnit.SECONDS)
        .join();
    System.out.println("got " + i);
}
ForkJoinPool.commonPool().awaitQuiescence(1, TimeUnit.DAYS);

with timeout of 5
got 42
with timeout of 15
5 Calculated!
got 5