为什么Java 17抛出一个RejectedExecutionException添加任务到ForkJoinPool?&l

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有什么变化，我不知道吗?

import java.io.IOException;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse.BodyHandlers;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import static java.util.concurrent.TimeUnit.MINUTES;
import static java.util.stream.Collectors.toList;
public class Test {
public static void main(String[] args) throws ExecutionException, InterruptedException {
final List<String> urls = List.of("https://stackoverflow.com", "https://stackoverflow.com", "https://stackoverflow.com");
// This succeeds on JDK 16, 17
retrieveMany(urls, 4);
// This fails on JDK 17, but succeeds on 16
retrieveMany(urls, 3);
}
private static List<String> retrieveMany(List<String> urls, int threads) throws InterruptedException, ExecutionException {
return new ForkJoinPool(threads, ForkJoinPool.defaultForkJoinWorkerThreadFactory, (t, e) -> {}, true, 0, threads, 1, null, 1, MINUTES)
.submit(() -> urls.parallelStream()
.map(url -> {
try {
return HttpClient.newBuilder().build().send(HttpRequest.newBuilder(URI.create(url)).build(), BodyHandlers.ofString()).body();
} catch (IOException | InterruptedException aE) { }
return null;
})
.collect(toList()))
.get();
}
}

Answer 1

您已经提交了一个任务，但它在内部使用parallelStream()，然后在同一分支连接池的不同线程上运行每个http。

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

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

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

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