Как уменьшить поток фьючерсов в Java? - PullRequest
Новая
       27

Как уменьшить поток фьючерсов в Java?

7 голосов
/ 17 января 2020

Предполагая, что я дал Stream из Futures, который я хочу уменьшить, вызвав метод Stream#reduce. Но я не хочу уменьшать саму Futures, а результат Future (Future#get). Проблема в том, что метод get может выдать ExecutionException и не дает результата в этом случае.

Это причина, по которой 

Stream<Future<Integer>> stream = ...;
BinaryOperator<Integer> sum = (i1, i2) -> i1 + i2;  
stream.map(future -> future.get())
      .reduce(sum); // does not work, get needs to handle exceptions!

Итак, я должен поймать исключения: 

stream.map(future -> {
    Integer i = null;
    try {
        i = future.get();
    } catch (InterruptedException e) {
    } catch (ExecutionException e) {}
    return i;
}).reduce(sum);

Но при таком подходе у меня могут возникнуть проблемы, потому что могут появиться значения null.

Итак, чтобы избавиться от них, я пришлось бы отфильтровать те, где появился ExecutionException: 

stream.filter(future -> {
    Integer i = null;
    try {
        i = future.get();
    } catch (InterruptedException e) {
    } catch (ExecutionException e) {
    }
    return i != null;
})
.map(future -> {
    Integer i = null;
    try {
        i = future.get();
    } catch (InterruptedException e) {
    } catch (ExecutionException e) {
    }
    return i;
}).reduce(sum);

Я думаю, этот код будет работать .. Но я не хочу верить, что это единственный и самый умный способ уменьшить фьючерсы.

Есть идеи или предложения?

Ответы [ 3 ]

4 голосов
/ 17 января 2020

Вы можете сначала извлечь значение из будущего, а затем отфильтровать нуль: 

Integer result = stream
    .map(future -> {
        try {
          return future.get();
        } catch (InterruptedException | ExecutionException e) {
        }
        return null; })
    .filter(Objects::nonNull)
    .reduce(sum)
    .orElse(0);
2 голосов
/ 17 января 2020

Вы можете сделать это, используя flatMap: 

    public static void main(String[] args) {
        Stream<Future<Integer>> yourStream = null;

        int sum = yourStream.flatMap(YourClass::unpack)
            .mapToInt(Integer::intValue)
            .sum()
            .orElse(0);
    }

    public static <T> Stream<T> unpack(Future<T> future) {
        try {
            return Stream.of(future.get());
        } catch (InterruptedException e) {
            return Stream.empty();
        } catch (ExecutionException e) {
            return Stream.empty();
        }
    }
2 голосов
/ 17 января 2020

Один из способов упростить это может быть: 

void reduceImpl(Stream<Future<Integer>> stream) {
    Optional<Integer> integerOptional = stream
            .map(this::transform)
            .filter(Objects::nonNull)
            .reduce(Integer::sum);
}

private Integer transform(Future<Integer> future) {
    try {
        return future.get();
    } catch (InterruptedException | ExecutionException e) {
        return null; // should ideally be handled properly
    }
}
...