Использование java 8 CompletableFuture рекурсивным образом - PullRequest
Новая
       50

Использование java 8 CompletableFuture рекурсивным образом

1 голос
/ 27 мая 2020

Мне нужно загрузить записи из базы данных, а затем последовательно выполнить над ними некоторые действия, и я хочу реорганизовать наш существующий код, чтобы воспользоваться преимуществами API CompletableFuture

основная идея c равно

  1. a. загрузить записи в очередь b. для каждой записи в очереди выполнить:
    1. выполнить задачу A (занимает определенное время)
    2. выполнить задачу B (занимает еще больше времени) на основе задачи A
    3. выполнить задачу C (также требуется время)

некоторые записи, конечно, заканчиваются раньше, чем другие, поэтому я хочу, чтобы, как только один будет завершен, другой (из очереди) начнет выполнение

У меня есть следующий код 

gamesQueue = new PriorityQueue<Game>();
completable = new LinkedList<CompletableFuture<String>>();
Instant start = Instant.now();
Executor pool = Executors.newCachedThreadPool();
DB.getInstance().getConn().useHandle(handle -> {
   handle.attach(GameService.class).list().stream().forEach(game -> gamesQueue.add(game));
});

, это загрузит мои записи, тогда я выполняя следующее 

while (!gamesQueue.isEmpty()) {

        CompletableFuture<String> step1 = CompletableFuture.supplyAsync(this::getCurrentGame, pool)
                .thenApply(CapitalLetterTask::capitalize).thenApply(game-> {
                    if(game.isEndedExceptionaly()) {
                        handleNumber5Exception(game);
                        throw new CompletionException(new NumberFiveException(game, "got number 5"));
                    }
                    return game;
                })
                .thenApply(AddStarTask::addStarToGame).exceptionally(ExceptionHandler::handleWeirdEx);
        completable.add(step1);

    }


    while (!completable.isEmpty()) {
        CompletableFuture<String> completed = completable.poll();
        if (completed.isDone()) {
            try {
                log.debug("result: {}", completed.get());
            } catch (Exception ex) {

            }

        } else if(completed.isCompletedExceptionally()) {
            try{ //never enters here
                log.debug("This is a bad result:{} ", completed.get());
            }catch(Exception ex) {

            }

        }
         else {
            completable.add(completed);
        }
    }
    Instant end = Instant.now();
    Duration duration = Duration.between(start, end);
    log.debug("Total time: {}", duration.getSeconds());
    loadGames();  <-- call itself
}

, я не доволен своей обработкой исключений, более того, я вижу, что даже фьючерсы, которые должны были быть пойманы на первом этапе, все еще передаются дальше по цепочке

what я хотел бы спросить, как реализовать его таким образом, чтобы каждый метод мог генерировать свое собственное исключение, пойманное и обработанное, не передавая его дальше по цепочке

...