本文介绍如何使用 executorservice 并行执行大量 runnable 任务，同时为每个任务单独设置 5 秒超时；超时时主动中断长任务、记录日志，并确保线程池资源高效复用。
本文介绍如何使用 executorservice 并行执行大量 runnable 任务，同时为每个任务单独设置 5 秒超时；超时时主动中断长任务、记录日志，并确保线程池资源高效复用。
在高并发场景中，仅靠 invokeAll(tasks, timeout, unit) 无法满足「为每个 Runnable 单独设超时」的需求——它会以整个批处理为单位等待（即所有任务最多运行 5 秒），而非每个任务独立计时。真正需要的是：每个任务启动时启动一个对应的倒计时监控器，一旦超时即调用 Future.cancel(true) 中断其执行。
以下是一个生产就绪的实现方案，兼顾线程安全、可观察性与资源可控性：
✅ 核心设计思路
- 使用 ExecutorService 执行实际任务（推荐 newFixedThreadPool(N)，N 通常为 CPU 核心数）；
- 使用独立的 ScheduledExecutorService 负责超时调度，避免阻塞主执行线程；
- 将 Runnable 包装为 Callable
，返回实际耗时，便于后续统计； - 每个任务提交后立即注册一个 5 秒后触发的 schedule() 任务，检查并取消未完成的 Future；
- 主线程不阻塞等待，而是轮询或批量 get() 获取结果（建议配合 isDone() + get(0, TimeUnit.NANOSECONDS) 非阻塞检查）。
✅ 完整可运行示例
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
public class ParallelTaskWithTimeout {
private static final int CORE_POOL_SIZE = Runtime.getRuntime().availableProcessors();
private static final long TIMEOUT_MS = 5_000; // 5 seconds per task
public static void main(String[] args) throws InterruptedException {
ExecutorService workerPool = Executors.newFixedThreadPool(CORE_POOL_SIZE);
ScheduledExecutorService timeoutScheduler = Executors.newSingleThreadScheduledExecutor();
List<future>> futures = new ArrayList();
AtomicInteger timeoutCount = new AtomicInteger(0);
// Submit 100 tasks with individual timeout monitoring
for (int i = 0; i {
try {
int sleepTime = 1 + (int) (Math.random() * 10); // 1–10 sec
System.out.printf("[TASK-%d] START → will sleep %d sec%n", taskId, sleepTime);
Thread.sleep(sleepTime * 1000);
System.out.printf("[TASK-%d] DONE ✓ (%d sec)%n", taskId, sleepTime);
} catch (InterruptedException e) {
System.out.printf("[TASK-%d] CANCELLED ✗ (interrupted)%n", taskId);
Thread.currentThread().interrupt(); // restore interrupt status
}
};
// Wrap as Callable to measure execution time
Future<long> future = workerPool.submit(() -> {
long start = System.currentTimeMillis();
task.run();
return System.currentTimeMillis() - start;
});
// Schedule timeout check: cancel if not done within TIMEOUT_MS
timeoutScheduler.schedule(() -> {
if (!future.isDone()) {
boolean cancelled = future.cancel(true);
if (cancelled) {
System.err.printf("[TIMEOUT] TASK-%d forcibly cancelled after %d ms%n", taskId, TIMEOUT_MS);
timeoutCount.incrementAndGet();
}
}
}, TIMEOUT_MS, TimeUnit.MILLISECONDS);
futures.add(future);
}
// Wait for all futures to complete (with timeout handling)
System.out.println("→ Waiting for all tasks to finish or timeout...");
for (int i = 0; i f = futures.get(i);
try {
Long duration = f.get(10, TimeUnit.SECONDS); // allow extra margin
if (duration != null) {
System.out.printf("[RESULT] TASK-%d completed in %d ms%n", i, duration);
}
} catch (ExecutionException e) {
System.err.printf("[ERROR] TASK-%d failed: %s%n", i, e.getCause());
} catch (TimeoutException e) {
System.err.printf("[TIMEOUT] TASK-%d timed out during result retrieval%n", i);
} catch (CancellationException e) {
// already logged at cancellation time
}
}
System.out.printf("✅ Summary: %d / 100 tasks timed out%n", timeoutCount.get());
// Graceful shutdown
workerPool.shutdown();
timeoutScheduler.shutdown();
if (!workerPool.awaitTermination(30, TimeUnit.SECONDS)) {
workerPool.shutdownNow();
}
if (!timeoutScheduler.awaitTermination(10, TimeUnit.SECONDS)) {
timeoutScheduler.shutdownNow();
}
}
}</long></future>
⚠️ 关键注意事项
- 任务必须响应中断：future.cancel(true) 本质是调用 Thread.interrupt()，因此你的 Runnable 内部需正确处理 InterruptedException（如及时退出循环、释放资源），否则无法真正终止。
- 避免共享 ScheduledExecutorService 泄漏：务必在最后调用 shutdown()，否则 JVM 不会退出（守护线程默认不阻止进程结束，但 newSingleThreadScheduledExecutor() 创建的是非守护线程）。
- 不要复用 Future 实例做多次 cancel()：Future.cancel() 是幂等的，但重复调用无意义；确保每个任务只被一个 schedule() 监控。
- 性能权衡：每任务启动一个定时任务会带来少量调度开销（100 个任务 ≈ 100 次调度），但在万级以下规模完全可接受；若需极致性能，可考虑基于 CompletableFuture.orTimeout() 的响应式方案（Java 9+）。
✅ 替代方案（Java 9+ 推荐）
若项目已升级至 Java 9 或更高版本，可更简洁地使用 CompletableFuture：
CompletableFuture.supplyAsync(() -> {
// your task logic here
return Duration.between(start, Instant.now()).toMillis();
}, workerPool)
.orTimeout(5, TimeUnit.SECONDS)
.exceptionally(ex -> {
if (ex instanceof TimeoutException) {
System.err.println("Task timed out!");
}
return -1L;
});
但注意：orTimeout 仅终止 CompletableFuture 链，不会中断底层线程，仍需确保任务本身支持中断。
综上，采用「独立定时器 + Future.cancel(true)」是最通用、可控且兼容 JDK 8+ 的方案，适用于绝大多数需要精细化超时控制的并行任务场景。
Java免费学习笔记：立即使用
解锁 Java 大师之旅：从入门到精通的终极指南
