java - 使用了Java CompletableFuture线程池

Question

在下面的代码中，无论我将i的最大值设置为多少，线程总数都不会超过13。它使用什么线程池？在哪里可以找到其默认设置？

public static void main(String[] args) {
    // write your code here
    for (int i = 0; i <= 5; i++) {

        System.out.println("kick off" + i);
        CompletableFuture.runAsync(() -> {
            try {
                Thread.sleep(1000);

                System.out.println(java.lang.Thread.activeCount());
            }
            catch (Exception e) {
                System.out.println("error");
            }
        });

    }
    System.out.println(java.lang.Thread.activeCount());
    try {
        Thread.sleep(10000);
    }
    catch (InterruptedException e) {
        e.printStackTrace();
    }
}

Answer 1

回答
它是由您的系统设置或当前的处理器数量决定的。

文档和代码
从CompletableFuture的文档中:

All async methods without an explicit Executor argument are performed using the ForkJoinPool.commonPool() (unless it does not support a parallelism level of at least two, in which case, a new Thread is created to run each task). This may be overridden for non-static methods in subclasses by defining method defaultExecutor(). [...]

从ForkJoinPool#commonPool()的文档中:

Returns the common pool instance. This pool is statically constructed; [...]

从ForkJoinPool类本身的文档中:

The parameters used to construct the common pool may be controlled by setting the following system properties:

• java.util.concurrent.ForkJoinPool.common.parallelism - the parallelism level, a non-negative integer
• java.util.concurrent.ForkJoinPool.common.threadFactory - the class name of a ForkJoinPool.ForkJoinWorkerThreadFactory. The system class loader is used to load this class.
• java.util.concurrent.ForkJoinPool.common.exceptionHandler - the class name of a Thread.UncaughtExceptionHandler. The system class loader is used to load this class.
• java.util.concurrent.ForkJoinPool.common.maximumSpares - the maximum number of allowed extra threads to maintain target parallelism (default 256).

If no thread factory is supplied via a system property, then the common pool uses a factory that uses the system class loader as the thread context class loader.

从那里开始，我们必须检查实际的源代码。从ForkJoinPool.java#L3208:

common = AccessController.doPrivileged(new PrivilegedAction<>() {
    public ForkJoinPool run() {
        return new ForkJoinPool((byte)0); }});

来自ForkJoinPool.java#L2345的构造函数:

// [...]
String pp = System.getProperty("java.util.concurrent.ForkJoinPool.common.parallelism");
// [...]
if (pp != null)
    parallelism = Integer.parseInt(pp);
// [...]
if (parallelism < 0 && // default 1 less than #cores
    (parallelism = Runtime.getRuntime().availableProcessors() - 1) <= 0)
        parallelism = 1;
// [...]
int n = (parallelism > 1) ? parallelism - 1 : 1;
n |= n >>> 1; n |= n >>> 2; n |= n >>> 4; n |= n >>> 8; n |= n >>> 16;
n = (n + 1) << 1;
// [...]
this.workQueues = new WorkQueue[n];

然后你去。它由系统设置或当前处理器数量决定。

数学示例
假设您没有为java.util.concurrent.ForkJoinPool.common.parallelism设置任何内容，让我们快速为该代码做数学运算:
在这种情况下，parallelism以-1开头，所以我们有

if (parallelism < 0 && // default 1 less than #cores
    (parallelism = Runtime.getRuntime().availableProcessors() - 1) <= 0)
        parallelism = 1;

让我们假设您的计算机上有8个内核。因此，您执行parallelism = Runtime.getRuntime().availableProcessors() - 1，它将7分配给parallelism。您无需输入if，所以我们继续。
接下来我们有

int n = (parallelism > 1) ? parallelism - 1 : 1;

减去其中之一，即n = 6。
然后

n |= n >>> 1; n |= n >>> 2; n |= n >>> 4; n |= n >>> 8; n |= n >>> 16;
n = (n + 1) << 1;

结果为16。
这样做基本上是减去2，然后转到2的下一个幂，然后将其加倍。
因此，如果您具有6核心，则转到4，然后转到8，然后再加上2和16。

那为什么13？
那么，为什么会得到13呢？我想您有4或5处理器，因此该池将使用8线程。但是，您正在测量的是Java总共使用的线程总数。

System.out.println(java.lang.Thread.activeCount());

Java现在可能会在13 - 8 = 5线程周围使用其他“备用”线程。