Java线程池 #

Executor
└── ExecutorService
    ├── ThreadPoolExecutor
    │   └── ScheduledThreadPoolExecutor
    └── ForkJoinPool

import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;

ThreadPoolExecutor executor = new ThreadPoolExecutor(
    5,                              // 核心线程数
    10,                             // 最大线程数
    60L,                            // 空闲线程存活时间
    TimeUnit.SECONDS,               // 时间单位
    new ArrayBlockingQueue<>(100),  // 任务队列
    Executors.defaultThreadFactory(),  // 线程工厂
    new ThreadPoolExecutor.AbortPolicy()  // 拒绝策略
);

1. 线程数 < corePoolSize → 创建新线程
2. 线程数 = corePoolSize → 加入队列
3. 队列满 → 创建新线程直到maximumPoolSize
4. 线程数 = maximumPoolSize → 执行拒绝策略

// AbortPolicy：抛异常（默认）
new ThreadPoolExecutor.AbortPolicy();

// CallerRunsPolicy：调用者执行
new ThreadPoolExecutor.CallerRunsPolicy();

// DiscardPolicy：丢弃任务
new ThreadPoolExecutor.DiscardPolicy();

// DiscardOldestPolicy：丢弃最老任务
new ThreadPoolExecutor.DiscardOldestPolicy();

ExecutorService executor = Executors.newFixedThreadPool(5);

// 固定线程数，无界队列
// corePoolSize = maximumPoolSize = n

ExecutorService executor = Executors.newCachedThreadPool();

// 可缓存线程池，线程数无限
// corePoolSize = 0, maximumPoolSize = Integer.MAX_VALUE

ExecutorService executor = Executors.newSingleThreadExecutor();

// 单线程执行器
// corePoolSize = maximumPoolSize = 1

ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(5);

// 延迟执行
scheduler.schedule(() -> {
    System.out.println("延迟1秒执行");
}, 1, TimeUnit.SECONDS);

// 定期执行
scheduler.scheduleAtFixedRate(() -> {
    System.out.println("每2秒执行");
}, 0, 2, TimeUnit.SECONDS);

// 不推荐使用Executors创建线程池
// 1. FixedThreadPool和SingleThreadExecutor使用无界队列，可能OOM
// 2. CachedThreadPool线程数无限，可能OOM

// 推荐：手动创建ThreadPoolExecutor
ThreadPoolExecutor executor = new ThreadPoolExecutor(
    5, 10, 60L, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(1000)
);

executor.execute(() -> {
    System.out.println("执行任务");
});

Future<?> future1 = executor.submit(() -> {
    System.out.println("执行任务");
});

Future<Integer> future2 = executor.submit(() -> {
    return 100;
});

try {
    Integer result = future2.get();
    System.out.println(result);
} catch (Exception e) {
    e.printStackTrace();
}

List<Callable<Integer>> tasks = Arrays.asList(
    () -> 1,
    () -> 2,
    () -> 3
);

List<Future<Integer>> futures = executor.invokeAll(tasks);

Integer result = executor.invokeAny(tasks);  // 返回第一个完成的结果

executor.shutdown();  // 平滑关闭，等待任务完成

if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
    executor.shutdownNow();  // 强制关闭
}

List<Runnable> notExecuted = executor.shutdownNow();  // 立即关闭
// 返回未执行的任务列表

executor.isShutdown();     // 是否已关闭
executor.isTerminated();   // 是否已终止

ThreadPoolExecutor executor = new ThreadPoolExecutor(...);

// 监控信息
executor.getActiveCount();      // 活动线程数
executor.getCompletedTaskCount();  // 已完成任务数
executor.getQueue().size();     // 队列大小
executor.getPoolSize();         // 当前线程数
executor.getCorePoolSize();     // 核心线程数
executor.getMaximumPoolSize();  // 最大线程数

// CPU核心数 + 1
int cores = Runtime.getRuntime().availableProcessors();
ThreadPoolExecutor executor = new ThreadPoolExecutor(
    cores + 1, cores + 1, 0L, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>()
);

// CPU核心数 * 2
int cores = Runtime.getRuntime().availableProcessors();
ThreadPoolExecutor executor = new ThreadPoolExecutor(
    cores * 2, cores * 2, 60L, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>()
);