代码之家 › 专栏 › 技术社区 › Kartik

为什么不建议使用基于AtomicInteger的流解决方案?

atomicinteger java-stream thread-safety java-8 java

Kartik · 技术社区 · 7 年前

假设我有这个水果清单:

List<String> f = Arrays.asList("Banana", "Apple", "Grape", "Orange", "Kiwi");

我需要给每个水果预先准备一个序列号并打印出来。水果的顺序或序列号并不重要。所以这是一个有效的输出:

4. Kiwi
3. Orange
1. Grape
2. Apple
5. Banana

解1

AtomicInteger number = new AtomicInteger(0);

String result = f.parallelStream()
        .map(i -> String.format("%d. %s", number.incrementAndGet(), i))
        .collect(Collectors.joining("\n"));

解决方案2

String result = IntStream.rangeClosed(1, f.size())
        .parallel()
        .mapToObj(i -> String.format("%d. %s", i, f.get(i - 1)))
        .collect(Collectors.joining("\n"));

问题

为什么解决方案1是一种坏做法?我在很多地方见过 AtomicInteger 基础解决方案不好(比如 this answer ,特别是在并行流处理中(这就是我使用上面的并行流尝试遇到问题的原因)。

我看过这些问题/答案:
In which cases Stream operations should be stateful?
Is use of AtomicInteger for indexing in Stream a legit way?
Java 8: Preferred way to count iterations of a lambda?

他们只是提到(除非我遗漏了什么)“可能会发生意想不到的结果”。像什么?在这个例子中会发生这种情况吗?如果没有,你能给我举个例子吗?

至于“ 不保证映射器函数的应用顺序 “,好吧,这就是并行处理的本质,所以我接受它,而且在这个特定的例子中,顺序并不重要。

原子数 是线程安全的,所以它不应该是并行处理中的问题。

有人能举例说明在使用这种基于状态的解决方案时哪些情况下会出现问题吗?

3 回复 | 直到 7 年前

Andrew 7 年前

还要注意,尝试从行为参数访问可变状态会给您带来一个关于安全和性能 ;如果不同步对该状态的访问,则会发生数据争用,因此代码被破坏,但是 如果您确实同步了对该状态的访问,那么您可能会因为争用而破坏您想要从中受益的并行性。 最好的方法是完全避免流操作的状态行为参数;通常有一种方法可以重新构造流管道以避免状态。

^{Package

java.util.stream

, Stateless behaviors}

从线程安全性和正确性的角度来看,解决方案1没有错。不过,性能(作为并行处理的优势)可能会受到影响。

为什么解决方案1是一种坏做法?

我不会说这是一种不好的做法或是一些不可接受的事情。这只是不建议为了性能。

他们只是提到(除非我遗漏了什么)“可能会发生意想不到的结果”。像什么?

“意想不到的结果”是一个非常宽泛的术语,通常指不适当的同步,“刚才到底发生了什么?”-像行为一样。

在这个例子中会发生这种情况吗?

事实并非如此。你可能不会遇到问题。

如果没有,你能给我举个例子吗?

改变 AtomicInteger 对 int *,替换 number.incrementAndGet() 具有 ++number 还有一个。

^{*装箱

int

(例如,基于包装器、基于数组)以便在lambda中使用它}

Andrew 7 年前

看看斯图亚特马克的回答 here -他使用的是有状态谓词。

这是一些潜在的问题,但是如果你不关心它们或者不真正理解它们,你应该很好。

首先是order,它在当前的并行处理实现中展示,但是如果您不关心order,就像在您的示例中一样,您是可以的。

第二个是潜在速度 AtomicInteger 如果你关心这个,增加一个简单的int的速度会慢很多倍。

第三个更微妙。有时不能保证 map 将执行,例如,自Java-9以来:

 someStream.map(i -> /* do something with i and numbers */)
           .count();

这里的要点是,由于您正在计数,因此不需要进行映射,因此跳过了映射。一般来说,影响某些中间操作的元件不能保证到达终端操作。想象一下 map.filter.map 在这种情况下,第一个映射可能比第二个映射“看到”更多的元素,因为某些元素可能被过滤。所以不建议依赖于这个,除非你能准确地解释发生了什么。

在您的例子中,imo,您做您所做的事情是非常安全的;但是如果您稍微更改了代码,这就需要额外的推理来证明它的正确性。我会选择解决方案2,因为它对我来说更容易理解,而且它没有上面列出的潜在问题。

zack 7 年前

案例2-在API中,IntStream类的Notes通过1种for循环的增量步骤返回从startinclusive(inclusive)到endinclusive(inclusive)的顺序IntStream,因此并行流正在逐个处理它并提供正确的顺序。

 * @param startInclusive the (inclusive) initial value
 * @param endInclusive the inclusive upper bound
 * @return a sequential {@code IntStream} for the range of {@code int}
 *         elements
 */
public static IntStream rangeClosed(int startInclusive, int endInclusive) {

案例1-很明显,列表将并行处理,因此顺序将不正确。由于映射操作是并行执行的,因此由于线程调度的不同,相同输入的结果可能因运行而异,因此不能保证在同一线程中对同一流管道中的“相同”元素执行不同的操作,也不能保证映射器函数如何应用于流中的特定元素。

Source Java Doc