Example Analysis of parallelStream concurrency Security in Java8

This article mainly introduces the example analysis of parallelStream concurrency security in Java8, which is very detailed and has certain reference value. Friends who are interested must finish it!

Background

Java8's stream interface greatly reduces the complexity of for circular writing. Stream provides a series of aggregation interfaces such as map/reduce/collect and supports concurrent operations: parallelStream.

In the process of crawler development, it is often encountered to traverse a large collection to do repeated operations, at this time if the use of serial execution will be quite time-consuming, so it will generally use multi-threading to speed up. Java8's paralleStream provides concurrent execution capabilities using the fork/join framework. But if it is not used properly, it is easy to get into misunderstanding.

Is Java8's paralleStream thread safe?

A simple example, in the following code, use stream's forEach interface to traverse 1-10000 and insert it into three ArrayList. The insertion of the first list uses serial traversal, the second uses paralleStream, and the third uses paralleStream while synchronizing the insert list operation with ReentryLock:

Private static List list1 = new ArrayList (); private static List list2 = new ArrayList (); private static List list3 = new ArrayList (); private static Lock lock = new ReentrantLock (); public static void main (String [] args) {IntStream.range (0, 10000) .forEach (list1::add); IntStream.range (0, 10000). Parallel (). ForEach (list2::add); IntStream.range (0, 10000) .forEach (I-> {lock.lock (); try {list3.add (I);} finally {lock.unlock ()) }}); System.out.println ("size of serial execution:" + list1.size ()); System.out.println ("size of parallel execution:" + list2.size ()); System.out.println ("size of locked parallel execution:" + list3.size ());}

Execution result:

Size of serial execution: 10000

Size of parallel execution: 9595

Size of locked parallel execution: 10000

And the size of the parallel execution in each result is different, while the serial and locked results are always correct. Obviously, the actions performed in stream.parallel.forEach () are not thread-safe.

So since paralleStream is not thread-safe, do all non-atomic operations in it have to be locked? I found the answer on stackOverflow:

Https://codereview.stackexchange.com/questions/60401/using-java-8-parallel-streams

Https://stackoverflow.com/questions/22350288/parallel-streams-collectors-and-thread-safety

In the answers to the above two questions, it is confirmed that the forEach interface of paralleStream can not guarantee synchronization, and a solution is also proposed: using collect and reduce interfaces.

Http://docs.oracle.com/javase/tutorial/collections/streams/parallelism.html

The concurrent operation of stream is also introduced in Javadoc:

The Collections Framework provides synchronization wrappers, which add automatic synchronization to an arbitrary collection, making it thread-safe.

The Collections framework provides synchronous wrappers that make the operations in it thread safe.

So next, let's take a look at how the collect interface is used.

Collect interface of stream

Don't talk too much and go straight to the source code, the collect method handle in Stream.java:

R collect (Collector