微信公众平台

简介

Java8发布自2012年，距今已经有相当长的一段时间了，作为现在互联网公司当中最流行的一个Java版本，其许多特性也被大众广泛认可。Stream作为其中一个实用的特性，也常常被人津津乐道。ParallelStream是Stream中并行的一个版本，它有自己不可替代的地位，但是滥用会出现线上问题哦~

之前线上遇到了一个超时问题，最后排查到是因为ParallelStream导致的，这次采坑的经历也更大家分享一下，顺便说一下ParallelStream的原理与使用方式。

问题分析

不久前的一个晚上，线上一个节点出现了异常抖动，并导致了一些请求的超时，但持续时间非常短。

1.案发现场

按照现有的监控，我们能得到的一些指标

那个节点rt上涨，并伴随着各种超时
线程数、cpu、load均有上涨
当时没有fgc，ygc时间也不长
io、流量、网络均正常
上下文切换次数有较大幅度的下降
当时下游应用、redis、数据库等都没有抖动

2.初步分析

其实上述现象比较常见，一般都是因为某些原因导致某个地方堵住了，为了保证应用的吞吐量，应用的线程数会涨上去，同时伴随着rt增高，cpu、load等指标上涨的情况。但是有一个地方很奇怪，就是当时上下游并没有存在任何hang住的地方。

因为问题持续的时间非常短，发生的时机非常随机，排查问题的利器jstack几乎没用，排查工作一度陷入僵局。只能根据日志，通过想象一点点尝试还原案发现场。日志观察也是排查这种瞬间毛刺的一个好方法，有时间可以专门出一期如何排查线上问题

3.解决问题

很幸运的是，在日志上我们发现了一些端倪。问题发生前，有一个调用频率很低的请求的日志，并且伴随着每次问题，这个请求都在之前出现，就好比连环凶杀案现场，监控总是拍到一个人，那他肯定有嫌疑。最终我们在代码里面发现了今天我们的主角的名字ParallelStream。

解决问题很简单，把ParallelStream换成普通的线程池写法就行了

至于为什么会导致这个问题，接下来我们来深入探究一下。

源码分析

1.并行与线程池

(1) Stream的实现

相对于一般的Stream来说，ParallelStream多了一层并行的语义，顾名思义，在流处理的时候它是并行的。一般涉及到并行就会与线程池相关，ParallelStream的实现也是依赖于线程池的。

 final <R> R evaluate(TerminalOp<E_OUT, R> terminalOp) {        assert getOutputShape() == terminalOp.inputShape();        if (linkedOrConsumed)            throw new IllegalStateException(MSG_STREAM_LINKED);        linkedOrConsumed = true;
        return isParallel()               ? terminalOp.evaluateParallel(this, sourceSpliterator(terminalOp.getOpFlags()))               : terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags()));    }

重点关注evaluateParallel方法，并行模式下，会走到这个分支。我们来看看evaluateParallel其中的一个实现

  @Override        public <P_IN> R evaluateParallel(PipelineHelper<T> helper,                                         Spliterator<P_IN> spliterator) {            return new ReduceTask<>(this, helper, spliterator).invoke().get();        }

invoke的实现

 public final V invoke() {        int s;        if ((s = doInvoke() & DONE_MASK) != NORMAL)            reportException(s);        return getRawResult();    }   
    private int doInvoke() {        int s; Thread t; ForkJoinWorkerThread wt;        return (s = doExec()) < 0 ? s :            ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?            (wt = (ForkJoinWorkerThread)t).pool.            awaitJoin(wt.workQueue, this, 0L) :            externalAwaitDone();    }

渐渐地我们引出了ForkJoin的Pool，对于ForkJoin不熟悉的小伙伴，不要着急，后面我们会详细介绍。

看到这里，我们知道了ParallelStream与ForkJoinPool的关系，但是我们在调用ParallelStream中并没有显式的传入线程池变量，那么ParallelStream用的到底是什么线程池呢？这里我就不贴冗长的源码了，直接说结果

如果当前线程就是ForkJoinWorkerThread，那么线程池就是当前线程所属的ForkJoinPool
反之这个线程池就是ForkJoinPool.commonPool()

(2) CommonPool

 static {        //冗长的参数设置代码省略
        commonMaxSpares = DEFAULT_COMMON_MAX_SPARES;        //默认线程工厂        defaultForkJoinWorkerThreadFactory =            new DefaultForkJoinWorkerThreadFactory();        modifyThreadPermission = new RuntimePermission("modifyThread");
        //开始建commonPool啦        common = java.security.AccessController.doPrivileged            (new java.security.PrivilegedAction<ForkJoinPool>() {                //重点关注makeCommonPool                public ForkJoinPool run() { return makeCommonPool(); }});        int par = common.config & SMASK; // report 1 even if threads disabled        commonParallelism = par > 0 ? par : 1;    }

    private static ForkJoinPool makeCommonPool() {        int parallelism = -1;        ForkJoinWorkerThreadFactory factory = null;        UncaughtExceptionHandler handler = null;        //从环境变量中获取参数        try {  // ignore exceptions in accessing/parsing properties            String pp = System.getProperty                ("java.util.concurrent.ForkJoinPool.common.parallelism");            String fp = System.getProperty                ("java.util.concurrent.ForkJoinPool.common.threadFactory");            String hp = System.getProperty                ("java.util.concurrent.ForkJoinPool.common.exceptionHandler");            if (pp != null)                parallelism = Integer.parseInt(pp);            if (fp != null)                factory = ((ForkJoinWorkerThreadFactory)ClassLoader.                           getSystemClassLoader().loadClass(fp).newInstance());            if (hp != null)                handler = ((UncaughtExceptionHandler)ClassLoader.                           getSystemClassLoader().loadClass(hp).newInstance());        } catch (Exception ignore) {        }        if (factory == null) {            if (System.getSecurityManager() == null)                factory = new DefaultCommonPoolForkJoinWorkerThreadFactory();            else // use security-managed default                factory = new InnocuousForkJoinWorkerThreadFactory();        }        //并行度为处理器数-1，最小是1        if (parallelism < 0 && // default 1 less than #cores            (parallelism = Runtime.getRuntime().availableProcessors() - 1) <= 0)            parallelism = 1;        if (parallelism > MAX_CAP)            parallelism = MAX_CAP;        return new ForkJoinPool(parallelism, factory, handler, LIFO_QUEUE,                                "ForkJoinPool.commonPool-worker-");    }

这个线程池commonPool的初始化在ForkJoinPool#makeCommonPool
并行度默认处理器数量减1（不是核数减1）可以通过java.util.concurrent.ForkJoinPool.common.parallelism参数手动设置

(3) 小结

简单总结一下，ParallelStream背后依赖于ForkJoinPool.commonPool线程池，这个池子大小为处理器数减1，可以通过环境变量控制

2.ForkJoin

(1) 简介

ForkJoin作为java8的另一大特性，我们另写一篇文章专门探究都是不过分的，这里也只能简单入个门，帮助理解ParallelStream。

先来看一张著名的图

用大白话来解构一下这张图就是领导手里有一个任务，他拆成两个任务分给手下两个人，等他们都完成了，领导再把两个手下的任务报告合并起来，作为这整个任务的报告，其中“任务”类比为复杂的计算任务，“手下”与“领导”都是线程。大家是不是觉得虽然有3个人（三个线程），但是这个领导啥都不干（主线程啥都没干），就干等着，十分不合理？所以实际情况是领导（主线程）把任务拆分成两个（Fork），一个给手下，一个给自己。至此这个复杂的任务已经拆分了一次了，那么要是“手下”还是觉得任务太难太多怎么办呢？他自己也可以当一次“领导”，按照上面的步骤把任务拆分给“手下”。这个任务中用到的人数（线程数），我们称为并行度（parallelism）

在ForkJoinPool中，每个线程都有自己的任务队列，并且实际上任务会有快有慢，为了更好的利用线程，这里会有一种“工作窃取”机制——有些“手下”干得慢，就帮别的“手下”多干点活儿。

ParallelStream的底层线程池就是ForkJoinPool，进行遍历的时候，也会对集合中的元素进行拆分，并行着进行遍历。

(2) 工作窃取

普通的线程池，应用所有worker公用一个任务队列，所有没有窃取这种说法（任务是大家的，不是自己的）。但是ForkJoinPool每个Woker会有自己的工作队列（workQueue），所以会有窃取的概念。

  //开始跑任务    public void run() {        if (workQueue.array == null) { // only run once            Throwable exception = null;            try {                onStart();                //重点关注                pool.runWorker(workQueue);            } catch (Throwable ex) {                exception = ex;            } finally {                try {                    onTermination(exception);                } catch (Throwable ex) {                    if (exception == null)                        exception = ex;                } finally {                    pool.deregisterWorker(this, exception);                }            }        }    }
    final void runWorker(WorkQueue w) {        w.growArray();                   // allocate queue        int seed = w.hint;               // initially holds randomization hint        int r = (seed == 0) ? 1 : seed;  // avoid 0 for xorShift        for (ForkJoinTask<?> t;;) {            //scan获取任务（工作窃取在吗在这儿）            if ((t = scan(w, r)) != null)                w.runTask(t);            else if (!awaitWork(w, r))                break;            r ^= r << 13; r ^= r >>> 17; r ^= r << 5; // xorshift        }    }
    private ForkJoinTask<?> scan(WorkQueue w, int r) {        WorkQueue[] ws; int m;        if ((ws = workQueues) != null && (m = ws.length - 1) > 0 && w != null) {            //各种不为空判断            int ss = w.scanState;                     // initially non-negative            for (int origin = r & m, k = origin, oldSum = 0, checkSum = 0;;) {                WorkQueue q; ForkJoinTask<?>[] a; ForkJoinTask<?> t;                int b, n; long c;                if ((q = ws[k]) != null) {                    //找个队列（k的计算一直要往上追溯，实际是个随机值）                    if ((n = (b = q.base) - q.top) < 0 &&                        (a = q.array) != null) {      // non-empty                        long i = (((a.length - 1) & b) << ASHIFT) + ABASE;                        //b是尾端下标，ASHIFT是数组元素偏移量，ABASE是基础的偏移量，使用偏移量方便后面用cas                        if ((t = ((ForkJoinTask<?>)                                  U.getObjectVolatile(a, i))) != null &&                            q.base == b) {                            if (ss >= 0) {                                //>=0说明非空闲，那肯定有任务可以偷                                if (U.compareAndSwapObject(a, i, t, null)) {                                    //cas获取任务                                    q.base = b + 1;                                    if (n < -1)       // signal others                                        signalWork(ws, q);                                    return t;                                }                            }                            else if (oldSum == 0 &&   // try to activate                                     w.scanState < 0)                                tryRelease(c = ctl, ws[m & (int)c], AC_UNIT);                                //明明有任务，偷懒不干活，叫起来干活                        }                        //任务获取失败，说明发生竞争，重新开始随机                        if (ss < 0)                   // refresh                            ss = w.scanState;                        r ^= r << 1; r ^= r >>> 3; r ^= r << 10;                        origin = k = r & m;           // move and rescan                        oldSum = checkSum = 0;                        continue;                    }                    checkSum += b;                }                if ((k = (k + 1) & m) == origin) {    // continue until stable                    没任务可偷，k自增遍历下个队列，与上掩码m和起始值origin相同说明已经遍历一个来回了                    if ((ss >= 0 || (ss == (ss = w.scanState))) &&                        oldSum == (oldSum = checkSum)) {                        if (ss < 0 || w.qlock < 0)    // already inactive                            break;                            //如果跑了一圈还是闲着的，那就算了，退出循环。                        int ns = ss | INACTIVE;       // try to inactivate                        long nc = ((SP_MASK & ns) |                                   (UC_MASK & ((c = ctl) - AC_UNIT)));                        w.stackPred = (int)c;         // hold prev stack top                        U.putInt(w, QSCANSTATE, ns);                        if (U.compareAndSwapLong(this, CTL, c, nc))                            ss = ns;                        else                            w.scanState = ss;         // back out                    }                    checkSum = 0;                }            }        }        return null;    }

整个执行任务的过程围绕工作窃取展开，随机从一个队列中获取任务，直到循环了一圈都没拿到任务。一般这个循环会按照线性自增进行遍历，但如果发生明显的竞争，则重新随机地选择一个队列开始。

代码还是相对晦涩的，尤其是变量命名，但细想这些变量命名又是一些常用变量的缩写，也算是雅信达了，大家有时间可以品一品。

(3) 小结

ForkJoin源码有助于我们更加深入理解ParallelStream

3.注意的地方

其实看完原理的分析，这个ParallelStream可能会涉及到的问题，就一目了然了

(1) 非线程安全集合的使用

我们在使用ParallelStream的过程中，稍微不注意，就会使用一些线程不安全的集合操作。比如说在ParallelStream中对ArrayList进行一些操作等。因为ParallelStream中的操作可以理解为都是并行的，使用线程不安全的集合就会造成问题。常见的一些线程不安全的集合，例如ArrayList，HashMap，HashSet等，在普通Stream中使用是没有问题的，但是在ParallelStream中使用，就会有问题。我们在使用过程中一定要注意这个问题，不然就会导致一系列问题，报并发异常，集合少元素，集合存在元素为空等现象。

举一反三，一般的多线程编程过程中，也要注意这一点，code review的时候经常会看到有通过在多线程里面往ArrayList里面塞元素，这其实是有问题的。

但是，注意到了这一点，我们就能随心所欲地使用ParallelStream了吗？答案也不是的，ParallelStream存在一个致命的问题。

(2) 全局公用的线程池CommonPool

这个问题藏得比较隐蔽。我们知道ParallelStream底层使用的是ForkJoinPool的CommonPool，这个线程池其实并不只有ParallelStream在用，CompletableFuture也会使用的，但是好就好在CompletableFuture是可以用线程池作为入参的，也就是说CompletableFuture是可以做到底层不使用CommonPool的。

但是，只有ParallelStream使用ForkJoinPool.CommonPool难道就不会有问题了吗？本身线程池的创建时参数的设置就是十分复杂的，只用一个线程池会有以下一系列的问题

一些本身rt不低的但需要减少rt的请求，伴随着不小的并发，使用这个线程数为执行器数减1的线程池，除了第一次请求，后续请求都不会达到想要的效果。可以理解为，本身一个部门就没多少人（CommonPool线程池数量不多），一下子来了超多的任务（请求高并发），任务肯定就积压了（实际吞吐量远不如一个请求一个线程）
因为不同类型的（cpu密集，io密集），不同要求（优先rt，优先吞吐量）的任务公用一个线程池，任务之间会相互影响。你的请求平时量不大你想要低rt，用到了ParallelStream，别人的请求量很大，想要减少线程切换提高吞吐量，也用到了ParallelStream，这两者在一起就会相互影响，最后rt不低，吞吐量不高。

(3) 小结

并发编程有许多要注意的地方，ParallelStream本质也是并发编程，一些注意的点不能忘记；

作为开发来说，你不知道你之前的程序员会用ParallelStream干了什么，也不知道你之后的程序员会干些什么。那么编写使用ParallelStream的代码就会非常危险。

稍佳实践

说了那么多注意点，那ParallelStream他的最佳实践是什么呢？这边抛砖引玉，简单列举一些稍佳的实践。

1.cpu密集型任务

ForkJoinPool.CommonPool的线程数就很明确得说明了，这个线程池其实更适合cpu密集型的任务，同样的，ParallelStream也是更加适合进行一些cpu密集的操作。比如说统计一篇文章里面每个字母出现的次数等等之类的，这就有发挥大家的想象了。

2.使用ManagedBlocker

这个ManagedBlocker其实是ForkJoinPool中的一个特性，他可以显式地告诉线程池当前执行卡住了，如果并行度不够，那就要另外重新再开线程保证任务的执行。

public static interface ManagedBlocker {        /**         * Possibly blocks the current thread, for example waiting for         * a lock or condition.         *         * @return {@code true} if no additional blocking is necessary         * (i.e., if isReleasable would return true)         * @throws InterruptedException if interrupted while waiting         * (the method is not required to do so, but is allowed to)         * 具体用于阻塞的方法，需要自己实现，比如说等待一把锁，或者一个Condition         */        boolean block() throws InterruptedException;
        /**         * Returns {@code true} if blocking is unnecessary.         * @return {@code true} if blocking is unnecessary         * 返回当前阻塞是不是必要的         */        boolean isReleasable();    }

这个应该是官方给出的最优的解法了，这样一来，我们的ForkJoinPool就可以用来执行一些io密集型的任务了。但就实际效果来看，不是很建议，首先使用难度高不说，存在着很多更好，更简单的解决方案（没必要硬着头皮用）

3.自定义线程池提交ParallelStream

我们在前面“Stream的实现-doInvoke的实现”中，我们知道，ParallelStream优先使用的是当前线程所在的ForkJoinPool，所以我们可以通过向自定义线程池提交ParallelStream任务的方式，指定ParallelStream底层使用的线程池。

ForkJoinPool forkJoinPool = new ForkJoinPool(4);forkJoinPool.submit(() -> xxx.parallelStream().collect(Collectors.toList())).get();

但这个代码就有点脱了裤子放屁的感觉了，为什么不直接使用普通的线程池，普通的提交任务，而是用ParallelStream呢？

总结

ParallelStream作为并行计算的一把利器，其设计的出发点是好的，但是使用场景的错误，以及使用方式的错误也让它变得不是那么可靠。我们应该理解其原理，更加正确，更加合理的使用这把双刃剑。

若有收获，就点个赞吧

由一个线上问题引出的ParallelStream最佳实践

简介

问题分析

1.案发现场

2.初步分析

3.解决问题

源码分析

1.并行与线程池

(1) Stream的实现

(2) CommonPool

(3) 小结

2.ForkJoin

(1) 简介

(2) 工作窃取

(3) 小结

3.注意的地方

(1) 非线程安全集合的使用

(2) 全局公用的线程池CommonPool

(3) 小结

稍佳实践

1.cpu密集型任务

2.使用ManagedBlocker

3.自定义线程池提交ParallelStream

总结

投诉已提交