Java中关于CompletableFuture类的思考

最近遇到了一个OOM的问题,提示的是无法创建更多的线程,定位问题,发现是类似下面的这段代码出现了问题,用JConsole监测,发现某一时段线程数量忽然飙升,由此引发了下面的思考

情景再现

public class DemoController {
    private ExecutorService executorService = Executors.newWorkStealingPool(20);

    @RequestMapping("/test")
    public String test() {
        ExecutorService forkJoinPool = Executors.newWorkStealingPool(10);
        CompletableFuture[] completableFutures = new CompletableFuture[600];
        for (int i = 0; i < 600; i++) {
            int j = i;
            completableFutures[i] = CompletableFuture.runAsync(() -> {
                getAssociatedInfo(forkJoinPool);
            }, forkJoinPool);
        }
        CompletableFuture<Void> voidCompletableFuture = CompletableFuture.allOf(completableFutures);
        voidCompletableFuture.join();
        return "OK";
    }

    public String getAssociatedInfo(ExecutorService service) {
        CompletableFuture<String> trialAssociatedInfoCompletableFuture
                = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println("按理说你已在运行,不是吗");
                TimeUnit.SECONDS.sleep(100);
                System.out.println("你已经完成了");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "a";
        }, executorService);
        CompletableFuture<Void> voidCompletableFuture = CompletableFuture.allOf(trialAssociatedInfoCompletableFuture);
        voidCompletableFuture.join();
        return "ok";
    }
}
复制代码
Java中关于CompletableFuture类的思考

在这段代码中,http线程启600个任务,使用自定义的线程池并发数量为10个。每一个任务启一个子任务,使用类定义的线程池,并发数量为20个。在我的理解中,按理说最多多三十多个线程数量才对,但是短时间内居然飙升好几百,那么这几百个线程到底是如何产生的呢?在研究了源代码之后,终于理解了其中的奥秘。

代码探究

completableFutures[i] = CompletableFuture.runAsync(() -> {
                    getAssociatedInfo(forkJoinPool);
                }, forkJoinPool);
复制代码

这一句的作用是启异步任务,交由forkJoinPool线程池管理,当线程池数量不足10个时,启动一个线程,立即执行,当超过10个时,加入任务队列。

CompletableFuture<Void> voidCompletableFuture = CompletableFuture.allOf(completableFutures);
复制代码

allOf的作用是递归地构造完成树,汇总并返回成一个总任务,如下图所示:

Java中关于CompletableFuture类的思考

这里的任务1、任务2等就是我们前面定义的任务,每个任务对象存储着一个结果,当最终任务有结果时,必须要下面的汇总任务都有结果,进而每一个定义的任务都要有结果,通俗来说,就是对voidCompletableFuture的管理即为对所有定义任务的管理。

// 从多线程的角度,若任务未完成,会阻塞
    voidCompletableFuture.join();
    return "OK";
CompletableFuture->join():
    return reportJoin((r = result) == null ? waitingGet(false) : r);
CompletableFuture->waitingGet():
    Signaller q = null;
    boolean queued = false;
    int spins = -1;
    Object r;
    // 当返回任务不为空,循环结束
    while ((r = result) == null) {
        if (spins < 0)
            spins = (Runtime.getRuntime().availableProcessors() > 1) ?
                1 << 8 : 0; // Use brief spin-wait on multiprocessors
        else if (spins > 0) {
            if (ThreadLocalRandom.nextSecondarySeed() >= 0)
                --spins;
        }
        else if (q == null)
            // 实例化一个信号量 --1
            q = new Signaller(interruptible, 0L, 0L);
        else if (!queued)
            queued = tryPushStack(q);
        else if (interruptible && q.interruptControl < 0) {
            q.thread = null;
            cleanStack();
            return null;
        }
        else if (q.thread != null && result == null) {
            try {
                // 若迟迟没有返回结果,最终会走到这个方法中,下面是ForkJoinPool对信号量的管理
                ForkJoinPool.managedBlock(q);
            } catch (InterruptedException ie) {
                q.interruptControl = -1;
            }
        }
    }
ForkJoinPool->managedBlock():
    Thread t = Thread.currentThread();
    if ((t instanceof ForkJoinWorkerThread) &&
        (p = (wt = (ForkJoinWorkerThread)t).pool) != null) {
        WorkQueue w = wt.workQueue;
        while (!blocker.isReleasable()) {
            // 
            if (p.tryCompensate(w)) {   // --2
                try {
                    do {} while (!blocker.isReleasable() &&
                                 !blocker.block());
                } finally {
                    U.getAndAddLong(p, CTL, AC_UNIT);
                }
                break;
            }
        }
    }
    else {
        do {} while (!blocker.isReleasable() &&
                     !blocker.block());
    }
ForkJoinPool->tryCompensate():  // --2
     canBlock = add && createWorker(); // throws on exception
复制代码
  • 获取当前线程,判断其类型,如果当前线程是非forkJoin线程的话,那么走else方法,直到返回结果为止;
  • 如果是forkJoin线程的话,当执行2处的代码时,还会进行一系列复杂的判断,若仍然迟迟得不到返回结果,会新建一个线程,帮助执行线程池里的任务。多出来的那几百个线程确实出自于此;
CompletableFuture->Signaller->Signaller():  // --1
    Signaller(boolean interruptible, long nanos, long deadline) {
        // thread变量是当前线程
         this.thread = Thread.currentThread();
         this.interruptControl = interruptible ? 1 : 0;
         this.nanos = nanos;
         this.deadline = deadline;
    }
复制代码

反思

到第一个voidCompletableFuture.join(),该线程是http线程,由forkJoinPool线程池管理,最多10个线程并行,然后到waitingGet(),由于其不是forkJoin线程,因此走的是else方法

到第二个voidCompletableFuture.join(),该线程是forkJoinPool执行的任务,每一个任务都会执行一次getAssociatedInfo方法,由executorService线程池管理,最多20个线程并行,然后到waitingGet(),由于它是forkJoin线程,所以会新建一个线程,帮助执行forkJoinPool线程池里的任务,然而受到executorService线程池数量的制约,即使线程数多了,也不能加快执行,随着越来越多getAssociatedInfo方法的Join,导致了线程数量的飙升,又不能即时释放,最终导致了OOM的发生

解决方案

猜想:将http线程的任务与forkJoinPool线程池的任务放在同一线程池,这样每当forkJoinPool线程池新产生一个线程时,都能窃取到任务从而执行,并且随着线程数量的上升,越来越多的任务被执行,这样就减少了线程创建的数量。最终的结果果然如此

Java中关于CompletableFuture类的思考

原文 

https://juejin.im/post/5eef3ca9f265da02bc6b2d88

本站部分文章源于互联网,本着传播知识、有益学习和研究的目的进行的转载,为网友免费提供。如有著作权人或出版方提出异议,本站将立即删除。如果您对文章转载有任何疑问请告之我们,以便我们及时纠正。

PS:推荐一个微信公众号: askHarries 或者qq群:474807195,里面会分享一些资深架构师录制的视频录像:有Spring,MyBatis,Netty源码分析,高并发、高性能、分布式、微服务架构的原理,JVM性能优化这些成为架构师必备的知识体系。还能领取免费的学习资源,目前受益良多

转载请注明原文出处:Harries Blog™ » Java中关于CompletableFuture类的思考

赞 (0)
分享到:更多 ()

评论 0

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址