有条件地向 Java 8 流添加操作

Question

我想知道是否可以根据流外部设置的某种条件向流中添加操作。例如，如果我的limit 变量不等于-1，我想在流中添加限制操作。

我的代码目前看起来像这样，但我还没有看到其他以这种方式使用流的示例，其中将 Stream 对象重新分配给对其自身应用的中间操作的结果：

// Do some stream stuff
stream = stream.filter(e -> e.getTimestamp() < max);

// Limit the stream
if (limit != -1) {
   stream = stream.limit(limit);
}

// Collect stream to list
stream.collect(Collectors.toList());

正如stackoverflow post 中所述，在调用终端操作之前不会实际应用过滤器。由于我在调用终端操作之前重新分配了流的值，所以上面的代码仍然是使用 Java 8 流的正确方法吗？

Answer 1

链接的一系列调用与存储中间返回值的一系列调用之间没有语义差异。因此，以下代码片段是等价的：

a = object.foo();
b = a.bar();
c = b.baz();

和

c = object.foo().bar().baz();

在任何一种情况下，每个方法都会根据前一次调用的结果来调用。但在后一种情况下，中间结果不会被存储，而是在下一次调用时丢失。在流 API 的情况下，不得在调用 next 方法后使用中间结果，因此链接是使用流的自然方式，因为它本质上确保您不会在返回的引用上调用多个方法。

不过，只要遵守不多次使用返回引用的约定，将引用存储到流中并没有错。通过使用它们在您的问题中的方式，即用下一次调用的结果覆盖变量，您还可以确保您不会在返回的引用上调用多个方法，因此，这是一个正确的用法。当然，这只适用于相同类型的中间结果，所以当你使用map或flatMap，获取不同引用类型的流时，你不能覆盖局部变量。那么你要注意不要再使用旧的局部变量了，但是，正如上面所说，只要你下次调用后不使用它，中间存储就没有问题。

有时，您必须存储它，例如

try(Stream<String> stream = Files.lines(Paths.get("myFile.txt"))) {
    stream.filter(s -> !s.isEmpty()).forEach(System.out::println);
}

请注意，该代码等效于以下替代方案：

try(Stream<String> stream = Files.lines(Paths.get("myFile.txt")).filter(s->!s.isEmpty())) {
    stream.forEach(System.out::println);
}

和

try(Stream<String> srcStream = Files.lines(Paths.get("myFile.txt"))) {
    Stream<String> tmp = srcStream.filter(s -> !s.isEmpty());
    // must not be use variable srcStream here:
    tmp.forEach(System.out::println);
}

它们是等价的，因为总是在 filter 的结果上调用 forEach，而在 Files.lines 的结果上总是调用它，而最终的 close() 操作在哪个结果上调用都没有关系，因为关闭影响整个流管道。

---

一句话，你用它的方式，就是对的。

---

我什至更喜欢这样做，因为当您不想应用限制时不链接limit 操作是表达您意图的最简洁方式。还值得注意的是，建议的替代方案可能在很多情况下都有效，但它们在语义上不等效：

.limit(condition? aLimit: Long.MAX_VALUE)

假设您可以遇到的最大元素数是Long.MAX_VALUE，但流可以有更多的元素，甚至可能是无限的。

.limit(condition? aLimit: list.size())

当流源为list 时，打破了流的惰性求值。原则上，可变流源可能会合法地任意更改，直到开始终端操作。结果将反映到目前为止所做的所有修改。当您添加包含list.size() 的中间操作时，即此时列表的实际大小，随后应用于此点和终端操作之间的集合的修改可能会使该值具有与预期的“实际上没有”不同的含义限制”语义。

与“Non Interference” section of the API documentation比较：

对于表现良好的流源，可以在终端操作开始之前修改源，这些修改将反映在覆盖的元素中。例如，考虑以下代码：

List<String> l = new ArrayList(Arrays.asList("one", "two"));
Stream<String> sl = l.stream();
l.add("three");
String s = sl.collect(joining(" "));

首先创建一个包含两个字符串的列表：“one”；和“二”。然后从该列表创建一个流。接下来通过添加第三个字符串来修改列表：“three”。最后，流的元素被收集并连接在一起。由于列表在终端收集操作开始之前被修改，结果将是一个字符串“一二三”。

当然，这是一个罕见的极端情况，通常情况下，程序员将制定整个流管道而不修改其间的源集合。尽管如此，不同的语义仍然存在，一旦进入这种极端情况，它可能会变成一个很难找到的错误。

此外，由于它们不等价，流 API 永远不会将这些值识别为“实际上没有限制”。即使指定Long.MAX_VALUE 也意味着流实现必须跟踪已处理元素的数量以确保遵守限制。因此，不添加limit 操作比添加一个程序员期望永远不会超过的限制具有显着的性能优势。

Answer 2

有两种方法可以做到这一点

// Do some stream stuff
List<E> results = list.stream()
                  .filter(e -> e.getTimestamp() < max);
                  .limit(limit > 0 ? limit : list.size())
                  .collect(Collectors.toList());

或

// Do some stream stuff
stream = stream.filter(e -> e.getTimestamp() < max);

// Limit the stream
if (limit != -1) {
   stream = stream.limit(limit);
}

// Collect stream to list
List<E> results = stream.collect(Collectors.toList());

由于这是函数式编程，您应该始终处理每个函数的结果。您应该特别避免以这种编程风格修改任何内容，并尽可能将所有内容视为不可变的。

由于我在调用终端操作之前重新分配了流的值，所以上面的代码仍然是使用 Java 8 流的正确方法吗？

它应该可以工作，但是它读起来是命令式和函数式编码的混合体。我建议按照我的第一个答案将其写为固定流。

Answer 3

我认为你的第一行应该是：

stream = stream.filter(e -> e.getTimestamp() < max);

以便您在后续操作中使用过滤器返回的流而不是原始流。

Answer 4

我知道这有点太晚了，但我自己也有同样的问题，并没有找到令人满意的答案，但是，受这个问题和答案的启发，我得出了以下解决方案：

return Stream.of( ///< wrap target stream in other stream ;)
    /*do regular stream stuff*/ 
    stream.filter(e -> e.getTimestamp() < max)
  ).flatMap(s -> limit != -1 ? s.limit(limit) : s) ///< apply limit only if necessary and unwrap stream of stream to "normal" stream
  .collect(Collectors.toList()) ///< do final stuff