java - 将 `IntStream.generate` 与有状态 `IntSupplier` 一起使用时有哪些陷阱？

Question

假设我想使用带有斐波那契数列的 Java 8 Stream。 Java 提供的最接近完成此操作的方法是 IntStream.iterate 但这要求我仅根据前一个数字计算下一个 Fibonacci，但我需要两个前一个数字。为了记住之前生成的两个数字，我可以使用 IntStream.generate 和有状态的 IntSupplier 来解决这个问题，如下所示:

class FibonacciDemo {
    public static void main(String[] args) {
        IntStream.generate(new FibonacciGenerator()).limit(10).forEach(System.out::println);
    }
}
class FibonacciGenerator implements IntSupplier {
    private int previous = 0;
    private int previousPrevious = 1;

    @Override
    public int getAsInt() {
        final int next = previous + previousPrevious;
        previousPrevious = previous;
        previous = next;
        return next;
    }
}

这行得通，但令我惊讶的是，我在手册中找不到任何信息，无论这是一件坏事，不应该这样做，还是会出现什么问题。所以我阅读了一些源代码，并认为当生成器将在并行流中使用时，生成器将在多个 Spliterator 之间共享。 所以我想我应该使方法 getAsInt synchronized 以使其对于并行流是安全的。正确的？还有什么陷阱吗？

上面的代码只是一个示例，说明有状态的 IntSupplier 首先可能有用，但它仍然足够简单，每个人都可以看到发生了什么。问题通常是关于有状态的 IntSupplier 与 IntStream.generate 的结合，而不是直接关于 Fibonacci。 (尽管如果有更好的方法在没有状态 IntSupplier 的情况下完成上述工作，那当然会令人感兴趣。)

Answer 1

一般来说，虽然没有明确指定，generate() 方法是为无状态供应商设计的。有状态的供应商将仅与顺序流一起正常工作。在并行情况下，您不能仅通过添加 synchronized 来修复供应商。让我们简化您的问题并测试序列号的生成器。首先，没有同步:

public class IncrementDemo {
    public static void main(String[] args) {
        IntStream.generate(new IncrementGenerator()).parallel()
                 .limit(200).forEachOrdered(System.out::println);
    }
}
class IncrementGenerator implements IntSupplier {
    private int previous = 0;

    @Override
    public int getAsInt() {
        return previous++;
    }
}

此代码通常打印如下内容:

1
4
7
9
11
31
35
...
373
376
379
0
3
5
7
9
10
...
214
217
220
222

所以我们只有一些垃圾。有些数字甚至在重复。让我们将 getAsInt 方法同步:

0
3
6
9
12
15
...
221
223
226
1
4
7
10
13
...
356
359
362

现在数字从不重复，但结果仍然毫无用处。顺序不是固定的(即使我们使用了forEachOrdered)。它甚至不会生成前 200 个非负数，它只会生成一些 非重复的非负数。您的斐波那契生成器也是如此。添加 synchronized 将保证您有一些非重复的斐波纳契数(但请注意 int 将很快溢出斐波纳契)，但不能保证您会看到哪些数字。

要生成连续的斐波那契数，您可以使用 Stream.iterate 将状态封装在中间容器中，然后将此容器映射到结果数字，如下所示:

Stream.iterate(new int[] {0, 1}, pair -> new int[] {pair[1], pair[0]+pair[1]})
    .mapToInt(pair -> pair[1])
    .parallel()
    .limit(200)
    .forEachOrdered(System.out::println);

即使对于并行流，此类代码也会生成连续的斐波那契数。虽然并行在这里不太可能提高速度，但至少它可以正常工作(如果你在管道中添加一些昂贵的操作，它可能会提高速度)。