理解ParallelStream的并发机制
Java 8引入的ParallelStream极大地简化了并行处理集合的操作。默认情况下，ParallelStream利用ForkJoinPool.commonPool()来执行任务。这个公共线程池的大小通常由系统处理器核心数决定，具体可以通过java.util.concurrent.ForkJoinPool.common.parallelism系统属性进行配置。然而，这种全局配置对于特定的应用场景，尤其是当并行任务涉及大量阻塞I/O操作（如数据库查询、网络请求）时，可能并不理想。
当ParallelStream中的任务执行阻塞I/O操作时，例如在peek或map阶段调用一个会等待数据库响应的方法，执行该任务的线程就会被阻塞。如果commonPool中的所有线程都被阻塞，即使系统还有其他可用的CPU资源，并行流也无法继续处理新任务，可能导致性能下降甚至死锁。因此，在这些场景下，精确控制ParallelStream的线程数量变得尤为重要。
定制ParallelStream的线程池
虽然java.util.concurrent.ForkJoinPool.common.parallelism属性可以调整公共线程池的大小，但它是一个全局设置，会影响所有使用commonPool()的并行任务。对于需要独立控制特定ParallelStream并发度的场景，更推荐的方法是为该并行流操作创建一个独立的ForkJoinPool。
这种方法的核心思想是将并行流的操作封装在一个Callable任务中，然后将这个Callable提交给一个自定义的ForkJoinPool。这样，并行流内部的线程就会从这个自定义的线程池中获取，而不是默认的commonPool()。
立即学习“Java免费学习笔记（深入）”；
以下是一个示例代码，演示如何为包含数据库查询（模拟）的ParallelStream设置自定义线程池：
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;
import java.util.stream.Collectors;
public class CustomParallelStreamPool {
// 模拟一个对象服务，用于获取参数
static class ObjectService {
public String getParam(String field) {
// 模拟数据库查询耗时
try {
System.out.println(Thread.currentThread().getName() + " - Querying for field: " + field);
Thread.sleep(200); // 模拟I/O阻塞
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException(e);
}
return "Param for " + field;
}
}
// 模拟原始的doSomething方法，使用CompletableFuture和外部Executor
private String doSomething(String objectField, ExecutorService asyncExecutor, ObjectService objectService) {
// 注意：这里为了简化，直接在doSomething中等待CompletableFuture完成。
// 实际应用中，如果doSomething返回CompletableFuture，
// 并且流操作是异步的（如flatMap(obj -> asyncOperation(obj).toStream())），
// 则流线程不会阻塞。但如果流操作直接调用并等待结果，则会阻塞。
try {
return CompletableFuture.supplyAsync(() -> objectService.getParam(objectField), asyncExecutor)
.thenApply(param -> "Processed(" + param + ")")
.get(); // 阻塞等待CompletableFuture完成
} catch (InterruptedException | ExecutionException e) {
Thread.currentThread().interrupt();
throw new RuntimeException("Error processing object: " + objectField, e);
}
}
// 封装并行处理逻辑
public List<String> processParallel(List<String> objects, ExecutorService asyncExecutor, ObjectService objectService) {
return objects.parallelStream()
.map(object -> doSomething(object, asyncExecutor, objectService))
.collect(Collectors.toList());
}
public static void main(String[] args) throws InterruptedException, ExecutionException {
List<String> data = List.of("ItemA", "ItemB", "ItemC", "ItemD", "ItemE", "ItemF", "ItemG", "ItemH", "ItemI", "ItemJ");
int desiredParallelism = 3; // 期望的并行度
// 用于CompletableFuture的异步执行器（模拟数据库连接池）
ExecutorService asyncDbExecutor = Executors.newFixedThreadPool(desiredParallelism);
ObjectService objectService = new ObjectService();
// 创建一个自定义的ForkJoinPool
ForkJoinPool customPool = new ForkJoinPool(desiredParallelism);
try {
System.out.println("Starting parallel processing with custom ForkJoinPool (parallelism: " + desiredParallelism + ")");
long startTime = System.currentTimeMillis();
// 将并行流操作提交到自定义线程池
Callable<List<String>> task = () ->
new CustomParallelStreamPool().processParallel(data, asyncDbExecutor, objectService);
List<String> results = customPool.submit(task).get();
long endTime = System.currentTimeMillis();
System.out.println("Finished processing in " + (endTime - startTime) + " ms.");
System.out.println("Results: " + results);
} finally {
customPool.shutdown(); // 务必关闭自定义线程池
asyncDbExecutor.shutdown(); // 关闭异步执行器
if (!customPool.awaitTermination(5, java.util.concurrent.TimeUnit.SECONDS)) {
System.err.println("Custom ForkJoinPool did not terminate in time.");
}
if (!asyncDbExecutor.awaitTermination(5, java.util.concurrent.TimeUnit.SECONDS)) {
System.err.println("Async DB Executor did not terminate in time.");
}
}
}
}在上述代码中，我们创建了一个ForkJoinPool实例，其并行度被设置为desiredParallelism。然后，我们将processParallel方法（包含parallelStream操作）封装在一个Callable中，并提交给这个customPool。这样，processParallel内部使用的并行流就会从customPool中获取线程，从而实现了对特定并行流操作的线程数量限制。
注意事项：
- 资源管理：自定义的ForkJoinPool在使用完毕后必须调用shutdown()方法进行关闭，以释放资源。
- 实现细节：这种方法依赖于Stream API的内部实现细节，即当并行流在一个ForkJoinTask（Callable会被包装成ForkJoinTask）中运行时，它会尝试使用该任务所在的ForkJoinPool。虽然目前稳定，但未来API更新可能带来兼容性问题。
I/O密集型任务的深层考量
对于像数据库查询这样的I/O密集型任务，仅仅限制ParallelStream的线程数量可能不足以解决所有问题，甚至可能引入新的挑战。
- 数据库连接限制：每个数据库查询都需要一个数据库连接。如果你的ParallelStream线程数量（无论是commonPool还是自定义ForkJoinPool）超过了数据库连接池所能提供的最大连接数，那么即使有空闲的线程，它们也会因为等待连接而阻塞。这可能导致死锁、性能瓶颈或连接耗尽。在这种情况下，并行度应该与可用的数据库连接数相匹配，而不是简单地基于CPU核心数。
- 阻塞与非阻塞：如果doSomething方法内部的CompletableFuture是真正异步且非阻塞的（即doSomething立即返回一个CompletableFuture，而不是.get()等待结果），并且ParallelStream能够以非阻塞的方式处理这些CompletableFuture（例如，通过某种flatMap操作将CompletableFuture转换为流，或使用响应式编程），那么ParallelStream的线程不会长时间阻塞。然而，如果doSomething内部像示例中那样调用了CompletableFuture.get()，那么ParallelStream的线程依然会被阻塞。
- 复杂性：在复杂的微服务或Web应用中，多个并发请求可能同时触发这些异步任务。手动管理ParallelStream的线程池和数据库连接池之间的关系，以及处理潜在的资源竞争和死锁，会变得非常复杂且容易出错。
推荐的替代方案
对于高并发、I/O密集型且需要精细资源控制的场景，以下方案可能更为合适：
-
响应式编程框架：
- Spring WebFlux：基于Project Reactor，提供非阻塞的、事件驱动的Web栈。它通过少量线程处理大量并发请求，非常适合I/O密集型应用，能够有效利用数据库连接等资源。
- Quarkus/Micronaut：这些现代Java框架也提供了对响应式编程和非阻塞I/O的良好支持。
- Vert.x：一个事件驱动的、非阻塞的工具包，专为构建高性能、响应式应用而设计。 这些框架通过异步非阻塞I/O模型，将线程阻塞降到最低，从而能够以更少的线程处理更高的并发量。
-
自定义线程池与CompletableFuture的结合： 如果不想引入完整的响应式框架，但需要更好的控制，可以继续使用CompletableFuture，但要确保其背后的Executor是精心配置的，并且与数据库连接池的容量相匹配。
- 将ParallelStream替换为普通的Stream，然后手动使用CompletableFuture.supplyAsync提交任务到你自己的ExecutorService（例如，一个固定大小的线程池，其大小与数据库连接数一致）。
- 收集所有CompletableFuture，然后使用CompletableFuture.allOf().join()或CompletableFuture.join()等待所有任务完成。
import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.stream.Collectors; // ... (ObjectService and doSomething method from previous example) ... public class CustomExecutorWithCompletableFuture { private String doSomethingAsync(String objectField, ExecutorService asyncExecutor, ObjectService objectService) { // 返回CompletableFuture，不在此处阻塞 return CompletableFuture.supplyAsync(() -> objectService.getParam(objectField), asyncExecutor) .thenApply(param -> "Processed(" + param + ")") .join(); // 简化，实际可能在收集后统一join } public static void main(String[] args) { List<String> data = List.of("ItemA", "ItemB", "ItemC", "ItemD", "ItemE", "ItemF", "ItemG", "ItemH", "ItemI", "ItemJ"); int dbConnectionLimit = 3; // 假设数据库连接限制 // 创建一个固定大小的线程池，用于执行I/O密集型任务 ExecutorService dbQueryExecutor = Executors.newFixedThreadPool(dbConnectionLimit); ObjectService objectService = new ObjectService(); try { System.out.println("Starting processing with custom Executor and CompletableFuture (DB connections: " + dbConnectionLimit + ")"); long startTime = System.currentTimeMillis(); List<CompletableFuture<String>> futures = data.stream() .map(item -> CompletableFuture.supplyAsync(() -> new CustomParallelStreamPool().doSomething(item, dbQueryExecutor, objectService), dbQueryExecutor)) .collect(Collectors.toList()); // 等待所有CompletableFuture完成并获取结果 List<String> results = futures.stream() .map(CompletableFuture::join) // 阻塞等待每个future完成 .collect(Collectors.toList()); long endTime = System.currentTimeMillis(); System.out.println("Finished processing in " + (endTime - startTime) + " ms."); System.out.println("Results: " + results); } finally { dbQueryExecutor.shutdown(); if (!dbQueryExecutor.awaitTermination(5, java.util.concurrent.TimeUnit.SECONDS)) { System.err.println("DB Query Executor did not terminate in time."); } } } }这种方式将并行度控制权完全交给自定义的dbQueryExecutor，并且可以更好地与数据库连接池进行协调。
总结
控制ParallelStream的线程池大小是优化其性能的关键，尤其是在处理I/O密集型任务时。通过为特定操作创建自定义的ForkJoinPool，可以有效地限制并发度。然而，对于涉及外部资源（如数据库连接）的场景，更深层次的考量是必要的。在这种情况下，将并行度与可用资源相匹配，并考虑采用响应式编程框架或更精细的CompletableFuture与自定义ExecutorService结合的方案，往往能提供更健壮、高效且可扩展的解决方案。在选择方法时，务必权衡其复杂性、维护成本以及对应用整体架构的影响。
