本文介绍如何使用 executorservice 并行执行大量 runnable 任务，同时为每个任务单独设置 5 秒超时；超时时主动中断任务、记录日志，并确保线程池资源高效复用。
本文介绍如何使用 executorservice 并行执行大量 runnable 任务，同时为每个任务单独设置 5 秒超时；超时时主动中断任务、记录日志，并确保线程池资源高效复用。
在高并发批处理场景中（如 100 个异步作业），我们常需：
✅ 严格限制单个任务执行时长（如 ≤5 秒）
✅ 保证任务真正并行执行（受 CPU 核心数或线程池大小约束）
✅ 超时时立即中断该任务，释放线程，避免阻塞后续任务
✅ 不阻塞主线程，且能准确识别并日志化超时任务 ID
直接调用 Future.get(5, TimeUnit.SECONDS) 会阻塞当前线程，而 invokeAll(tasks, 5, SECONDS) 是对整个任务集合设全局超时，不符合“每个任务独立计时”的需求。正确解法是：为每个任务绑定专属超时调度器。
✅ 推荐方案：双 Executor 协作模型
- ExecutorService（固定线程池）：执行实际业务逻辑
- ScheduledExecutorService（单线程调度器）：为每个任务提交一个 5 秒后触发的取消动作
关键在于：每个任务封装为 Callable<Long>，返回真实耗时；同时由调度器监控其 Future.isDone() 状态，未完成则调用 future.cancel(true) 强制中断。
以下是完整、健壮、可直接运行的实现：
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
public class ParallelTimeoutRunner {
// 主执行器：按 CPU 核心数配置（示例为 4 线程）
private static final ExecutorService executor =
Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
// 超时调度器：单线程，避免调度竞争
private static final ScheduledExecutorService timeoutScheduler =
Executors.newSingleThreadScheduledExecutor();
public static void main(String[] args) throws InterruptedException {
List<Runnable> tasks = new ArrayList<>();
AtomicInteger timeoutCount = new AtomicInteger(0);
// 构造 100 个随机耗时任务（1–8 秒），带唯一 ID
for (int i = 0; i < 100; i++) {
final int taskId = i;
tasks.add(() -> {
try {
int sleepTime = 1 + (int) (Math.random() * 8); // 1–8 秒
System.out.printf("[TASK-%d] START → will sleep %d sec%n", taskId, sleepTime);
Thread.sleep(sleepTime * 1000);
System.out.printf("[TASK-%d] DONE in %d sec%n", taskId, sleepTime);
} catch (InterruptedException e) {
System.out.printf("[TASK-%d] INTERRUPTED (likely timeout)%n", taskId);
Thread.currentThread().interrupt(); // 恢复中断状态
}
});
}
// 提交所有任务，为每个任务绑定独立超时逻辑
List<Future<Long>> futures = new ArrayList<>();
for (int i = 0; i < tasks.size(); i++) {
final int taskId = i;
final Runnable task = tasks.get(i);
// 封装任务：执行 + 计时
Future<Long> future = executor.submit(() -> {
long start = System.nanoTime();
task.run();
return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
});
// 绑定超时调度：5 秒后检查是否完成，未完成则取消
timeoutScheduler.schedule(() -> {
if (!future.isDone()) {
boolean cancelled = future.cancel(true); // 中断正在运行的线程
if (cancelled) {
int count = timeoutCount.incrementAndGet();
System.err.printf("[TIMEOUT] TASK-%d forcibly cancelled after 5s (total timeouts: %d)%n",
taskId, count);
}
}
}, 5, TimeUnit.SECONDS);
futures.add(future);
}
// 非阻塞式结果收集（可选：等待全部完成）
System.out.println("→ All tasks submitted. Waiting for completion...");
for (Future<Long> future : futures) {
try {
Long duration = future.get(); // 此处 get 不会阻塞太久（因已超时取消）
if (duration > 5000) {
System.out.printf("[SLOW] Task completed in %d ms (exceeded 5s)%n", duration);
}
} catch (ExecutionException e) {
// 任务抛异常或被 cancel 时，get() 抛 ExecutionException（cause 是 CancellationException）
Throwable cause = e.getCause();
if (cause instanceof CancellationException) {
// 已由超时器处理，此处仅忽略或做二次标记
} else {
System.err.println("Task failed with exception: " + cause);
}
}
}
// 安全关闭
executor.shutdown();
timeoutScheduler.shutdown();
if (!executor.awaitTermination(10, TimeUnit.SECONDS)) {
executor.shutdownNow();
}
if (!timeoutScheduler.awaitTermination(2, TimeUnit.SECONDS)) {
timeoutScheduler.shutdownNow();
}
System.out.println("✓ All executors shut down.");
}
}⚠️ 关键注意事项
- 任务必须响应中断：Runnable 内部需定期检查 Thread.interrupted() 或捕获 InterruptedException，否则 cancel(true) 无法生效（例如纯 CPU 密集型循环无中断点，则超时无效）。
- 避免调度器资源泄漏：ScheduledExecutorService 必须显式 shutdown()，否则 JVM 不会退出。
- 线程池大小建议：newFixedThreadPool(N) 的 N 应基于 CPU 密集型（≈核数）或 I/O 密集型（≈2×核数）合理设定；本例中 Runtime.getRuntime().availableProcessors() 是良好起点。
- 日志与可观测性：生产环境建议使用 SLF4J + MDC 记录 taskId，便于链路追踪；超时统计可对接 Prometheus。
- 替代方案权衡：若任务本身支持 CompletableFuture.orTimeout()（Java 9+），可改用 supplyAsync(...).orTimeout(5, SECONDS)，但需将 Runnable 转为 Supplier，且仍需确保底层可中断。
✅ 总结
该方案实现了粒度精确、资源可控、非阻塞、可审计的并行超时控制：每个任务拥有独立倒计时，超时即中断、即日志、即腾出线程，完全符合“100 个任务并行跑，每个最多 5 秒”的核心诉求。无需第三方库，纯 JDK 并发工具即可稳健落地。
立即学习“Java免费学习笔记（深入）”；
