如何强制超时和取消异步完成未来作业

Question

我正在使用Java8，我想知道对3个异步作业强制超时的推荐方法，我将执行异步并从未来检索结果。请注意，所有3个作业的超时是相同的。如果超出时间限制，我还想取消作业。

我在想这样的事情：

// Submit jobs async
List<CompletableFuture<String>> futures = submitJobs(); // Uses CompletableFuture.supplyAsync

List<CompletableFuture<Void>> all = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));

try {
    allFutures.get(100L, TimeUnit.MILLISECONDS);
} catch (TimeoutException e){
   for(CompletableFuture f : future) {
      if(!f.isDone()) {
         /*
         From Java Doc:
         @param mayInterruptIfRunning this value has no effect in this
             * implementation because interrupts are not used to control
             * processing.
         */
         f.cancel(true);
      }
   }
}

List<String> output = new ArrayList<>();
for(CompeletableFuture fu : futures) {
   if(!fu.isCancelled()) { // Is this needed?
      output.add(fu.join());
   }
}

return output;

1. 像这样的东西有用吗？有更好的方法吗？
2. 如何正确地取消未来？Javadoc说，线程不能被中断？所以，如果我取消一个未来，并调用κ（），我会立即得到结果，因为线程不会被中断吗？
3. 在等待结束后，是否建议使用run（）或get（）来获取结果？

Answer 1

调用cancel后，您不能加入furure，因为您会得到一个异常。

终止计算的一种方法是让它引用未来并定期检查:如果它被取消了，从内部中止计算。如果计算是一个循环，在每一次迭代中你都可以进行检查，那么就可以做到这一点。

你需要它是一个CompletableFuture吗？因为另一种方法是避免使用CompleatableFuture，而是使用一个简单的Future或FutureTask：如果您使用Exec在调用时执行它future.cancel（true）将在拥有时终止计算。

回答问题:“调用join()，我会立即得到结果吗”。

不，你不会马上得到它，它会挂起等待完成计算：没有办法强迫一个需要很长时间才能在较短时间内完成的计算。

你可以称之为未来。complete（value）提供一个值，该值将由具有该未来引用的其他线程用作默认结果。

Answer 2

值得注意的是，在CompletableFuture上调用cancel实际上与在当前阶段调用completeExceptionally是一样的。取消不会影响之前的阶段。说完后说道:

1. 原则上，假设不需要上游取消，这样的东西就可以工作（从伪代码的角度来看，上面有语法错误）。
2. CloetableFuture取消不会中断当前线程。取消将导致所有下游阶段立即被C（取消）异常触发（将短路执行流程）。
3. 在调用者愿意无限期等待的情况下，“加入”和“获取”实际上是相同的。加入处理为您包装选中的异常。如果调用者想要超时，将需要get。

包括一个细分来说明取消时的行为。请注意，下游流程不会启动，但上游流程即使在取消后也会继续。

    public static void main(String[] args) throws Exception
    {
        int maxSleepTime = 1000;
        Random random = new Random();
        AtomicInteger value = new AtomicInteger();
        List<String> calculatedValues = new ArrayList<>();
        Supplier<String> process = () -> { try { Thread.sleep(random.nextInt(maxSleepTime)); System.out.println("Stage 1 Running!"); } catch (InterruptedException e) { e.printStackTrace(); } return Integer.toString(value.getAndIncrement()); };
        List<CompletableFuture<String>> stage1 = IntStream.range(0, 10).mapToObj(val -> CompletableFuture.supplyAsync(process)).collect(Collectors.toList());
        List<CompletableFuture<String>> stage2 = stage1.stream().map(Test::appendNumber).collect(Collectors.toList());
        List<CompletableFuture<String>> stage3 = stage2.stream().map(Test::printIfCancelled).collect(Collectors.toList());
        CompletableFuture<Void> awaitAll = CompletableFuture.allOf(stage2.toArray(new CompletableFuture[0]));
        try
        {
            /*Wait 1/2 the time, some should be complete. Some not complete -> TimeoutException*/
            awaitAll.get(maxSleepTime / 2, TimeUnit.MILLISECONDS);
        }
        catch(TimeoutException ex)
        {
            for(CompletableFuture<String> toCancel : stage2)
            {
                boolean irrelevantValue = false;
                if(!toCancel.isDone())
                    toCancel.cancel(irrelevantValue);
                else
                    calculatedValues.add(toCancel.join());
            }
        }
        System.out.println("All futures Cancelled! But some Stage 1's may still continue printing anyways.");
        System.out.println("Values returned as of cancellation: " + calculatedValues);
        Thread.sleep(maxSleepTime);
    }

    private static CompletableFuture<String> appendNumber(CompletableFuture<String> baseFuture) 
    {
        return baseFuture.thenApply(val -> {  System.out.println("Stage 2 Running"); return "#" + val; }); 
    }
    
    private static CompletableFuture<String> printIfCancelled(CompletableFuture<String> baseFuture) 
    { 
        return baseFuture.thenApply(val ->  { System.out.println("Stage 3 Running!"); return val; }).exceptionally(ex -> { System.out.println("Stage 3 Cancelled!"); return ex.getMessage(); }); 
    }

如果需要取消上游进程（例如：取消某些网络调用），则需要自定义处理。