stream 并行操作

如果要使stream中的操作并行,使用起来非常简单,只要加parallel()就可以了

    public static void main(String[] args) throws Exception {
        /**
         * 并行操作:parallel()
         * 顺序操作:sequential()
         */
        Optional<Integer> max = Stream.iterate(1, x -> x + 1).limit(20000).parallel().peek(x -> {
            System.out.println(Thread.currentThread().getName());
        }).max(Integer::compare);
        System.out.println(max.get());
    }

反之如果,想要让并行转换为串行也只要加sequential()

    public static void main(String[] args) throws Exception {
        /**
         * 并行操作:parallel()
         * 顺序操作:sequential()
         */
        Optional<Integer> max = Stream.iterate(1, x -> x + 1).limit(20000).parallel().sequential().peek(x -> {
            System.out.println(Thread.currentThread().getName());
        }).max(Integer::compare);
        System.out.println(max.get());
    }

sequential()和parallel()的前后位置决定了最后谁生效,最后一个声明会生效.

此外,stream api流用的线程池是

ForkJoinPool.commonPool()

内部默认核心线程数量是cpu的核心数

/**
     * Creates a {@code ForkJoinPool} with parallelism equal to {@link
     * java.lang.Runtime#availableProcessors}, using the {@linkplain
     * #defaultForkJoinWorkerThreadFactory default thread factory},
     * no UncaughtExceptionHandler, and non-async LIFO processing mode.
     *
     * @throws SecurityException if a security manager exists and
     *         the caller is not permitted to modify threads
     *         because it does not hold {@link
     *         java.lang.RuntimePermission}{@code ("modifyThread")}
     */
    public ForkJoinPool() {
        this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()),
             defaultForkJoinWorkerThreadFactory, null, false);
    }

如果需要手动执行线程数的话可以使用

//设置为5个线程加上主线程就是6个线程在同时执行
System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism","5");

当然也可以在java程序启动时,给定参数

-D java.util.concurrent.ForkJoinPool.common.parallelism=5
也可以达到相同的效果

stream 并行操作

推荐阅读更多精彩内容