This method should be overridden when the {@link #spliterator()} * method cannot return a spliterator that is {@code IMMUTABLE}, * {@code CONCURRENT}, or late-binding. (See {@link #spliterator()} * for details.) * * @implSpec * The default implementation creates a parallel {@code Stream} from the * collection's {@code Spliterator}. * * @return a possibly parallel {@code Stream} over the elements in this * collection * @since 1.8 */ default Stream parallelStream() { return StreamSupport.stream(spliterator(), true); } ``` `@return a possibly parallel` parallelStream`ForkJoinPool`**** [Custom thread pool in Java 8 parallel stream](https://stackoverflow.com/questions/21163108/custom-thread-pool-in-java-8-parallel-stream) > The parallel streams use the default`ForkJoinPool.commonPool`which[by default has one less threads as you have processors](http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ForkJoinPool.html), as returned by`Runtime.getRuntime().availableProcessors()`(This means that parallel streams use all your processors because they also use the main thread) ```java public static void main(String[] args) { IntStream list = IntStream.range(0, 10); Set threadSet = new HashSet(); // list.parallel().forEach(i -> { Thread thread = Thread.currentThread(); System.err.println("integer" + i + "" + "currentThread:" + thread.getName()); threadSet.add(thread); }); System.out.println("all threads" + Joiner.on("").join(threadSet.stream().map(Thread::getName).collect(Collectors.toList()))); } ``` [13932958-263c866e35df81e5.png] parallelStream`ForkJoinPool.commonPool` [13932958-e1836ce1a66f41ec.png] java 8 > `ForkJoinPool`7.2/`Runtime.getRuntime().available- Processors()` `java.util.concurrent.ForkJoinPool.common. parallelism` `System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism","12");` `ForkJoinPool` ```java // System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "12"); System.out.println(ForkJoinPool.getCommonPoolParallelism());// 12 System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "20"); System.out.println(ForkJoinPool.getCommonPoolParallelism());// 12 ``` `java.util.concurrent.ForkJoinPool.common.parallelism`finalJVM`ForkJoinPool` ```java IntStream range = IntStream.range(1, 100000); // parallelism new ForkJoinPool(parallelism).submit(() -> range.parallel().forEach(System.out::println)).get(); ``` parallelStream**parallelStream**IOparallelStreamparallelStream > Using a ForkJoinPool and submit for a parallel stream does not reliably use all threads. If you look at this ([Parallel stream from a HashSet doesn't run in parallel](https://stackoverflow.com/questions/28985704/parallel-stream-from-a-hashset-doesnt-run-in-parallel)) and this ([Why does the parallel stream not use all the threads of the ForkJoinPool?](https://stackoverflow.com/questions/36947336/why-does-the-parallel-stream-not-use-all-the-threads-of-the-forkjoinpool)), you'll see the reasoning. ```java // CPU System.out.println(Runtime.getRuntime().availableProcessors());// 4 // parallelStream System.out.println(ForkJoinPool.getCommonPoolParallelism());// 3 ``` parallelStreamCPU1CPU ```java static final int MAX_CAP = 0x7fff; // max #workers - 1 // public ForkJoinPool() { this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()), defaultForkJoinWorkerThreadFactory, null, false); }bs-channel ``` ## parallelStream[Fork/Join ](https://www.infoq.cn/article/fork-join-introduction/) Fork/Join - Fork/Join [13932958-dbceae46ea7c15c3.png] Collector*supplier*,*accumulator**combiner* - Fork/Join,CPU,cpu,work-stealingFork/Join,work-stealing [13932958-ffe0d5ddd7101bbc.png] ## parallelStream parallelStream 1) lambda 2) (forkAndJoin) [-CPU](https://blog.csdn.net/larva_s/article/details/90403578) [006-ThreadJava8parallelStream](https://blog.csdn.net/blueskybluesoul/article/details/82817007) (parallel)Stream parallelStream 1) **parallelStream** ```java List values = new ArrayList(); IntStream.range(1, 10000).parallel().forEach(values::add); System.out.println(values.size()); ``` values10000null Listadd addsize++`elementData[size++]`size++List,addListnull1015223349731015=10+10/222=15+15/233=22+22/2... `grow()`ArrayList`oldCapacity + (oldCapacity >> 1)`1.5 ```java /** * Increases the capacity to ensure that it can hold at least the * number of elements specified by the minimum capacity argument. * * @param minCapacity the desired minimum capacity */ private void grow(int minCapacity) { // overflow-conscious code int oldCapacity = elementData.length; int newCapacity = oldCapacity + (oldCapacity >> 1);// 1.5 if (newCapacity - minCapacity < 0) newCapacity = minCapacity; if (newCapacity - MAX_ARRAY_SIZE > 0) newCapacity = hugeCapacity(minCapacity); // minCapacity is usually close to size, so this is a win: elementData = Arrays.copyOf(elementData, newCapacity);// } /** * Appends the specified element to the end of this list. * * @param e element to be appended to this list * @return true (as specified by {@link Collection#add}) */ public boolean add(E e) { ensureCapacityInternal(size + 1); // Increments modCount!! array elementData[size++] = e;// Size return true; } ``` `collect()``reduce()` ```java List values = new ArrayList(); for (int i = 0; i < 10000; i++) { values.add(i); } List collect = values.stream().parallel().collect(Collectors.toList()); System.out.println(collect.size()); ``` 2) parallelStream CPUCPUCPU - I/O I/OI/OI/OCPUI/OI/O - CPU CPU 3) parallelStreamCPU 4) 5) NQNQ N * Q N * Q>100001000 6) parallelStreamStream,**,ThreadLocal 7) **** 8) lambdaparallel streamworkersparallel streamscommon ForkJoinPool