Is Java AtomicReference thread safe when used in parallelStream?

Question

I used parallelStream to get the longest string in the array, the code is the following, each time it starts, it gets a different result. AtomicReference be thread safe even when used in parallelStream? But why is this happening?

 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()); } 

at one time, I type “congratulations”, and for a while I type “platform”.

Answer 1

You call LongAccumulator.intValue() , which is documented as:

Returns the current value as int after narrowing the primitive conversion.

and following the link to the get() method, we find out:

Returns the current value. The return value is NOT an atomic snapshot; a call in the absence of simultaneous updates returns an accurate result, but parallel updates that occur during the calculation of the value may not be included.

Thus, although the AtomicReference.updateAndGet operation is thread safe, your simultaneous call to LongAccumulator.intValue() and LongAccumulator.accumulate not. A LongAccumulator designed to perform parallel accumulate operations, followed by retrieving the result after all accumulation operations are complete. Note that even if get() returned the correct snapshot, the fact that the call to intValue() and the subsequent accumulate() are two different, therefore non-atomic, operations made the operation still prone to data racing.

In most cases, if you are trying to manipulate data structures in forEach , you are using the wrong job tool, making the code unnecessarily complex and error prone. As Clayn hinted at the comment , words.parallelStream().max(Comparator.comparingInt(String::l‌​ength)) will do the job concisely and correctly.

Answer 2

Actually, I wrote about the problem that @Holger already mentioned, I was a bit late, but I am writing this as proof of @Holger's answer anyway . You can use the AtomicReference battery;

Here is the code:

 public static void main(String[] args) { AtomicReference < String > longest = new AtomicReference < > (); List < String > words = Arrays.asList("him", "he", "thanks", "strings", "congratulations", "platform"); words.parallelStream().forEach(next - > { longest.accumulateAndGet(next, (a, b) - > a != null && a.length() > b.length() ? a : b ); }); System.out.println(longest.get()); }