Shulou(Shulou.com)05/31 Report--

Today, Xiaobian will share with you the relevant knowledge points on how to solve the pit used by ParallelStream. The content is detailed and the logic is clear. I believe most people still know too much about this knowledge, so share this article for your reference. I hope you will gain something after reading this article. Let's find out together.

For example, the following code snippet makes people read it like poetry. However, if he didn't use it well, he would die.

List transactionsIds = widgets.stream() .filter(b -> b.getColor() == RED) .sorted((x,y) -> x.getWeight() - y.getWeight()) .mapToInt(Widget::getWeight) .sum();

This code has one key function, stream. It allows you to convert an ordinary list into a stream, and then you can manipulate the list in a pipeline-like way. Anyway, used ones are good.

Not familiar with these functions? You can refer to: "Map, flatMap everywhere, what do you mean?"

question is coming

What happens if we replace stream with parallelStream?

Literally, streams change from serial to parallel.

Since it is parallel, think about it with your ass and you will know that there must be thread safety issues. But we're not talking about thread-safe collections here. That's a low-level topic. At this stage, knowing how to use thread-safe collections in thread-unsafe environments is a basic skill.

This time, it was a performance problem of parallel flow.

We speak in code.

The following code opens eight threads, all of which are computing data using parallel streams. In the logic of execution, we let each task sleep for 1 second, which simulates the time-consuming wait for some I/O requests.

With stream, the program will return after 30 seconds, but we expect the program to return in more than 1 second because it is a parallel stream and deserves the title.

The test found that we waited a long time before the mission was completed.

static void paralleTest() { List numbers = Arrays.asList( 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29 ); final long begin = System.currentTimeMillis(); numbers.parallelStream().map(k -> { try { Thread.sleep(1000); System.out.println((System.currentTimeMillis() - begin) + "ms => " + k + " \t" + Thread.currentThread()); } catch (InterruptedException e) { e.printStackTrace(); } return k; }).collect(Collectors.toList()); } public static void main(String[] args) { // System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "20"); new Thread(() -> paralleTest()).start(); new Thread(() -> paralleTest()).start(); new Thread(() -> paralleTest()).start(); new Thread(() -> paralleTest()).start(); new Thread(() -> paralleTest()).start(); new Thread(() -> paralleTest()).start(); new Thread(() -> paralleTest()).start(); new Thread(() -> paralleTest()).start(); }

pit

In fact, this code takes different amounts of time to execute on different machines.

Since it is parallel, there must be a degree of parallelism. Too low for parallelism; too large for context switching. I am very frustrated to find that many advanced R & D, the thread pool of various parameters memorized thoroughly, a variety of tuning, even dare to turn an eye closed in the I/O intensive business using parallelStream.

To understand this parallelism, we need to look at the specific construction method. Find this code in the ForkJoinPool class.

try { // ignore exceptions in accessing/parsing properties String pp = System.getProperty ("java.util.concurrent.ForkJoinPool.common.parallelism"); if (pp != null) parallelism = Integer.parseInt(pp); fac = (ForkJoinWorkerThreadFactory) newInstanceFromSystemProperty( "java.util.concurrent.ForkJoinPool.common.threadFactory"); handler = (UncaughtExceptionHandler) newInstanceFromSystemProperty( "java.util.concurrent.ForkJoinPool.common.exceptionHandler"); } catch (Exception ignore) { } if (fac == null) { if (System.getSecurityManager() == null) fac = defaultForkJoinWorkerThreadFactory; else // use security-managed default fac = new InnocuousForkJoinWorkerThreadFactory(); } if (parallelism

< 0 && // default 1 less than #cores (parallelism = Runtime.getRuntime().availableProcessors() - 1) MAX_CAP) parallelism = MAX_CAP; 可以看到，并行度到底是多少，是由下面的参数来控制的。如果无法获取这个参数，则默认使用 CPU个数-1 的并行度。 可以看到，这个函数是为了计算密集型业务去设计的。如果你喂给它一大堆任务，它就会由并行执行退变成类似于串行的效果。 -Djava.util.concurrent.ForkJoinPool.common.parallelism=N 即使你使用-Djava.util.concurrent.ForkJoinPool.common.parallelism=N设置了一个初始值大小，它依然有问题。 因为，parallelism这个变量是final的，一旦设定，不允许修改。也就是说，上面的参数只会生效一次。 张三可能使用下面的代码，设置了并行度大小为20。 System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "20"); 李四可能用同样的方式，设置了这个值为30。那实际在项目中用的是哪个值，那就得问JVM是怎么加载的类信息了。 这种方式并不太非常靠谱。 一种解决方式 我们可以通过提供外置的forkjoinpool，也就是改变提交方式，来实现不同类型的任务分离。 代码如下所示，通过显式的代码提交，即可实现任务分离。 ForkJoinPool pool = new ForkJoinPool(30); final long begin = System.currentTimeMillis(); try { pool.submit(() ->

numbers.parallelStream().map(k -> { try { Thread.sleep(1000); System.out.println((System.currentTimeMillis() - begin) + "ms => " + k + " \t" + Thread.currentThread()); } catch (InterruptedException e) { e.printStackTrace(); } return k; }).collect(Collectors.toList())).get(); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); }

In this way, different scenes can have different parallelism. This approach is similar to CountDownLatch in that we need to manually manage resources.

The above is all the content of this article "How to solve the pit used by ParallelStream", thank you for reading! I believe everyone has a great harvest after reading this article. Xiaobian will update different knowledge for everyone every day. If you want to learn more knowledge, please pay attention to the industry information channel.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.