java - 为什么 Java 17 在将任务添加到 ForkJoinPool 时会抛出 RejectedExecutionException?
问题描述
我使用 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();
}
}
解决方案
您已经提交了一项任务,但该任务在parallelStream()
内部使用,然后在同一个分叉连接池的不同线程上运行每个 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,则处理在第一次调用时停止。如果您切换,那么当正在进行的请求数与 JDK17 正在使用的线程数相同时,allow = true
JDK17 将不会发送,并且会在其他请求完成时继续处理进一步的请求。RejectedExecutionException
parallelStream()
推荐阅读
- jquery - CSRF 验证失败。请求中止。没有表格
- javascript - 在d3.js中如何改变树形图的节点大小
- eclipse - Eclipse 2020-06 中“所有”Red Hat 附加组件的“无法读取存储库”
- python - 将字符串中的十六进制字符转换为字符
- python - 使用网站访问受限的 Web API (raspberry pi)
- angular - Ng-Model 不从输入标签更新对象(在带有键值管道的 ngFor 内)
- node.js - 在 nodejs 中使用 ES 模块时 PM2 崩溃
- javascript - 使用 moment.js 进行实时计算
- c++ - 当getter需要将受保护的资源作为现代c ++中的shared_ptr返回时,如何设计一次写入和多次读取?
- graphql - GraphQL DynamoDb 过滤器列表字段