本文介绍如何使用 executorservice 并行执行大量 runnable 任务，同时为每个任务单独设置 5 秒超时阈值，并在超时时主动中断、记录日志，避免阻塞线程池和主线程。
本文介绍如何使用 executorservice 并行执行大量 runnable 任务，同时为每个任务单独设置 5 秒超时阈值，并在超时时主动中断、记录日志，避免阻塞线程池和主线程。
在高并发场景中，仅靠 invokeAll(tasks, timeout, unit) 无法满足「每个任务独立超时」的需求——它限制的是整个批处理的总等待时间，而非单个任务的执行上限。正确做法是：为每个任务绑定专属超时监控机制，借助 ScheduledExecutorService 异步触发取消逻辑，确保线程资源高效复用。
以下是推荐的生产级实现方案（已适配原问题中的 100 个随机耗时任务）：
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
public class ParallelTimeoutExecutor {
private static final int CORE_POOL_SIZE = Runtime.getRuntime().availableProcessors();
private static final int TIMEOUT_SECONDS = 5;
public static void main(String[] args) throws InterruptedException {
ExecutorService workerPool = Executors.newFixedThreadPool(CORE_POOL_SIZE);
ScheduledExecutorService timeoutScheduler = Executors.newSingleThreadScheduledExecutor();
List<Runnable> tasks = new ArrayList<>();
AtomicInteger taskIdGenerator = new AtomicInteger(0);
// 构建 100 个带唯一 ID 的 Runnable 任务
for (int i = 0; i < 100; i++) {
int taskId = taskIdGenerator.getAndIncrement();
tasks.add(() -> executeTask(taskId));
}
// 提交任务并注册超时监控
List<Future<Long>> futures = new ArrayList<>();
for (int i = 0; i < tasks.size(); i++) {
final int taskId = i;
Runnable task = tasks.get(i);
Future<Long> future = workerPool.submit(() -> {
long start = System.nanoTime();
try {
task.run();
} finally {
long durationMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
System.out.printf("[✓] Task #%d completed in %.2f ms%n", taskId, (double) durationMs);
}
return durationMs;
});
// 5 秒后检查：若未完成，则强制中断
timeoutScheduler.schedule(() -> {
if (!future.isDone()) {
boolean wasInterrupted = future.cancel(true); // 中断执行线程
System.err.printf("[⚠] Task #%d TIMED OUT after %d seconds and was %s%n",
taskId, TIMEOUT_SECONDS, wasInterrupted ? "cancelled" : "already done");
}
}, TIMEOUT_SECONDS, TimeUnit.SECONDS);
futures.add(future);
}
// 主线程非阻塞等待所有结果（可选：加超时防止无限挂起）
for (Future<Long> future : futures) {
try {
future.get(6, TimeUnit.SECONDS); // 留 1 秒缓冲，避免 get() 阻塞过久
} catch (ExecutionException e) {
// 任务抛异常（如被中断后的 InterruptedException）
Throwable cause = e.getCause();
if (cause instanceof InterruptedException) {
System.err.println("[!] Task interrupted during execution.");
}
} catch (TimeoutException e) {
// 此处不应触发（因已由 scheduler cancel），仅作兜底
System.err.println("[!] Future.get() itself timed out — unexpected.");
}
}
// 安全关闭
workerPool.shutdown();
timeoutScheduler.shutdown();
if (!workerPool.awaitTermination(10, TimeUnit.SECONDS)) {
workerPool.shutdownNow();
}
if (!timeoutScheduler.awaitTermination(3, TimeUnit.SECONDS)) {
timeoutScheduler.shutdownNow();
}
}
// 模拟真实业务任务：支持响应中断
private static void executeTask(int taskId) {
int sleepTime = 1 + (int) (Math.random() * 8); // 1–8 秒
System.out.printf("[→] Task #%d started, expected to run %d seconds%n", taskId, sleepTime);
try {
Thread.sleep(sleepTime * 1000);
} catch (InterruptedException e) {
System.out.printf("[×] Task #%d was interrupted (likely due to timeout)%n", taskId);
Thread.currentThread().interrupt(); // 恢复中断状态，符合最佳实践
return;
}
}
}✅ 关键设计说明：
- 独立超时控制：每个 Future 绑定一个 ScheduledExecutorService 的延迟任务，精准实现「每任务 5 秒」超时，而非全局限制；
- 安全中断：调用 future.cancel(true) 会向执行线程发送中断信号，要求任务自身配合检查 Thread.interrupted() 或捕获 InterruptedException；
- 日志可追溯：输出含任务 ID、状态（完成/超时/中断）、耗时等信息，便于故障排查与性能分析；
- 资源防护：ScheduledExecutorService 使用单线程，避免调度开销；主线程 get() 加了短超时，杜绝永久阻塞；
- 优雅关闭：显式调用 shutdown() + awaitTermination() + shutdownNow() 三重保障，防止 JVM 残留线程。
⚠️ 重要注意事项：
立即学习“Java免费学习笔记（深入）”；
- 若任务内部不响应中断（如死循环且无 Thread.interrupted() 检查），cancel(true) 无法真正终止其执行，仅能标记 future.isCancelled() 为 true。因此，务必在长期运行逻辑中定期检测中断状态；
- 不建议复用 Executors.newFixedThreadPool() 返回的默认线程池用于定时调度——其 corePoolSize 可能过大，应单独创建轻量级 ScheduledExecutorService；
- 在 Spring 等容器环境中，应通过 @PreDestroy 或 DisposableBean 接口统一管理线程池生命周期，避免内存泄漏。
该方案兼顾并发效率、可观测性与健壮性，适用于微服务异步编排、批量数据校验、第三方 API 熔断等典型场景。
