使用ExecutorService时出现超时问题

Question

我有一个数十万可调用对象的列表。当每一个运行时，它都会根据给定的值执行一个可能很长的计算。正因为如此，我希望异步运行每个任务（最好是通过使用某种执行器），并在30秒后检索每次计算的结果，取消那些没有及时完成的结果。（所得值在其他地方使用。）

到目前为止，我就是这样实现它的：

private void process() {
    class Runner implements Callable<BigDecimal> {
        final int x, y;

        Runner(int x, int y) {
            this.x = x;
            this.y = y;
        }

        @Override
        public BigDecimal call() {
            BigDecimal gWidth = xMax.subtract(xMin), gHeight = yMax.subtract(yMin);
            BigDecimal gy = gHeight.multiply(BigDecimal.valueOf(-y)).divide(BigDecimal.valueOf(height)).add(yMax);
            BigDecimal gx = gWidth.multiply(BigDecimal.valueOf(x)).divide(BigDecimal.valueOf(width)).add(xMin);

            // The calculation begins when this method is called
            BigDecimal result = calculateAt(gx, gy);
            return result;
        }
    }

    ExecutorService exec = Executors.newCachedThreadPool();
    List<Runner> runners = new ArrayList<>();
    for (int y = 0; y < height; y++) {
        for (int x = 0; x < width; x++) {
            runners.add(new Runner(x, y));
        }
    }

    try {
        List<Future<BigDecimal>> results = exec.invokeAll(runners, 30, TimeUnit.SECONDS);
        for (Future<BigDecimal> future : results) {
            // Check if the future's task was cancelled and process the results
        }
    } catch (InterruptedException | ExecutionException ex) {
        ex.printStackTrace();
    }
    exec.shutdown();
}

// Extra variables and methods
BigDecimal xMin = BigDecimal.valueOf(-7),
           xMax = BigDecimal.valueOf(7),
           yMin = BigDecimal.valueOf(-7),
           yMax = BigDecimal.valueOf(7);
int height = 850, width = 850;
private BigDecimal calculateAt(BigDecimal x, BigDecimal y) {
    try {
        // Just to simulate a potential execution time
        Thread.sleep((ThreadLocalRandom.current().nextInt(45) + 1) * 1000);
    } catch (InterruptedException ex) {
        ex.printStackTrace();
    }
    return BigDecimal.ONE;
}

ArrayListrunners存储每个要执行的callable，然后将其发送到ExecutorService以运行所有任务。我遇到的问题是，任务似乎是同步启动的，而在超时30秒后，只有前40或5万个任务完成了，更不用说开始执行了。

似乎正在发生的是ExecutorService.InvokeAll方法只允许30秒的窗口用于列表中的所有任务开始和完成执行。相反，我需要的是在每个任务的基础上开始这个30秒的窗口，也就是说，一旦任务已经开始，就允许30秒的时间来完成它。InvokeAll似乎不能做到这一点，至少在使用NewCachedThreadPool时不能做到这一点。有Java库或其他实现方法吗？

Answer 1

我认为您可以使用completablefuture来解决问题。

例如，下面是一段基于问题片段的代码：

private static final ExecutorService executor = Executors.newCachedThreadPool();
private static final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(100);

private static void processAsync() {
  List<CompletableFuture<Object>> futureList = IntStream.range(0, height).boxed()
    .flatMap(y -> IntStream.range(0, width).boxed().map(x -> new Runner(x, y)))
    .map(runner ->
      CompletableFuture.anyOf(
        CompletableFuture.supplyAsync(runner, executor),
        timeout(Duration.ofSeconds(30))
      ).exceptionally(throwable -> {
        // timeout is handled here
        return BigDecimal.ZERO;
      })
    )
    .collect(Collectors.toList());

  CompletableFuture.allOf(futureList.toArray(new CompletableFuture<?>[0]))
    .thenAccept(v -> {
      List<BigDecimal> results = futureList.stream()
        .map(CompletableFuture::join)
        .map(r -> (BigDecimal) r)
        .collect(Collectors.toList());
      // process final results
      BigDecimal sum = results.stream().reduce(BigDecimal.ZERO, BigDecimal::add);
      System.out.println("Final sum: " + sum);
    })
    .exceptionally(throwable -> {
      System.out.println("Failed allOf with " + throwable);
      return null;
    });
}

private static CompletableFuture<BigDecimal> timeout(Duration duration) {
  CompletableFuture<BigDecimal> future = new CompletableFuture<>();
  scheduler.schedule(
    () -> future.completeExceptionally(new TimeoutException("Timeout " + Thread.currentThread().getName())), duration.toMillis(), MILLISECONDS);
  return future;
}

private static class Runner implements Supplier<BigDecimal> {...

这里的主要思想是使用completablefuture.anyof，并将其应用于有用任务的completablefuture和超时任务的COPMPLETablefuture。通过使用ScheduleDexecutorService和CompletableFuture.CompleteExceptional来实现超时。因此，基本上，anyof返回result或TimeoutException。