如何让ThreadPoolTaskExecutor在核心线程空闲时立即处理任务队列并在主线程结束前完成全部任务

首先，咱们拆解下你遇到的核心问题：

1. 误用了CountDownLatch的计数逻辑（只设了5，对应核心线程数，但你实际有1000个任务）
2. 错误调用了latch.wait()（应该用latch.await()，wait()需要持有对象锁，否则会抛出IllegalMonitorStateException，这很可能是导致你看到队列任务延迟执行的关键原因）
3. 主线程仅等待了前5个任务完成，没有跟踪所有1000个任务的执行状态

核心解决方案思路 #

要让主线程等待所有任务（包括队列中的）执行完毕，你需要跟踪每一个提交的任务，确保所有任务执行完成后再继续主线程逻辑。下面给你三种可行的实现方式：

方式一：用CountDownLatch跟踪全部任务 #

这是最直接的方式，把计数设为任务总数，每个任务执行完都触发countDown：

// 初始化CountDownLatch为任务总数（map的size）
CountDownLatch latch = new CountDownLatch(map.size());
Collection<Future<?>> futures = new LinkedList<>();

for (Map.Entry<String, Boolean> entry : map.entrySet()) {
    FutureTask<?> task = new FutureTask<>(() -> {
        try {
            // 你的任务业务逻辑
            return entry;
        } finally {
            // 不管任务成功失败，都要countDown，避免主线程永久阻塞
            latch.countDown();
        }
    });
    executor.execute(task);
    futures.add(task);
}

// 等待所有任务完成，可添加超时时间避免永久阻塞
try {
    latch.await();
    // 可选：latch.await(10, TimeUnit.MINUTES);
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
    log.error("等待任务完成时被中断", e);
}

// 所有任务执行完毕，打印日志
log.info("ACTIVE COUNT : " + executor.getActiveCount());
log.info("SIZE of the QUEUE : " + executor.getThreadPoolExecutor().getQueue().size());
log.info("LATCH WAIT : " + latch.getCount()); // 此处值应为0

方式二：用ExecutorService的invokeAll方法 #

ThreadPoolTaskExecutor实现了ExecutorService接口，可直接用invokeAll方法一次性提交所有任务并等待全部完成：

// 把所有CustomTask转换成Callable集合
List<Callable<Object>> tasks = new ArrayList<>();
for (Map.Entry<String, Boolean> entry : map.entrySet()) {
    tasks.add(new CustomTask(entry));
}

try {
    // invokeAll会阻塞主线程，直到所有任务完成或抛出异常
    executor.invokeAll(tasks);
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
    log.error("等待任务完成时被中断", e);
}

// 所有任务完成，打印日志
log.info("ACTIVE COUNT : " + executor.getActiveCount());
log.info("SIZE of the QUEUE : " + executor.getThreadPoolExecutor().getQueue().size());

方式三：遍历所有Future对象调用get() #

你已经收集了所有FutureTask，可以逐个调用get()方法，主线程会等待每个任务完成：

Collection<Future<?>> futures = new LinkedList<>();

for (Map.Entry<String, Boolean> entry : map.entrySet()) {
    FutureTask<?> task = new FutureTask<>(new CustomTask(entry));
    executor.execute(task);
    futures.add(task);
}

// 遍历所有Future，等待每个任务完成
for (Future<?> future : futures) {
    try {
        future.get();
        // 可选：future.get(10, TimeUnit.SECONDS); 添加超时时间
    } catch (InterruptedException | ExecutionException e) {
        log.error("任务执行失败", e);
        // 可在此处添加异常处理逻辑，比如重试或告警
    }
}

// 所有任务完成，打印日志
log.info("ACTIVE COUNT : " + executor.getActiveCount());
log.info("SIZE of the QUEUE : " + executor.getThreadPoolExecutor().getQueue().size());

关键注意点 #

• 永远用latch.await()而非latch.wait()：wait()是Object类的方法，需要先获取对象锁（用synchronized(latch)包裹），否则会抛出异常；而await()是CountDownLatch专门用于等待计数归零的方法，无需额外加锁。
• 异常处理：务必处理InterruptedException，避免主线程意外终止或永久阻塞。
• 超时机制：在调用await()或get()时建议添加超时时间，防止因单个任务卡住导致主线程无限等待。

内容的提问来源于stack exchange，提问作者irshad.ahmad