Java,哪个线程是顺序流执行的？

Question

在阅读有关流的文档时,我遇到了以下句子:

• ...试图从行为参数中访问可变状态会给你一个错误的选择...如果你没有同步访问那个状态,你就有数据竞争,因此你的代码被破坏...... [1]

• 如果行为参数确实有副作用...... [没有]保证在同一个线程中执行同一流管道中"相同"元素的不同操作.[2]

• 对于任何给定元素,可以在任何时间以及库选择的任何线程中执行该动作.[3]

这些句子不区分顺序流和并行流.所以我的问题是:

1. 在哪个线程中执行顺序流的管道？它始终是调用线程还是可以自由选择任何线程的实现？
2. 在流是顺序的情况下,哪个线程是forEach终端操作的action参数？
3. 使用顺序流时是否必须使用任何同步？

---

• [1 + 2] https://docs.oracle.com/javase/8/docs/api/java/util/stream/package-summary.html
• [3] https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html#forEach-java.util.function.Consumer-

Answer 1

这一切都归结为基于规范的保证，以及当前实现可能具有超出保证的其他行为的事实。

Java 语言架构师 Brian Goetz 在相关问题中就规范提出了相关观点：

规范的存在是为了描述调用者可以依赖的最小保证，而不是描述实现的作用。

[...]

当规范说“不保留属性 X”时，并不意味着属性 X 永远不会被遵守；这意味着实现没有义务保留它。[...]（HashSet不保证迭代其元素会保留它们插入的顺序，但这并不意味着这不会意外发生 - 你只是不能指望它。）

这一切都意味着，即使当前的实现碰巧具有某些行为特征，也不应该依赖它们，也不应该假设它们在库的新版本中不会改变。

顺序流管道线程

顺序流的管道在哪个线程中执行？它总是调用线程还是一个实现可以自由选择任何线程？

当前的流实现可能会也可能不会使用调用线程，并且可能会使用一个或多个线程。由于 API 没有指定这些内容，因此不应依赖此行为。

forEach执行线程

如果流是顺序的，则 forEach 终端操作的 action 参数在哪个线程中执行？

虽然当前的实现使用现有线程，但这不能依赖，因为文档指出线程的选择取决于实现。事实上，不能保证元素不会由不同的线程处理不同的元素，尽管当前的流实现也不会这样做。

根据 API：

对于任何给定的元素，可以在库选择的任何时间和任何线程中执行该操作。

请注意，虽然 API 在讨论遇到顺序时专门调用并行流，但 Brian Goetz 澄清了这一点，以阐明该行为的动机，而不是任何行为特定于并行流：

此处明确指出平行案例的目的是为了教学[...]。然而，对于不了解并行性的读者来说，几乎不可能不假设forEach会保留遭遇顺序，因此添加这句话是为了帮助阐明动机。

使用顺序流进行同步

使用顺序流时是否必须使用同步？

当前的实现可能会起作用，因为它们对顺序流的方法使用单个线程forEach。但是，由于流规范不保证它，因此不应依赖它。因此，应该使用同步，就好像这些方法可以由多个线程调用一样。

也就是说，流文档特别建议不要使用需要同步的副作用，并建议使用归约操作而不是可变累加器：

许多可能想使用副作用的计算可以更安全、更有效地表达，而不会产生副作用，例如使用归约而不是可变累加器。[...] 少量流操作，例如 forEach() 和 peek()，只能通过副作用进行操作；这些应该小心使用。

作为如何将不适当地使用副作用的流管道转换为不使用副作用的流管道的示例，以下代码在字符串流中搜索与给定正则表达式匹配的字符串，并将匹配项放入列表中。

     ArrayList<String> results = new ArrayList<>();
     stream.filter(s -> pattern.matcher(s).matches())
           .forEach(s -> results.add(s));  // Unnecessary use of side-effects!

该代码不必要地使用了副作用。如果并行执行，ArrayList 的非线程安全性会导致不正确的结果，并且添加所需的同步会导致争用，从而破坏并行性的好处。此外，在这里使用副作用是完全没有必要的；forEach() 可以简单地替换为更安全、更高效且更适合并行化的归约操作：

     List<String>results =
         stream.filter(s -> pattern.matcher(s).matches())
               .collect(Collectors.toList());  // No side-effects!