java - ForkJoinPool 性能 Java 8 对比 11

Question

考虑以下代码:

package com.sarvagya;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.stream.Collectors;

public class Streamer {
    private static final int LOOP_COUNT  = 2000;
    public static void main(String[] args){
        try{
            for(int i = 0; i < LOOP_COUNT; ++i){
                poolRunner();
                System.out.println("done loop " + i);
                try{
                    Thread.sleep(50L);
                }
                catch (Exception e){
                    System.out.println(e);
                }
            }
        }
        catch (ExecutionException | InterruptedException e){
            System.out.println(e);
        }

        // Add a delay outside the loop to make sure all daemon threads are cleared before main exits.
        try{
            Thread.sleep(10 * 60 * 1000L);
        }
        catch (Exception e){
            System.out.println(e);
        }
    }

    /**
     * poolRunner method.
     * Assume I don't have any control over this method e.g. done by some library.
     * @throws InterruptedException
     * @throws ExecutionException
     */
    private static void poolRunner() throws InterruptedException, ExecutionException {
        ForkJoinPool pool = new ForkJoinPool();
        pool.submit(() ->{
            List<Integer> numbers = Arrays.asList(1,2,3,4,5,6,7,8,9,10, 11,12,14,15,16);
            List<Integer> collect = numbers.stream()
                    .parallel()
                    .filter(xx -> xx > 5)
                    .collect(Collectors.toList());
            System.out.println(collect);
        }).get();
    }
}

在上面的代码中，poolRunner 方法正在创建一个 ForkJoinPool 并向它提交一些任务。当使用 Java 8 并将 LOOP_COUNT 保持为 2000 时，我们可以看到创建的最大线程数约为 3600，如下所示 图:剖析

图:线程信息。

一段时间后，所有这些线程都下降到将近 10 个。但是，在 OpenJDK 11 中保持相同的 LOOP_COUNT 将产生以下错误:

[28.822s][warning][os,thread] Failed to start thread - pthread_create failed (EAGAIN) for attributes: stacksize: 1024k, guardsize: 4k, detached.
[28.822s][warning][os,thread] Failed to start thread - pthread_create failed (EAGAIN) for attributes: stacksize: 1024k, guardsize: 4k, detached.
[28.822s][warning][os,thread] Failed to start thread - pthread_create failed (EAGAIN) for attributes: stacksize: 1024k, guardsize: 4k, detached.
Exception in thread "ForkJoinPool-509-worker-5" java.lang.OutOfMemoryError: unable to create native thread: possibly out of memory or process/resource limits reached
    at java.base/java.lang.Thread.start0(Native Method)
    at java.base/java.lang.Thread.start(Thread.java:803)
    at java.base/java.util.concurrent.ForkJoinPool.createWorker(ForkJoinPool.java:1329)
    at java.base/java.util.concurrent.ForkJoinPool.tryAddWorker(ForkJoinPool.java:1352)
    at java.base/java.util.concurrent.ForkJoinPool.signalWork(ForkJoinPool.java:1476)
    at java.base/java.util.concurrent.ForkJoinPool.deregisterWorker(ForkJoinPool.java:1458)
    at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:187)

它很快达到最大线程限制。将 LOOP_COUNT 保持在 500，工作正常，但是，这些线程被清除得非常非常慢，并达到大约 500 个线程的稳定状态。见下图:

图:OpenJDK 11 中的线程信息

图:在 OpenJDK 11 中进行分析

线程在 JDK 8 中是PARKED，但在 JDK 11 中是WAIT。Java 11 中守护线程的数量也应该减少，但是，它很慢而且不会'按预期工作。此外，假设我无法控制 poolRunner 方法。考虑这个方法是由一些外部库提供的。

这个问题是 OpenJDK 11 还是我在代码中做错了什么。谢谢。

Answer 1

您的代码正在创建大量 ForkJoinPool实例并且从不调用 shutdown()使用后在任何游泳池上。由于在 Java 8 的情况下，规范中没有任何内容保证工作线程会终止，因此这段代码甚至可能以 2000(⟨池数⟩)次 ⟨核心数⟩ 线程。

在实践中，观察到的行为源于 undocumented idle timeout两秒钟。请注意，根据评论，超时的结果是尝试减少工作人员的数量，这与终止不同。因此，如果 n 线程遇到超时，并非所有 n 线程都终止，但线程数会减少一个，其余线程可能会再次等待。此外，短语“初始超时值”已经暗示了这一点，每次发生时实际超时都会增加。所以需要 <em>n</em> * (<em>n</em> + 1) n 空闲工作线程由于此(未记录的)超时而终止的秒数。

从 Java 9 开始，有一个可配置的 keepAliveTime，可以在 new constructor 中指定。的 ForkJoinPool ，它还记录了默认值:

keepAliveTime

the elapsed time since last use before a thread is terminated (and then later replaced if needed). For the default value, use 60, TimeUnit.SECONDS.

该文档可能会误导人们认为现在所有工作线程在空闲 keepAliveTime 时可能会一起终止，但实际上，仍然存在一次只收缩一个池的行为，尽管现在，时间没有增加。所以现在，它最多需要 60 * <em>n</em> n 个空闲工作线程终止的秒数。由于之前的行为未指定，因此它甚至不是不兼容。

必须强调的是，即使超时行为相同，最终的最大线程数也可能会发生变化，因为当具有更好代码优化的较新 JVM 减少实际操作的执行时间时(无需人工插入 Thread.sleep(…))它会更快地创建新线程，而终止仍然受制于挂钟时间。

---

要点是，当您知道不再需要线程池时，您永远不应该依赖自动工作线程终止。相反，您应该调用 shutdown()完成后。

---

您可以使用以下代码验证行为:

int threadNumber = 8;
ForkJoinPool pool = new ForkJoinPool(threadNumber);
// force the creation of all worker threads
pool.invokeAll(Collections.nCopies(threadNumber*2, () -> { Thread.sleep(500); return ""; }));
int oldNum = pool.getPoolSize();
System.out.println(oldNum+" threads; waiting for dying threads");
long t0 = System.nanoTime();
while(oldNum > 0) {
    while(pool.getPoolSize()==oldNum)
        LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(200));
    long t1 = System.nanoTime();
    oldNum = pool.getPoolSize();
    System.out.println(threadNumber-oldNum+" threads terminated after "
        +TimeUnit.NANOSECONDS.toSeconds(t1 - t0)+"s");
}

Java 8:

8 threads; waiting for dying threads
1 threads terminated after 2s
2 threads terminated after 6s
3 threads terminated after 12s
4 threads terminated after 20s
5 threads terminated after 30s
6 threads terminated after 42s
7 threads terminated after 56s
8 threads terminated after 72s

Java 11:

8 threads; waiting for dying threads
1 threads terminated after 60s
2 threads terminated after 120s
3 threads terminated after 180s
4 threads terminated after 240s
5 threads terminated after 300s
6 threads terminated after 360s
7 threads terminated after 420s

^{从未完成，显然，至少最后一个工作线程还活着}