求助：CompletableFuture无法动态创建多线程，固定线程池致阻塞

嘿，我来帮你搞定这个CompletableFuture配合动态线程池的问题！你之前用固定大小的ExecutorService导致程序阻塞，核心原因是固定线程池的任务队列会积压未执行的任务，当队列满了或者线程都在忙时，新任务就会阻塞等待。咱们可以通过自定义动态可伸缩的线程池来解决这个问题，再结合CompletableFuture实现异步任务的动态调度。

第一步：创建动态可伸缩的线程池 #

首先我们需要一个能根据任务量动态增减线程的线程池，而不是固定大小的。这里用ThreadPoolExecutor自定义，核心思路是：

• 核心线程数设置为CPU核心数（合理利用硬件资源）
• 最大线程数设置为核心数的2倍（可根据你的业务压力调整，IO密集型任务可以设更高）
• 使用SynchronousQueue作为任务队列，它不会缓存任务，当核心线程忙时直接创建新线程
• 设置空闲线程存活时间，闲置线程超过时间会被回收，避免资源浪费

代码实现：

private static ExecutorService createDynamicThreadPool() {
    int corePoolSize = Runtime.getRuntime().availableProcessors();
    int maxPoolSize = corePoolSize * 2; 
    long keepAliveTime = 60L;
    TimeUnit unit = TimeUnit.SECONDS;
    
    // SynchronousQueue不缓存任务，直接提交给线程，触发线程扩容
    BlockingQueue<Runnable> workQueue = new SynchronousQueue<>();
    
    return new ThreadPoolExecutor(
        corePoolSize,
        maxPoolSize,
        keepAliveTime,
        unit,
        workQueue,
        Executors.defaultThreadFactory(),
        new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略：提交任务的线程自己执行，避免任务丢失
    );
}

第二步：改造CompletableFuture创建方法 #

接下来把你的createCompletableFuture方法改成使用这个动态线程池，同时处理多任务的异步执行与结果汇总：

import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

// 假设FileUploadMultiLocator是你的自定义类
private static CompletableFuture<Integer> createCompletableFuture(ByteArrayOutputStream baOS, int totalBytes, List<FileUploadMultiLocator> locators) {
    // 获取动态线程池（如果方法会被多次调用，建议把线程池做成单例，避免重复创建）
    ExecutorService dynamicExecutor = createDynamicThreadPool();
    
    // 将每个FileUploadMultiLocator的处理逻辑转为异步任务
    List<CompletableFuture<Integer>> taskFutures = locators.stream()
        .map(locator -> CompletableFuture.supplyAsync(() -> {
            try {
                // 这里替换成你的实际业务逻辑：处理locator，比如读取文件片段写入baOS
                int processedBytes = processLocator(locator, baOS);
                return processedBytes;
            } catch (Exception e) {
                // 把检查异常包装为CompletionException，方便后续异常处理
                throw new CompletionException("处理Locator失败", e);
            }
        }, dynamicExecutor))
        .collect(Collectors.toList());
    
    // 等待所有任务完成，汇总处理结果
    return CompletableFuture.allOf(taskFutures.toArray(new CompletableFuture[0]))
        .thenApply(v -> taskFutures.stream()
            .mapToInt(CompletableFuture::join)
            .sum()) // 汇总所有任务的处理字节数
        .whenComplete((totalProcessed, throwable) -> {
            // 任务全部完成后关闭线程池，避免线程泄露
            dynamicExecutor.shutdown();
            try {
                // 等待线程池优雅关闭，超时则强制终止
                if (!dynamicExecutor.awaitTermination(60, TimeUnit.SECONDS)) {
                    dynamicExecutor.shutdownNow();
                }
            } catch (InterruptedException e) {
                dynamicExecutor.shutdownNow();
                Thread.currentThread().interrupt();
            }
        });
}

// 示例业务方法：处理单个FileUploadMultiLocator
private static int processLocator(FileUploadMultiLocator locator, ByteArrayOutputStream baOS) throws Exception {
    // 你的业务逻辑实现，比如读取文件内容写入baOS
    // ...
    return 0; // 返回实际处理的字节数
}

关键注意事项 #

• 线程池复用：如果createCompletableFuture方法会被频繁调用，建议把dynamicExecutor做成静态单例，不要每次创建新线程池，减少资源开销。
• 拒绝策略调整：示例中用了CallerRunsPolicy，如果你的业务对任务丢失敏感可以用这个；如果是实时性要求高的任务，可以换成AbortPolicy（抛出异常）或者自定义拒绝策略。
• 异常处理：一定要把业务逻辑中的检查异常包装为CompletionException，这样后续可以通过exceptionally或者whenComplete捕获处理。
• 线程池关闭：必须在任务完成后关闭线程池，否则会导致线程泄露，占用系统资源。

内容的提问来源于stack exchange，提问作者Harisingh Rajput