为什么 Java 17 在将任务添加到 ForkJoinPool 时会抛出 RejectedExecutionException？

Question

我使用 Java 16 通过 HTTP 向 API 发出请求。为了整体加快速度，我将其加载到自定义ForkJoinPool. 我在下面编译了一个重现示例。

自从迁移到 Java 17（openjdk build 17.0.1+12-39）后，这会抛出 RejectedExecutionException：

Caused by: java.util.concurrent.RejectedExecutionException: Thread limit exceeded replacing blocked worker
    at java.base/java.util.concurrent.ForkJoinPool.tryCompensate(ForkJoinPool.java:1819)
    at java.base/java.util.concurrent.ForkJoinPool.compensatedBlock(ForkJoinPool.java:3446)
    at java.base/java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool.java:3432)
    at java.base/java.util.concurrent.CompletableFuture.waitingGet(CompletableFuture.java:1898)
    at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2072)
    at java.net.http/jdk.internal.net.http.HttpClientImpl.send(HttpClientImpl.java:553)
    at java.net.http/jdk.internal.net.http.HttpClientFacade.send(HttpClientFacade.java:119)
    at Test.lambda$retrieveMany$1(Test.java:30)

为什么会出现这种情况？ForkJoinPool 是否发生了我不知道的变化？

代码

Caused by: java.util.concurrent.RejectedExecutionException: Thread limit exceeded replacing blocked worker
    at java.base/java.util.concurrent.ForkJoinPool.tryCompensate(ForkJoinPool.java:1819)
    at java.base/java.util.concurrent.ForkJoinPool.compensatedBlock(ForkJoinPool.java:3446)
    at java.base/java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool.java:3432)
    at java.base/java.util.concurrent.CompletableFuture.waitingGet(CompletableFuture.java:1898)
    at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2072)
    at java.net.http/jdk.internal.net.http.HttpClientImpl.send(HttpClientImpl.java:553)
    at java.net.http/jdk.internal.net.http.HttpClientFacade.send(HttpClientFacade.java:119)
    at Test.lambda$retrieveMany$1(Test.java:30)

Answer 1

您已提交一项任务，但该任务parallelStream()在内部使用，然后在同一 fork join 池的不同线程上运行每个 http。

JDK16 和 17 处理池中所有可用线程都在使用的情况的方式有所不同 - 这就是参数saturated变得相关的地方。

当threads > urls.size()池永远不会饱和时，但在第二种情况下threads == urls.size()，所有线程都在使用中。null在 的构造函数中替换ForkJoinPool为saturate变量以查看何时触发饱和测试条件：

Predicate<? super ForkJoinPool> saturate = pool -> {
    boolean allow = false;
    System.out.println(Thread.currentThread().getName()+" saturate:"+allow);
    return allow;
};

在 JDK16 上，saturate谓词被调用多次但会继续，而在 JDK17 上，如果返回 false，则处理会在第一次调用时停止。如果您进行切换，那么当正在进行的请求数量与 正在使用的线程数量相同时， allow = trueJDK17 将不会发送，并且当其他请求完成时将继续处理进一步的请求。RejectedExecutionExceptionparallelStream()