如何实现 Java 8 Stream fluent API 和延迟计算

Question

我想找出一个与 Java 8 Stream 等效的简单实现，它可以让我探索延迟计算的查询算法的开发（例如 map()、filter()、reduce() 等）。 注意：实现比 Stream 更好的解决方案不是我的目标。另一方面，我唯一的目标是了解 Stream 内部结构。

然而，我发现的每个实现都是基于Iterable<T>，例如以下答案中提出的解决方案：

• How are lazy streams implemented in Java 8?
• streaming is not a container of data, but a container of logic
• Java8 internal iteration

然而，我对这些解决方案中的任何一个都不满意，因为：

1. 它们太冗长了。
2. 它们对于新的查询方法不灵活。包含新的查询方法需要对结构进行修改。
3. 尽管有查询参数，但它们并没有利用 Java 8 的新特性，例如：第一类函数或默认方法。
4. 他们都没有使用在Stream<T> 上使用的Spliterator<T> 方法。

我知道Spliterator<T> 旨在允许分区和并行处理，但我认为它独特的迭代器方法 (boolean tryAdvance(Consumer<t>)) 可以被利用来替代上面列出的那些方法。而且作为stated by Brian Goetz:

Spliterator 是更好的Iterator，即使没有并行性。 （它们通常也更容易编写，更难出错。）

那么，是否可以基于Stream<T> 的相同原理（并行处理部分除外）开发一个更易读、更简单、更简洁、更灵活的查询 API 实现？

如果是，你怎么做？我希望看到比上面列出的那些更简单的实现，并尽可能利用新的 Java 8 功能。

要求：

• 不要重用 Java 8 API 中的现有方法
• 并行处理功能超出了本问题的范围。
• 如果可能，最好不要使用Iterable<T> 方法。

我的问题的原因是什么？我认为学习查询 API（如 Stream）的最佳方法是尝试自己实现这些相同的方法。我在学习.net Linq的时候已经成功了。当然，我没有实现比 Linq 更好的实现，但这有助于我理解内部部分。因此，我正在尝试遵循相同的方法来学习 Stream。

这并不罕见。有许多针对其他技术的研讨会遵循这种方法，例如functional-javascript-workshop，大多数练习要求实现现有方法，例如：map()、filter()、reduce()、call()、bind()等等……

选择答案：现在我认为Miguel Gamboa’s answer 是我的选择，而不是Tagir Valeev’s answer，因为后者不允许在不完全遍历整个元素的情况下实现findAny() 或findFirst()通过dataSrc 的forEach()。但是，我认为Tagir Valeev’s answer 在一些中间操作的简洁实现以及性能方面还有其他优点，因为基于forEach() 的方法减少了调解访问数据结构内部的迭代代码的开销，如Brian Goetz on point 2 of its answer

Answer 1

首先，我不得不说我喜欢 Lambda 和 Stream API 的设计。 JDK 中的实现也很棒，性能也很好。而且我不确定您学习实施/自己做的目的是否好。但我确实在我的开源库 AbacusUtil 中实现了 Stream API，既顺序又并行。以下是 github 上的源代码：Stream。我不能说它有多好，比较JDK中的实现。但就个人而言，我认为实现非常简单明了。而且它的性能也很高。

披露：我是AbacusUtil的开发者。

Answer 2

在没有短路支持的情况下实现无状态操作的子集非常容易。您只需要注意始终坚持内部迭代。基本构建块是forEach 操作，它可以为每个输入元素执行给定的操作。 forEach 方法的主体是唯一在不同阶段发生变化的东西。所以我们可以用抽象的forEach方法创建抽象类，或者接受一个实际上是forEach的主体的函数。我会坚持第二种方法：

public final class MyStream<T> {
    private final Consumer<Consumer<T>> action;

    public MyStream(Consumer<Consumer<T>> action) {
        this.action = action;
    }

    public void forEach(Consumer<T> cons) {
        action.accept(cons);
    }
}

现在让我们创建一些简单的来源：

public static <T> MyStream<T> of(Iterable<T> elements) {
    // just redirect to Iterable::forEach
    return new MyStream<>(elements::forEach);
}

@SafeVarargs
public static <T> MyStream<T> of(T... elements) {
    return of(Arrays.asList(elements));
}

public static MyStream<Integer> range(int from, int to) {
    return new MyStream<>(cons -> {
        for(int i=from; i<to; i++) cons.accept(i);
    });
}

现在是中间操作。他们只需要调整action 接收的消费者来执行其他操作：

public <U> MyStream<U> map(Function<T, U> mapper) {
    return new MyStream<>(cons -> forEach(e -> cons.accept(mapper.apply(e))));
}

public MyStream<T> filter(Predicate<T> pred) {
    return new MyStream<>(cons -> forEach(e -> {
        if(pred.test(e))
            cons.accept(e);
    }));
}

public <U> MyStream<U> flatMap(Function<T, MyStream<U>> mapper) {
    return new MyStream<>(cons -> forEach(e -> mapper.apply(e).forEach(cons)));
}

public MyStream<T> peek(Consumer<T> action) {
    return new MyStream<>(cons -> forEach(e -> {
        action.accept(e);
        cons.accept(e);
    }));
}

public MyStream<T> skip(long n) {
    return new MyStream<>(cons -> {
        long[] count = {0};
        forEach(e -> {
            if(++count[0] > n)
                cons.accept(e);
        });
    });
}

现在让我们使用forEach创建一些终端操作：

public T reduce(T identity, BinaryOperator<T> op) {
    class Box {
        T val = identity;
    }
    Box b = new Box();
    forEach(e -> b.val = op.apply(b.val, e));
    return b.val;
}

public Optional<T> reduce(BinaryOperator<T> op) {
    class Box {
        boolean isPresent;
        T val;
    }
    Box b = new Box();
    forEach(e -> {
        if(b.isPresent) b.val = op.apply(b.val, e);
        else {
            b.val = e;
            b.isPresent = true;
        }
    });
    return b.isPresent ? Optional.empty() : Optional.of(b.val);
}

public long count() {
    return map(e -> 1L).reduce(0L, Long::sum);
}

public Optional<T> maxBy(Comparator<T> cmp) {
    return reduce(BinaryOperator.maxBy(cmp));
}

public Optional<T> minBy(Comparator<T> cmp) {
    return reduce(BinaryOperator.minBy(cmp));
}

所以我们现在有了我们的流。让我们试试吧：

System.out.println(MyStream.of(1,2,3,4,5).map(x -> x*2)
                           .reduce(0, Integer::sum));
// 30

System.out.println(MyStream.of("a", "stream", "of", "some", "strings")
                           .flatMap(x -> MyStream.of(", ", x))
                           .skip(1).reduce("", String::concat));
// a, stream, of, some, strings

System.out.println(MyStream.range(0, 100)
                           .filter(x -> x % 3 == 0).count());
// 34

等等。这样的实现非常简单，但与实际 Stream API 中的情况非常接近。当然，当您添加短路、并行流、原始特化和更多有状态的操作时，事情会变得更加复杂。

请注意，与 Stream API 不同的是，MyStream 可以多次重复使用：

MyStream<Integer> range = range(0, 10);
range.forEach(System.out::println);
range.forEach(System.out::println); // works perfectly

Answer 3

使用函数式编程并利用 Java 8 默认方法，我们可以实现一个简短而干净的延迟计算查询 API 的解决方案。例如，查看如何在下面的 Queryable 类型中轻松实现 map() 和 forEach() 方法，然后您可以像这样使用它：

List<String> data = Arrays.asList("functional", "super", "formula");
Queryable.of(data) // <=> data.stream().
     .map(String::length)
     .forEach(System.out::println);

如果您将Queryable.of(dataSrc) 调用替换为dataSrc.stream()，您将获得相同的结果。以下示例说明了map() 和forEach() 方法的实现。在Queryable repository查看完整的解决方案和更详细的描述。

更新与@srborlongan comment。将forEach签名从forEach(Consumer<T>)更改为forEach(Consumer<? super T>)并将of从of(Collection<T>)更改为of(Iterable<T>)

@FunctionalInterface
public interface Queryable<T>{

  abstract boolean tryAdvance(Consumer<? super T> action); // <=> Spliterator::tryAdvance

  static <T> boolean truth(Consumer<T> c, T item){
    c.accept(item);
    return true;
  }

  public static <T> Queryable<T> of(Iterable<T> data) {
    final Iterator<T> dataSrc = data.iterator();
    return action -> dataSrc.hasNext() && truth(action, dataSrc.next());
  }

  public default void forEach(Consumer<? super T> action) {
    while (tryAdvance(action)) { }
  }

  public default <R> Queryable<R> map(Function<T, R> mapper) {
    return action -> tryAdvance(item -> action.accept(mapper.apply(item)));
  }
}