Java 8 泛型 : Reducing a Stream of Consumers to a single Consumer

Question

我怎样才能写一个方法来组合 Stream的 Consumers成单个 Consumer使用 Consumer.andThen(Consumer) ?

我的第一个版本是:

<T> Consumer<T> combine(Stream<Consumer<T>> consumers) {
    return consumers
            .filter(Objects::nonNull)
            .reduce(Consumer::andThen)
            .orElse(noOpConsumer());
}

<T> Consumer<T> noOpConsumer() {
    return value -> { /* do nothing */ };
}

此版本使用 JavaC 和 Eclipse 编译。但它太具体了:Stream不能是 Stream<SpecialConsumer> , 如果 Consumers不完全属于 T 类型但是它的父类(super class)型，不能使用:

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

这不会编译，这是正确的。改进后的版本是:

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

但是 Eclipse 和 JavaC 都不能编译它:
eclipse (4.7.3a):

The type Consumer does not define andThen(capture#7-of ? extends Consumer<? super T>, capture#7-of ? extends Consumer<? super T>) that is applicable here

JavaC (1.8.0172):

error: incompatible types: invalid method reference
.reduce(Consumer::andThen)
incompatible types: Consumer<CAP#1> cannot be converted to Consumer<? super CAP#2>
where T is a type-variable:
T extends Object declared in method <T>combine(Stream<? extends Consumer<? super T>>)
where CAP#1,CAP#2 are fresh type-variables:
CAP#1 extends Object super: T from capture of ? super T
CAP#2 extends Object super: T from capture of ? super T

但它应该可以工作:Consumer 的每个子类也可以用作 Consumer。并且每个父类(super class)型 X 的消费者也可以消费 Xs。我尝试将类型参数添加到流版本的每一行，但这无济于事。但是如果我用传统的循环写下来，它会编译:

<T> Consumer<T> combine(Collection<? extends Consumer<? super T>> consumers) {
    Consumer<T> result = noOpConsumer()
    for (Consumer<? super T> consumer : consumers) {
        result = result.andThen(consumer);
    }
    return result;
}

(为简洁起见，省略了过滤掉空值。)

因此，我的问题是:如何让 JavaC 和 Eclipse 相信我的代码是正确的？或者，如果不正确:为什么循环版本正确但 Stream 不正确版本？

Answer 1

您使用单参数 Stream.reduce(accumulator) 具有以下签名的版本:

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

BinaryOperator<T> accumulator只能接受 T 类型的元素，但你有:

<? extends Consumer<? super T>>

我建议您使用 Stream.reduce(...) 的三参数版本替代方法:

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

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);