在ParallelStream中使用java isomicReference线程是否安全？

Question

我使用parallelStream来获取数组中最长的字符串,代码跟随,每次运行时,我得到不同的结果.AtomicReference假设即使在parallelStream中使用时也是线程安全的？但为什么会这样呢？

public static void main(String[] args) {
  AtomicReference<String> longest = new AtomicReference<>();
  LongAccumulator accumulator = new LongAccumulator(Math::max, 0);
  List<String> words = Arrays.asList("him", "he", "thanks", "strings", "congratulations", "platform");
  words.parallelStream().forEach(
          next -> longest.updateAndGet(
                     current -> {
                        String result = next.length() > accumulator.intValue() ? next : current;
                        accumulator.accumulate(next.length());
                        return result;
                     }
                  )
  );
  System.out.println(longest.get());
}

有一段时间,我打印出"祝贺",有时我打印出"平台".

Answer 1

您正在调用LongAccumulator.intValue()哪个记录为:

在缩小基元转换后,将当前值作为int 返回.

并按照get()方法的方法,我们将学习:

返回当前值.返回的值不是原子快照; 在没有并发更新的情况下调用会返回准确的结果,但是在计算值时发生的并发更新可能不会合并.

因此,虽然AtomicReference.updateAndGet操作是线程安全的,但是并发调用LongAccumulator.intValue()和LongAccumulator.accumulate不是.A LongAccumulator用于执行并发accumulate操作,然后在完成所有累积操作后获取结果.请注意,即使get()返回正确的快照,调用intValue()和后续操作accumulate()是两个不同的,因此非原子操作的事实使得操作仍然容易发生数据争用.

在大多数情况下,如果您发现自己试图操纵a中的数据结构forEach,则使用错误的工具来完成工作,从而使代码变得不必要地复杂且容易出错.正如Clayn 在评论中暗示的那样,一个 words.parallelStream().max(Comparator.comparingInt(String::l??ength))意志就会简明扼要地完成工作.