带记忆的 Java8 流

Question

如何在 java8（可能是记忆过程）中重用已经通过流迭代计算的值？

如果流被复制或再次提供，它将被重新计算。在某些情况下，最好用内存换取 CPU 时间。从头开始收集所有内容可能不是一个好主意，因为流用于查找满足谓词的第一个项目。

Stream<Integer> all = Stream.of(1,2,3,4,5, ...<many other values>... ).
      map(x->veryLongTimeToComputeFunction(x));
System.out.println("fast find of 2"+all.filter(x->x>1).findFirst());

//both of these two lines generate a "java.lang.IllegalStateException: stream has already been operated upon or closed"
System.out.println("no find"+all.filter(x->x>10).findFirst());
System.out.println("find again"+all.filter(x->x>4).findFirst());

---

问题类似于Copy a stream to avoid "stream has already been operated upon or closed" (java 8)

Answer 1

规范的内存中流源是一个集合。一个简单的、不能并行的流记忆可以实现如下：

public static void main(String[] args) {
    Supplier<Stream<Integer>> s=memoize(
        IntStream.range(0, 10_000)
                 .map(x -> veryLongTimeToComputeFunction(x))
    );
    System.out.println("First item > 1  "+s.get().filter(x -> x>1 ).findFirst());
    System.out.println("First item > 10 "+s.get().filter(x -> x>10).findFirst());
    System.out.println("First item > 4  "+s.get().filter(x -> x>4 ).findFirst());
}
static int veryLongTimeToComputeFunction(int arg) {
    System.out.println("veryLongTimeToComputeFunction("+arg+")");
    return arg;
}

public static <T> Supplier<Stream<T>> memoize(BaseStream<T,?> stream) {
    Spliterator<T> sp=stream.spliterator();
    class S extends Spliterators.AbstractSpliterator<T> {
        ArrayList<T> mem=new ArrayList<>();
        S() { super(sp.estimateSize(), sp.characteristics()); }
        public boolean tryAdvance(Consumer<? super T> action) {
            return sp.tryAdvance(item -> {
                mem.add(item);
                action.accept(item);
            });
        }
    }
    S s=new S();
    return () -> Stream.concat(s.mem.stream(), StreamSupport.stream(s, false));
}

在向供应商请求下一个 Stream 之前，请注意完成 Stream 处理。

Answer 2

Java 8 流本质上是惰性的。对流执行的操作按垂直顺序进行评估。 您想要达到的目标可以使用以下代码实现：

Stream.of(1,2,3,4,5, ...<many other values>... )
    .map(x-> veryLongTimeToComputeFunction(x))
    .filter(x-> x > 1)
    .findFirst();

这将确保仅在找不到匹配的第一个元素之前调用veryLongTimeToComputeFunction()。之后，操作将终止。 在最坏的情况下，如果最后一个数字与条件匹配，则会为所有数字调用veryLongTimeToComputeFunction。

您还可以将并行流与 findAny() 方法结合使用。它将加快性能。

Answer 3

流并不意味着保存，它们即将处理数据。

示例：您正在观看 dvd，用 Java 术语来说，dvd 就像一个集合，从您的 dvd 播放器传输到电视的数据是一个流。你不能保存流，但是你可以刻录 cd，用 java 术语收集它。

还有其他选择：

• 将流操作或谓词提取/重构为获取流作为参数并返回流的方法
• 使用缓存框架：例如Spring 中的方法可以使用 @Cacheable 进行注释。第一次调用执行方法，后续调用在定义的时间内从缓存中获取结果
• 如果您正在为长时间运行的任务寻找非阻塞执行，请查看RxJava

Answer 4

为什么不在veryLongTimeToComputeFunction 中使用memoization？您可以将备忘录缓存作为参数传递给函数。

Answer 5

我建议将您的 Stream 收集到一个列表中，然后在列表的流上运行您的过滤器。