在哪些情况下流操作应该是有状态的？

Question

在javaodoc for the stream package，Parallelism 部分的末尾，我读到：

大多数流操作都接受描述用户指定行为的参数，这些参数通常是 lambda 表达式。为了保持正确的行为，这些行为参数必须是无干扰的，并且在大多数情况下必须是无状态的。

我很难理解这个“在大多数情况下”。在哪些情况下可以接受/需要有状态的流操作？

我的意思是，我知道这是可能的，特别是在使用顺序流时，但同一个 javadoc 明确指出：

除了被明确识别为非确定性的操作，例如findAny()，流是顺序执行还是并行执行不应改变计算结果。

还有：

另请注意，尝试从行为参数访问可变状态会使您在安全性和性能方面做出错误的选择； [...] 最好的方法是避免有状态的行为参数完全流式操作；通常有一种方法可以重组流管道以避免有状态。

所以，我的问题是：在什么情况下使用有状态的流操作是一个好习惯（而不是通过副作用工作的方法，例如forEach）？

一个相关的问题可能是：为什么会有副作用工作，例如forEach？我总是会做一个很好的旧 for 循环以避免在我的 lambda 表达式中产生副作用。

Answer 1

“在大多数情况下”我很难理解这一点。在哪些情况下可以接受/需要有状态的流操作？

假设以下场景。您有一个Stream<String>，您需要按自然顺序列出项目，并在每个项目前加上订单号。因此，例如在输入时您有：Banana、Apple 和 Grape。输出应该是：

1. Apple
2. Banana
3. Grape

你如何在 Java Stream API 中解决这个任务？很容易：

List<String> f = asList("Banana", "Apple", "Grape");

AtomicInteger number = new AtomicInteger(0);
String result = f.stream()
  .sorted()
  .sequential()
  .map(i -> String.format("%d. %s", number.incrementAndGet(), i))
  .collect(Collectors.joining("\n"));

现在，如果您查看此管道，您将看到 3 个有状态操作：

• sorted()——根据定义是有状态的。请参阅Stream.sorted() 的文档：

  这是一个有状态的中间操作

• map() – 本身可能是无状态的，但在这种情况下它不是。要标记位置，您需要跟踪已标记的项目数量；
• collect() – 是可变归约 操作（从文档到Stream.collect()）。根据定义，可变操作是有状态的，因为它们会更改（改变）共享状态。

关于为什么sorted() 是有状态的存在一些争议。来自 Stream API 文档：

无状态操作（例如过滤器和映射）在处理新元素时不会保留先前看到的元素的状态——每个元素都可以独立于对其他元素的操作进行处理。在处理新元素时，有状态的操作（例如 distinct 和 sorted）可能会合并来自先前看到的元素的状态。

因此，当将术语 stateful/stateless 应用于 Stream API 时，我们更多地谈论的是流的函数处理元素，而不是作为一个整体的函数处理流.

另请注意，术语 stateless 和 deterministic 之间存在一些混淆。它们不一样。

确定性函数在给定相同参数的情况下提供相同的结果。

无状态函数不保留之前调用的状态。

这些是不同的定义。并且在一般情况下不依赖于彼此。确定性是关于函数结果值的，而无状态是关于函数实现的。

Answer 2

无状态函数对相同的输入返回相同的输出，“无论如何”。

用 Java 等命令式语言创建非无状态函数很容易。例如

    func = input -> currentTime();

如果我们使用有状态的func 执行stream.map(func)，则生成的流将取决于在运行时如何调用func；应用程序的行为将难以理解（但不是那么难）。

如果func 是无状态的，stream.map(func) 将始终产生相同的流，无论map 如何实现和执行。这是很好的和可取的。

请注意，“无论如何”意味着无状态函数必须是线程安全的。

---

如果一个函数返回void，它不总是无状态的吗？嗯……stateless 还有另一个含义——调用无状态函数不应该有对应用程序“重要”的副作用。

如果func 没有“重要”的副作用，那么随意调用func 是安全的。例如，stream.map(func) 可以安全地多次调用func，即使是在同一个元素上。 （但别担心，Stream 永远不会这样做）。

什么是“重要”的副作用？这是非常主观的。

至少，调用fun 会花费一些CPU 时间，这并不是完全免费的。这可能与性能关键的应用程序有关；或在昂贵的平台上（咳嗽 AWS）。

如果func 在硬盘上记录某些内容，它可能是也可能不是“重要”的副作用。 （它也需要 $$）

如果func 查询一个成本高昂的外部服务，这是非常令人担忧的，它可能会让你破产。

现在，忘掉钱吧。纯粹从应用程序逻辑的角度来看，func 可能会导致应用程序依赖的某个状态发生突变；即使func 为相同的输入返回相同的输出，它仍然不能被视为“无状态”。例如，如果在stream.map(func) 中，func 将每个元素添加到列表中，然后应用程序使用该列表，则生成的列表将取决于在运行时如何调用func。这是函数式程序员所讨厌的。

如果我们做stream.forEach( e->log(e) )，它是无状态的吗？如果

• 我们不关心log的成本
• log() 可以同时调用
• 我们不关心日志条目的顺序
• 日志条目对此应用程序的逻辑没有影响

Answer 3

如有疑问，只需查看文档即可进行具体操作。例子：

1. Stream.map映射器参数：

   mapper - 适用于每个元素的无干扰、无状态函数

   这里的文档明确指出该函数必须是无状态的。

2. Stream.forEach动作参数：

   action - 对元素执行的非干扰操作

   这里没有指定动作是无状态的，所以它可以是有状态的。

一般来说，它总是明确地写在每个方法文档上。

Answer 4

有状态流 lambda 示例：

• collect(Collector)：Collector 根据定义是有状态的，因为它必须收集集合（状态）中的所有元素。
• forEach(Consumer)：Consumer 根据定义是有状态的，除非它是一个黑洞（无操作）。
• peek(Consumer)：Consumer 根据定义是有状态的，因为如果不将其存储在某处（例如日志），为什么还要偷看。

所以，Collector 和 Consumer 是两个定义为有状态的 lambda 接口。

所有其他的，例如Predicate、Function、UnaryOperator、BinaryOperator 和 Comparator，应该是无状态的。