\u672c\u6587\u4e3b\u8981\u7814\u7a76\u4e00\u4e0bparallelStream\u600e\u4e48\u4f7f\u7528\u81ea\u5b9a\u4e49\u7684\u7ebf\u7a0b\u6c60<\/p>\n
java\/util\/concurrent\/ForkJoinPool.java<\/p>\n
public class ForkJoinPool extends AbstractExecutorService {\n\n public ForkJoinPool(int parallelism,\n ForkJoinWorkerThreadFactory factory,\n UncaughtExceptionHandler handler,\n boolean asyncMode) {\n this(checkParallelism(parallelism),\n checkFactory(factory),\n handler,\n asyncMode ? FIFO_QUEUE : LIFO_QUEUE,\n "ForkJoinPool-" + nextPoolId() + "-worker-");\n checkPermission();\n }\n\n private ForkJoinPool(int parallelism,\n ForkJoinWorkerThreadFactory factory,\n UncaughtExceptionHandler handler,\n int mode,\n String workerNamePrefix) {\n this.workerNamePrefix = workerNamePrefix;\n this.factory = factory;\n this.ueh = handler;\n this.config = (parallelism & SMASK) | mode;\n long np = (long)(-parallelism); \/\/ offset ctl counts\n this.ctl = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK);\n }\n\n private static ForkJoinPool makeCommonPool() {\n int parallelism = -1;\n ForkJoinWorkerThreadFactory factory = null;\n UncaughtExceptionHandler handler = null;\n try { \/\/ ignore exceptions in accessing\/parsing properties\n String pp = System.getProperty\n ("java.util.concurrent.ForkJoinPool.common.parallelism");\n String fp = System.getProperty\n ("java.util.concurrent.ForkJoinPool.common.threadFactory");\n String hp = System.getProperty\n ("java.util.concurrent.ForkJoinPool.common.exceptionHandler");\n if (pp != null)\n parallelism = Integer.parseInt(pp);\n if (fp != null)\n factory = ((ForkJoinWorkerThreadFactory)ClassLoader.\n getSystemClassLoader().loadClass(fp).newInstance());\n if (hp != null)\n handler = ((UncaughtExceptionHandler)ClassLoader.\n getSystemClassLoader().loadClass(hp).newInstance());\n } catch (Exception ignore) {\n }\n if (factory == null) {\n if (System.getSecurityManager() == null)\n factory = new DefaultCommonPoolForkJoinWorkerThreadFactory();\n else \/\/ use security-managed default\n factory = new InnocuousForkJoinWorkerThreadFactory();\n }\n if (parallelism < 0 && \/\/ default 1 less than #cores\n (parallelism = Runtime.getRuntime().availableProcessors() - 1) <= 0)\n parallelism = 1;\n if (parallelism > MAX_CAP)\n parallelism = MAX_CAP;\n return new ForkJoinPool(parallelism, factory, handler, LIFO_QUEUE,\n "ForkJoinPool.commonPool-worker-");\n }\n\n}<\/code><\/pre>\nparallelStream\u9ed8\u8ba4\u4f7f\u7528\u7684\u662fcommon\u7684ForkJoinPool\uff0c\u53ef\u4ee5\u901a\u8fc7\u7cfb\u7edf\u5c5e\u6027\u6765\u8bbe\u7f6eparallelism\u7b49<\/p><\/blockquote>\n
ForkJoinPoolFactoryBean<\/h2>\n
org\/springframework\/scheduling\/concurrent\/ForkJoinPoolFactoryBean.java<\/p>\n
public class ForkJoinPoolFactoryBean implements FactoryBean<ForkJoinPool>, InitializingBean, DisposableBean {\n\n private boolean commonPool = false;\n\n private int parallelism = Runtime.getRuntime().availableProcessors();\n\n private ForkJoinPool.ForkJoinWorkerThreadFactory threadFactory = ForkJoinPool.defaultForkJoinWorkerThreadFactory;\n\n @Nullable\n private Thread.UncaughtExceptionHandler uncaughtExceptionHandler;\n\n private boolean asyncMode = false;\n\n private int awaitTerminationSeconds = 0;\n\n @Nullable\n private ForkJoinPool forkJoinPool;\n\n \/\/......\n\n @Override\n public void destroy() {\n if (this.forkJoinPool != null) {\n \/\/ Ignored for the common pool.\n this.forkJoinPool.shutdown();\n\n \/\/ Wait for all tasks to terminate - works for the common pool as well.\n if (this.awaitTerminationSeconds > 0) {\n try {\n this.forkJoinPool.awaitTermination(this.awaitTerminationSeconds, TimeUnit.SECONDS);\n }\n catch (InterruptedException ex) {\n Thread.currentThread().interrupt();\n }\n }\n }\n }\n\n} <\/code><\/pre>\nspring3.1\u63d0\u4f9b\u4e86ForkJoinPoolFactoryBean\uff0c\u53ef\u4ee5\u7528\u4e8e\u521b\u5efa\u5e76\u6258\u7ba1forkJoinPool<\/p><\/blockquote>\n
\u793a\u4f8b<\/h2>\n\u914d\u7f6e<\/h3>\n@Configuration\npublic class ForkJoinConfig {\n\n @Bean\n public ForkJoinPoolFactoryBean forkJoinPoolFactoryBean() {\n ForkJoinPoolFactoryBean factoryBean = new ForkJoinPoolFactoryBean();\n factoryBean.setCommonPool(false);\n \/\/ NOTE LIFO_QUEUE FOR working steal from tail of queue\n factoryBean.setAsyncMode(true); \/\/ NOTE true FIFO_QUEUE, false LIFO_QUEUE\n factoryBean.setParallelism(10);\n \/\/ factoryBean.setUncaughtExceptionHandler();\n factoryBean.setAwaitTerminationSeconds(60);\n return factoryBean;\n }\n}<\/code><\/pre>\n\u4f7f\u7528<\/h3>\n @Autowired\n ForkJoinPoolFactoryBean forkJoinPoolFactoryBean;\n\n public void streamParallel() throws ExecutionException, InterruptedException {\n List<TodoTask> result = forkJoinPoolFactoryBean.getObject().submit(new Callable<List<TodoTask>>() {\n @Override\n public List<TodoTask> call() throws Exception {\n return IntStream.rangeClosed(1, 20).parallel().mapToObj(i -> {\n log.info("thread:{}", Thread.currentThread().getName());\n return new TodoTask(i, "name"+i);\n }).collect(Collectors.toList());\n }\n }).get();\n result.stream().forEach(System.out::println);\n }<\/code><\/pre>\ncommon\u7684workerName\u524d\u7f00\u4e3aForkJoinPool.commonPool-worker-
\u81ea\u5b9a\u4e49\u7684workerName\u524d\u7f00\u9ed8\u8ba4\u4e3aForkJoinPool- nextPoolId() -worker-<\/p><\/blockquote>\n
\u5c0f\u7ed3<\/h2>\n
parallelStream\u9ed8\u8ba4\u4f7f\u7528\u7684\u662fcommonPool\uff0c\u662fstatic\u4ee3\u7801\u5757\u9ed8\u8ba4\u521d\u59cb\u5316\uff0c\u9488\u5bf9\u4e2a\u522b\u573a\u666f\u53ef\u4ee5\u81ea\u5b9a\u4e49ForkJoinPool\uff0c\u5c06parallelStream\u4f5c\u4e3a\u4e00\u4e2a\u4efb\u52a1\u4e22\u8fdb\u53bb\uff0c\u8fd9\u6837\u5b50\u5c31\u4e0d\u4f1a\u5f71\u54cd\u9ed8\u8ba4\u7684commonPool\u3002<\/p>\n