如何在没有isFinite（）和isOrdered（）方法的情况下安全地使用Java流？

Question

有一个问题是java方法应该返回集合还是流，Brian Goetz回答说，即使对于有限序列，流通常也应该是首选的。

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

如果parallel是我要在Stream（）上执行的操作的一个问题，那么我可以调用isParallel（）来检查或sequential来确保计算是并行的（如果我记得这样做的话）。

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

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

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来说似乎是很有价值的属性。

此外，流有时可能需要关闭，但最常见的情况是不关闭。如果我从一个方法（从一个方法参数）消耗一个流，我一般应该调用close吗？

我希望有一些代码片段，可以用来在处理流之前验证关于它的假设，比如>

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

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

不仅如此，有时甚至在运行时也不确定（例如在java 11-IntStream.Generate(()->1).TakeWhile(x->externalCondition(x)))。

你能做的是：

>

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

>

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

stream.spliterator().HasCharacteries(Spliterator.Signed)-如果是Signed将返回true。

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

>

stream.sequential()/stream.parallel()-显式设置首选消费类型。

对于潜在的无限流，在每个场景中假设您的最坏情况。

1. 例如，假设您想要收听一个tweet流，直到找到Venkat的一个--这是一个潜在的无限操作，但您希望等待直到找到这样的tweet。因此在本例中，只需使用stream.filter(tweet->isByVenkat(tweet)).findany()-它将循环直到出现这样的tweet为止（或永远）。
2. 另一种情况（可能是更常见的情况）是希望对所有元素执行某些操作，或者只尝试一定的时间（类似于超时）。为此，我建议在调用操作(collection或allmatch或类似操作）之前始终调用stream.limit(x)，其中x是您愿意容忍的尝试量。

在完成了所有这些之后，我只想提到，我认为返回流通常不是一个好主意，并且我会尽量避免它，除非有很大的好处。