如何在没有 isFinite() 和 isOrdered() 方法的情况下安全地使用 Java Streams？

Question

关于 java 方法是否应该返回 Collections or Streams 的问题是，Brian Goetz 在其中回答即使对于有限序列，通常也应该首选 Streams。

但在我看来，目前对来自其他地方的 Streams 的许多操作无法安全执行，并且无法进行防御性代码保护，因为 Streams 不会显示它们是无限的还是无序的。

如果并行是我想在 Stream() 上执行的操作的问题，我可以调用 isParallel() 来检查或顺序以确保计算是并行的（如果我记得的话）。

但如果有序性或有限性（大小）与我的程序的安全性有关，我就无法编写保护措施。

假设我使用了一个实现这个虚构接口的库：

public interface CoordinateServer {
    public Stream<Integer> coordinates();
    // example implementations:
    // finite, ordered, sequential
    // IntStream.range(0, 100).boxed()
    // final AtomicInteger atomic = new AtomicInteger();
    
    // // infinite, unordered, sequential
    // Stream.generate(() -> atomic2.incrementAndGet()) 

    // infinite, unordered, parallel
    // Stream.generate(() -> atomic2.incrementAndGet()).parallel()
    
    // finite, ordered, sequential, should-be-closed
    // Files.lines(Path.path("coordinates.txt")).map(Integer::parseInt)
}

那么我可以在这个流上安全地调用哪些操作来编写正确的算法？

似乎如果我想将元素写入文件作为副作用，我需要关注流是并行的：

// if stream is parallel, which order will be written to file?
coordinates().peek(i -> {writeToFile(i)}).count();
// how should I remember to always add sequential() in  such cases?

如果是并行的，基于什么线程池是并行的？

如果我想对流进行排序（或其他非短路操作），我需要小心它是无限的：

coordinates().sorted().limit(1000).collect(toList()); // will this terminate?
coordinates().allMatch(x -> x > 0); // will this terminate?

我可以在排序之前施加一个限制，但是如果我期望一个未知大小的有限流，那应该是哪个幻数？

最后也许我想并行计算以节省时间然后收集结果：

// will result list maintain the same order as sequential?
coordinates().map(i -> complexLookup(i)).parallel().collect(toList());

但是如果流没有被排序（在那个版本的库中），那么结果可能会由于并行处理而变得混乱。但是，除了不使用并行（这违背了性能目的）之外，我该如何防范呢？

集合明确表示有限或无限，是否有顺序，并且它们不携带处理模式或线程池。这些似乎是 API 的宝贵属性。

另外，Streams may sometimes need to be closed，但大多数情况下不是。如果我从一个方法（来自一个方法参数）消费一个流，我通常应该调用 close 吗？

另外，流可能已经被消费了，能够优雅地处理这种情况会很好，所以check if the stream has already been consumed会很好；

我希望有一些代码 sn-p 可用于在处理流之前验证有关流的假设，例如>

Stream<X> stream = fooLibrary.getStream();
Stream<X> safeStream = StreamPreconditions(
    stream, 
    /*maxThreshold or elements before IllegalArgumentException*/
    10_000,
    /* fail with IllegalArgumentException if not ordered */
    true
    )

Answer 1

就我所见，在看了一些东西（一些实验和here）之后，没有办法确定流是否是有限的。

不仅如此，有时甚至在运行时都无法确定（例如在 java 11 - IntStream.generate(() -> 1).takeWhile(x -> externalCondition(x)) 中）。

你可以做的是：

1. 您可以通过几种方式确定它是否是有限的（请注意，在这些方面收到错误并不意味着它是无限的，只是它可能是无限的）：

   1. stream.spliterator().getExactSizeIfKnown() - 如果它具有已知的确切大小，则它是有限的，否则它将返回 -1。

   2. stream.spliterator().hasCharacteristics(Spliterator.SIZED) - 如果是 SIZED 将返回 true。

2. 您可以通过假设最坏的情况来保护自己（取决于您的情况）。

   1. stream.sequential()/stream.parallel() - 明确设置您的首选消费类型。

   2. 对于潜在的无限流，假设每种情况下的最坏情况。

      1. 例如，假设您想听一连串推文，直到找到Venkat 的推文 - 这可能是一个无限操作，但您想等到找到这样的推文。因此，在这种情况下，只需选择 stream.filter(tweet -> isByVenkat(tweet)).findAny() - 它会迭代直到出现这样的推文（或永远）。
      2. 另一种情况（可能是更常见的情况）是想要对所有元素执行某项操作，或者只尝试一定的时间（类似于超时）。为此，我建议在调用您的操作之前始终调用 stream.limit(x)（collect 或 allMatch 或类似名称），其中 x 是您愿意容忍的尝试次数。

毕竟，我只想提一下，我认为返回流通常不是一个好主意，除非有很大的好处，否则我会尽量避免它。