Java 中并行执行 Runnable 任务并自动终止超时任务的完整实现方案

本文介绍如何使用 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免费学习笔记（深入）”；