如何将两个 java8 流操作 - 一个终端和一个惰性 - 组合成一个操作？

Question

我正在做一些 "algebra" Java 8 的流，也就是说，我正在尝试编写一个简单的操作 Op，它将两个流作为输入并产生另一个流作为结果.

所以我有这个简单的代码，其目的是打印一系列数字中的第二个最高值：

import java.util.Arrays;
import java.util.stream.IntStream;

public class SecundHighestValue {

    public static void main(String[] args) {

        //setting the input parameters
        int [] numbers = {1, 2, 3, 4, 3, 4, 2, 1};

        IntStream S1 = Arrays.stream(numbers);
        IntStream S2 = Arrays.stream(new int[] {Arrays.stream(numbers).max().getAsInt()} );

        // setting the operation
        IntStream S3 = S1.filter(x-> x != S2.toArray()[0]); // doesn't work

        /*** does work  ***
        int  maxNumber = S2.toArray()[0];
        IntStream S3 = S1.filter(x-> x != maxNumber);
        */

        // accessing the operation's result stream S3
        int secundMaxNumber = S3.max().getAsInt();
        System.out.println("the secund highest value in the serie " +
                    Arrays.toString(numbers) + " is " + secundMaxNumber);   
    }
}

这个程序不行，除非我这样拆分单行操作:

    int  maxNumber = S2.toArray()[0];
    IntStream S3 = S1.filter(x-> x != maxNumber);

将操作保持在一行中将引发此异常：

线程异常 "main" java.lang.IllegalStateException: 流已被操作或关闭 ...

我知道这与 filter() 方法固有的惰性有关。 API 解释：

Stream operations are divided into intermediate (Stream-producing) operations and terminal (value- or side-effect-producing) operations. Intermediate operations are always lazy.

事实上，堆栈跟踪显示直到我尝试在下一行中访问它的结果，该操作才会执行。

这种行为是 java8 中的设计缺陷吗？这是一个错误吗？最重要的是，我怎样才能将操作保持在一行中并使其正常工作？

Answer 1

你有四行：

 IntStream S1 = Arrays.stream(numbers);
 IntStream S2 = Arrays.stream(new int[] {Arrays.stream(numbers).max().getAsInt()} );
 int  maxNumber = S2.toArray()[0];
 IntStream S3 = S1.filter(x-> x != maxNumber);
 int secundMaxNumber = S3.max().getAsInt();

2 相同：

int  maxNumber = Arrays.stream(numbers).max().getAsInt();
int secundMaxNumber = Arrays.stream(numbers).filter(x-> x != maxNumber).max().getAsInt();

很难重复使用流，所以最好以一种方式进行，最好计算一个变量中的最大值并重复使用而不是每次都计算它

Answer 2

这不起作用的原因：

IntStream S3 = S1.filter(x-> x != S2.toArray()[0]);

是因为S2只能作用一次。过滤器为 S3 中的每个条目重新计算它。

把它 filter 想象成一个 for 循环，把 s2 想象成一个只能准备一次的值。您可以将流与 System.in 进行比较 - 一旦您读取了该值，就无法重新读取它。你必须得到一个新的。

更多信息：该操作并不懒惰，因为您有这行代码使其成为终端：

secundMaxNumber = S3.max().getAsInt();

旁注：要获得第 X 个 maxNumber，您也可以这样做：您不需要多次使用流。

S1.sorted().limit(x).skip(x-1).findFirst().getAsInt();

参考文献：

https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html#limit-long-

https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html#skip-long-

Answer 3

如果可以通过源进行流式传输并且不昂贵，例如数组，您可以只流式传输两次，如 :

int maxNumber = Arrays.stream(numbers).max().getAsInt();
int secondMaxNumber = Arrays.stream(numbers).filter(x-> x != maxNumber).max().getAsInt();

如果流式传输两次不可能或成本很高，您需要一个自定义收集器来有效地获取第二大值，即只保留必要的两个值。例如

final class SecondMax {
    long max=Long.MIN_VALUE, semi=max;

    void add(int next) {
        if(next>semi) {
            if(next>max) {
                semi=max;
                max=next;
            }
            else if(next<max) {
                semi=next;
            }
        }
    }
    void merge(SecondMax other) {
        if(other.max>Long.MIN_VALUE) {
            add((int)other.max);
            if(other.semi>Long.MIN_VALUE) add((int)other.semi);
        }
    }
    OptionalInt get() {
        return semi>Long.MIN_VALUE? OptionalInt.of((int)semi): OptionalInt.empty();
    }
}

有了这个助手，您可以在单个流操作中获取值：

OptionalInt secondMax = Arrays.stream(array)
  .collect(SecondMax::new, SecondMax::add, SecondMax::merge).get();

Answer 4

由于输入是一个int数组，所以@azro提供的解决方案对我来说已经足够了。第二个@Holger：不必定义新的 class:

final Supplier<int[]> supplier = () -> new int[] { Integer.MIN_VALUE, Integer.MIN_VALUE };
final ObjIntConsumer<int[]> accumulator = (a, i) -> {
    if (i > a[0]) {
        a[1] = a[0];
        a[0] = i;
    } else if (i != a[0] && i > a[1]) {
        a[1] = i;
    }
};

int secondMax = Arrays.stream(nums).collect(supplier, accumulator, (a, b) -> {})[1];

或者用第三方库中提供的API：AbacusUtil

int secondMax = IntStream.of(nums).distinct().kthLargest(2).get();

如何将两个 java8 流操作 - 一个终端和一个惰性 - 组合成一个操作？

How to combine two java8 stream operations - one terminal and one lazy - into a single operation?

java

functional-programming

lazy-evaluation

java-8

java-stream