Java 8泛型：将消费者流减少为单个消费者

Question

如何使用消费者编写一种方法，将消费者的流组合成单个消费者。然后（消费者）？

我的第一个版本是：

<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

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）：

类型消费者没有定义，然后（捕获#7-of？扩展消费者）

JavaC（1.8.0172）：

错误：不兼容类型：无效的方法引用
。减少（消费者::andSo）
不兼容类型：消费者

但它应该是有效的：Consumer的每个子类也可以用作Consumer。超级X的每个消费者也可以消费X。我试图向流版本的每一行添加类型参数，但这没有帮助。但如果我用传统的循环写下来，它会编译：

<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相信我的代码是正确的？或者，如果它不正确：为什么循环版本正确，而流版本不正确？




Answer 1

如果您有很多消费者，应用Consumer.and然后（）将创建一个巨大的消费者包装树，该树被递归处理以调用每个原始消费者。

因此，简单地建立一个消费者列表，并创建一个简单的消费者来对其进行迭代，可能会更有效：

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

或者，如果您可以保证生成的使用者只被调用一次，并且流在那时仍然有效，那么您可以直接在流上迭代：

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

Answer 2

您在方法定义中忘记了一件小事。它目前是：

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

但你是在报复消费者

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

现在应该可以了

Answer 3

使用一个参数流。减少（累加器）具有以下签名的版本：

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

二进制运算符

<? extends Consumer<? super T>>

我建议您使用一个三参数版本的Stream.reduce（...）方法来代替：

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

双功能

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

第三个参数二进制运算符

此外，为了更好地理解，可以将上述代码表示为：

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