Как выполнить очистку ресурса для CompletableFuture в Java? - PullRequest
Новая
       32

Как выполнить очистку ресурса для CompletableFuture в Java?

0 голосов
/ 12 октября 2018

У меня есть фрагмент кода в CompletableFuture, который выполняет повторную попытку, если есть исключения, иначе задание завершается.Я передал ресурс Supplier и Consumer для выполнения задачи и хочу закрыть этот ресурс после завершения всех задач (success/exception после 3 попыток).

Вот частькода: 

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) Также есть способы вывести число повторов только в случае исключения?

Ответы [ 2 ]

0 голосов
/ 12 октября 2018

1) Для очистки ресурса используйте whenComplete или whenCompleteAsync

2) Для подсчета повторов используйте int[] длины 1 или AtomicInteger.(Это значение доступно независимо от того, выброшено 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());
0 голосов
/ 12 октября 2018

Поскольку вам не нужны промежуточные результаты, самое простое решение - просто обернуть ваш 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));

Чтобы очистить свой контекст, просто добавьте вызов whenComplete() к полученному будущему.Это будет выполнено независимо от результата будущего. 

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