Java Streams - 过滤先前过滤的值

Question

我正在试验 Java 的 Streams，并试图找出可能的方法以及它们的优缺点。目前我正在尝试使用流来实现 Eratosthenes 筛，但似乎找不到一种好方法来循环遍历先前过滤的值而不将它们存储在单独的集合中。

我想要完成这样的事情：

IntStream myStream = IntStream.range(0,3);
myStream.filter(s -> {
    System.out.print("[filtering "+s+"] ");
    myStream.forEach(q -> System.out.print(q+", "));
    System.out.println();
    return true; //eventually respond to values observed on the line above
});

期望输出：

[filtering 0] 
[filtering 1] 0, 
[filtering 2] 0, 1, 
[filtering 3] 0, 1, 2, 

请注意，在过滤每个新值时，会观察所有先前过滤的值。这样可以轻松实现埃拉托色尼筛法，因为我可以过滤掉所有非质数，并为每个新值检查与之前通过质数过滤器的所有数字的可分性。

但是，上面的示例在 NetBeans 中给我一个错误：

local variables referenced from a lambda expression must be final or effectively final

这似乎是因为我在已经作用于 myStream 的过滤器中引用了 myStream。是否有解决此错误的好方法（即制作仅包含到目前为止已过滤的值的流的最终副本），或者是否有更好的方法来解决此类问题而不使用单独的集合来存储价值观？

Answer 1

我设法使用埃拉托色尼筛法创建了一个无限的素数Stream，但它实际上并没有使用过去的值。相反，它删除了尾部中素数的倍数（以一种懒惰的方式，因为尾部是无限的），就像最初的埃拉托色尼筛算法一样。为此，我使用了Iterator 作为辅助（因为Stream 只能使用一次）并为流实现了lazyConcat。

class StreamUtils {
    public static IntStream fromIterator(PrimitiveIterator.OfInt it) {
        return StreamSupport.intStream(
                Spliterators.spliteratorUnknownSize(it, Spliterator.ORDERED), false);
    }

    public static IntStream lazyConcat(Supplier<IntStream> a, Supplier<IntStream> b) {
        return StreamSupport.intStream(new Spliterator.OfInt() {
            boolean beforeSplit = true;
            Spliterator.OfInt spliterator;

            @Override
            public OfInt trySplit() {
                return null;
            }

            @Override
            public long estimateSize() {
                return Long.MAX_VALUE;
            }

            @Override
            public int characteristics() {
                return Spliterator.ORDERED;
            }

            @Override
            public boolean tryAdvance(IntConsumer action) {
                boolean hasNext;
                if (spliterator == null) {
                    spliterator = a.get().spliterator();
                }
                hasNext = spliterator.tryAdvance(action);
                if (!hasNext && beforeSplit) {
                    beforeSplit = false;
                    spliterator = b.get().spliterator();
                    hasNext = spliterator.tryAdvance(action);
                }
                return hasNext;
            }
        }, false);
    }
}

我的埃拉托色尼筛流看起来像这样：

class Primes {
    public static IntStream stream() {
        return sieve(IntStream.iterate(2, n -> n + 1));
    }

    private static IntStream sieve(IntStream s) {
        PrimitiveIterator.OfInt it = s.iterator();
        int head = it.nextInt();
        IntStream tail = StreamUtils.fromIterator(it);
        return StreamUtils.lazyConcat(
                () -> IntStream.of(head),
                () -> sieve(tail.filter(n -> n % head != 0)));
    }
}

那么我们可以这样使用：

System.out.println(Primes.stream().limit(20).boxed().collect(Collectors.toList()));

输出：

[2、3、5、7、11、13、17、19、23、29、31、37、41、43、47、53、59、61、67、71]

我认为这是一个很好的练习，但它似乎效率很低，而且根本不适合堆栈。

Answer 2

您不能多次处理 Stream，因此无法在 filter 方法中调用 myStream.forEach。

您可以在过滤器中创建一个新的IntStream。

请注意，您必须向外部 Stream 管道添加一些终端操作才能对其进行处理：

IntStream myStream = IntStream.range(0,4);
myStream.filter(s -> {
    System.out.print("[filtering "+s+"] ");
    IntStream.range(0,s).forEach(q -> System.out.print(q+", "));
    System.out.println();
    return true; //eventually respond to values observed on the line above
}).forEach(i->{});

这会产生：

[filtering 0] 
[filtering 1] 0, 
[filtering 2] 0, 1, 
[filtering 3] 0, 1, 2, 

Answer 3

流是否是正确的工具是有争议的，但.filter() 绝对不是。过滤器应该是无状态的，所以这个想法不应该放在首位。根据您回答中的示例，收集器可能是一个可行的解决方案。

List<Integer> primes = IntStream.range(2, UPPER_BOUND)
  .collect(ArrayList::new,
          (list, number) -> { 
                for(int j=0; j < list.size(); j++) {
                    int prime = list.get(j);

                    if(prime > Math.sqrt(number)) {
                        break;
                    }

                    if(number % prime == 0) {
                        return;
                    }
                }

                list.add(number);
          },
          List::addAll);

ArrayList::new 创建一个新列表，然后消费者将其引用为list。消费流中的每个元素都会被调用，number 是元素。

List::addAll 只与无论如何不能用于此算法的并行流相关。

Answer 4

其他答案表明我一直尝试的方法是不可能的，必须使用单独的集合。

为了提供更完整的答案，我想提供一种使用流来解决此问题的有效方法，并将其与更传统的方法进行比较。

使用流列出素数（使用埃拉托色尼筛）：

List<Integer> primes = new ArrayList<Integer>();

IntStream.iterate(2, i -> i + 1)
    .limit(UPPER_BOUND)
    .filter(i -> {
        for(int j=0; j<primes.size(); j++) {
            int prime = primes.get(j);

            if(prime > Math.sqrt(i)) {
                break;
            }

            if(i % prime == 0) {
                return false;
            }
        }
        return true;
    })
    .forEach(primes::add);

传统的、等效的、不使用流的方法：

List<Integer> primes = new ArrayList<Integer>();

for(int i=2; i < UPPER_BOUND; i++) {
    boolean isPrime = true;

    for(int j=0; j<primes.size(); j++) {
        int prime = primes.get(j);

        if(prime > Math.sqrt(i)) {
            break;
        }

        if(i % prime == 0) {
            isPrime = false;
            break;
        }
    }

    if(isPrime) {
        primes.add(i);
    }
}

性能比较：

对每个函数的一些实验一致表明，在这种情况下，传统方法实际上比使用流更快。与传统方法相比（在我的机器上平均分别为 106 毫秒和 70 毫秒），流方法找到所有低于 100 万的素数所需的时间一直是 1.5 倍。

如果流的 .parallel() 函数可以轻松实现问题的并行化，这种性能差异可能很容易弥补。但是，在这种情况下，并行化并不容易，因为 ArrayList 不是线程安全的，并且会很快导致错误和/或不准确的结果。

结论：

假设其他答案是正确的，那么在 Java 中不可能在同一流的过滤器中过滤已过滤的数据。

可以使用流处理列出素数。但是，在等待比我自己更好的解决方案之前，目前最好还是坚持使用传统的无流方法。