如何在 Java 中为 CompletableFuture 执行资源清理？答案

【问题标题】：How to perform resource cleanup for CompletableFuture in Java?如何在 Java 中为 CompletableFuture 执行资源清理？
【发布时间】：2018-10-12 08:39:50
【问题描述】：

我在CompletableFuture 中有一段代码，如果有异常则执行重试，否则完成任务。我已将资源传递给Supplier 和Consumer 以执行任务，并希望在所有任务完成后关闭这些资源（重试3 次后success/exception）。

这是一段代码：

Supplier mySupplier = new MySupplier(localContext);
CompletableFuture<String> future = CompletableFuture.supplyAsync(mySupplier);
for(int j = 0; j < (retryCount - 1); j++) {
    LOGGER.debug("MySupplier accept() Retry count: "+j);
    future = future.handleAsync((value, throwable) -> throwable == null? CompletableFuture.completedFuture(value): CompletableFuture.supplyAsync(mySupplier)).thenComposeAsync(Function.identity());
}

我打算将它放在我的供应商的 finally 块下，但如果发生第一个异常，资源将被关闭，我需要它们进行接下来的两次重试。

1) 如何让它工作？

2) 还有什么方法可以只在异常情况下打印重试次数？

【问题讨论】：

标签： java multithreading java-threads completable-future

【解决方案1】：

由于您似乎并不关心中间结果，因此最简单的解决方案是简单地将您的 Supplier 包装在另一个处理重试的文件中：

class SupplierRetrier<T> implements Supplier<T> {
    private static final Logger LOGGER = LoggerFactory.getLogger(SupplierRetrier.class);
    final Supplier<T> wrappee;
    final int maxRetries;

    SupplierRetrier(Supplier<T> wrappee, int maxRetries) {
        Objects.requireNonNull(wrappee);
        if (maxRetries <= 0) {
            throw new IllegalArgumentException("maxRetries must be more than 0: " + maxRetries);
        }
        this.wrappee = wrappee;
        this.maxRetries = maxRetries;
    }

    @Override
    public T get() {
        RuntimeException lastException = null;
        for (int i = 0; i < maxRetries; i++) {
            try {
                LOGGER.info("MySupplier accept() Retry count: "+i);
                return wrappee.get();
            } catch (RuntimeException e) {
                lastException = e;
            }
        }
        throw lastException;
    }
}

然后您可以简单地使用它：

CompletableFuture<String> future = CompletableFuture.supplyAsync(
        new SupplierRetrier<>(mySupplier, retryCount));

为了清理您的上下文，只需在生成的 future 上添加一个 whenComplete() 调用。无论未来的结果如何，这都会执行。

future.whenComplete((r, e) -> {
    try {
        localContext.close();
    } catch (Exception e2) {
        throw new RuntimeException("Failed to close context", e2);
    }
});

【讨论】：

Didier L，它只解决了我的第二个问题，知道如何解决第一个问题吗？
@Lother 抱歉，我忘记了那部分，我也为此添加了解决方案。

【解决方案2】：

1) 对于资源清理，请使用 whenComplete 或 whenCompleteAsync

2) 对于重试计数，请使用长度为 1 或 AtomicInteger 的 int[]。（无论是否抛出Exception，该值都可用）

int[] retryCounter = { 0 };
// AtomicInteger retryCounter = new AtomicInteger();

for (int i = 0; i < noOfRetries; i++)
{
  CompletableFuture<CompletableFuture<String>> handleAsync = cf.handleAsync((result, throwable) ->
    {
      if (throwable == null)
        return CompletableFuture.completedFuture(result);

      retryCounter[0]++;
      // retryCounter.incrementAndGet();

      return CompletableFuture.supplyAsync(supplier);
    });
  cf = handleAsync.thenCompose(Function.identity());
}

cf = cf.whenCompleteAsync((result, throwable) ->
  {
    System.out.println("Clean up");

    System.out.println("Retry count: " + retryCounter[0]);
    // System.out.println("Retry count: " + retryCounter.get());
  });

System.out.println("Wating for result...");
System.out.println("Result: " + cf.get());

【讨论】：

谢谢文卡塔拉朱。