Java 8 Generics: сокращение потока потребителей для одного потребителя

Ответ 1

Вы используете Stream.reduce(accumulator) версию Stream.reduce(accumulator) которая имеет следующую подпись:

Optional<T> reduce(BinaryOperator<T> accumulator);

BinaryOperator<T> accumulator может принимать только элементы типа T, но у вас есть:

<? extends Consumer<? super T>>

Я предлагаю вам использовать Stream.reduce(...) версию метода Stream.reduce(...):

<U> U reduce(U identity,
             BiFunction<U, ? super T, U> accumulator
             BinaryOperator<U> combiner);

BiFunction<U,? super T, U> accumulator BiFunction<U,? super T, U> accumulator может принимать параметры двух разных типов, имеет менее ограничительную границу и более подходит для вашей ситуации. Возможным решением может быть:

<T> Consumer<T> combine(Stream<? extends Consumer<? super T>> consumers) {
    return consumers.filter(Objects::nonNull)
                    .reduce(t -> {}, Consumer::andThen, Consumer::andThen);
}

Третий аргумент BinaryOperator<U> combiner вызывается только в параллельных потоках, но в любом случае было бы разумно обеспечить его правильную реализацию.

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

<T> Consumer<T> combine(Stream<? extends Consumer<? super T>> consumers) {

    Consumer<T> identity = t -> {};
    BiFunction<Consumer<T>, Consumer<? super T>, Consumer<T>> acc = Consumer::andThen;
    BinaryOperator<Consumer<T>> combiner = Consumer::andThen;

    return consumers.filter(Objects::nonNull)
                    .reduce(identity, acc, combiner);
}

Теперь вы можете написать:

Stream<? extends Consumer<? super Foo>> consumers = Stream.of();
combine(consumers);

Ответ 2

Вы забыли небольшую вещь в определении вашего метода. В настоящее время это:

<T> Consumer<T> combine(Stream<? extends Consumer<? super T>> consumers) {}

Но вы сохраняете Consumer<? super T> Consumer<? super T>. Поэтому, изменяя тип возврата, он почти работает. Теперь вы принимаете аргумент consumers типа Stream<? extends Consumer<? super T>> Stream<? extends Consumer<? super T>> Stream<? extends Consumer<? super T>>. В настоящее время это не работает, потому что вы работаете с различными подклассами и реализациями Consumer<? super T> Consumer<? super T> (из-за upperbounded подстановочные extends). Вы можете преодолеть это, бросив все ? extends Consumer<? super T> ? extends Consumer<? super T> ? extends Consumer<? super T> в вашем Stream к простому Consumer<? super T> Consumer<? super T>. Например:

<T> Consumer<? super T> combine(Stream<? extends Consumer<? super T>> consumers) {
    return consumers
        .filter(Objects::nonNull)
        .map(c -> (Consumer<? super T>) c)
        .reduce(Consumer::andThen)
        .orElse(noOpConsumer());
}

Теперь это должно работать

Ответ 3

Если у вас много потребителей, применение метода Consumer.andThen() создаст огромное дерево потребительских оболочек, которое будет рекурсивно обрабатываться для вызова каждого оригинального потребителя.

Таким образом, было бы более эффективно просто составить список потребителей и создать простого потребителя, который выполняет итерации по ним:

<T> Consumer<T> combine(Stream<? extends Consumer<? super T>> consumers) {
    List<Consumer<? super T>> consumerList = consumers
            .filter(Objects::nonNull)
            .collect(Collectors.toList());
    return t -> consumerList.forEach(c -> c.accept(t));
}

В качестве альтернативы, если вы можете гарантировать, что результирующий потребитель будет вызываться только один раз, и что Stream прежнему будет действителен в это время, вы можете просто перебрать его непосредственно по потоку:

return t -> consumers
        .filter(Objects::nonNull)
        .forEach(c -> c.accept(t));