线程池原理与实践

为什么使用线程池？

• 降低资源消耗：复用线程，避免频繁创建销毁
• 提高响应速度：任务到达时无需等待线程创建
• 便于管理：统一管理、调优、监控

线程池核心参数

/**
 * 线程池核心参数
 * @author yjhu
 */
public ThreadPoolExecutor(
    int corePoolSize,       // 核心线程数
    int maximumPoolSize,    // 最大线程数
    long keepAliveTime,     // 空闲线程存活时间
    TimeUnit unit,          // 时间单位
    BlockingQueue<Runnable> workQueue,  // 任务队列
    ThreadFactory threadFactory,         // 线程工厂
    RejectedExecutionHandler handler     // 拒绝策略
)

参数详解

参数	说明
corePoolSize	常驻线程数，即使空闲也不会被回收（除非设置 allowCoreThreadTimeOut）
maximumPoolSize	线程池最大容量
keepAliveTime	非核心线程空闲超过此时间会被回收
workQueue	任务等待队列
threadFactory	创建线程的工厂，可自定义线程名
handler	队列满且线程数达到最大时的处理策略

线程池工作流程

                    ┌──────────────────────────────────────┐
                    │         任务提交 execute(task)        │
                    └──────────────────┬───────────────────┘
                                       ↓
                    ┌──────────────────────────────────────┐
                    │     当前线程数 < corePoolSize ?       │
                    └──────────┬───────────────┬───────────┘
                               │ Yes           │ No
                               ↓               ↓
                    ┌────────────────┐   ┌────────────────┐
                    │  创建核心线程   │   │   队列未满 ?    │
                    │  执行任务      │   └───┬───────┬────┘
                    └────────────────┘       │ Yes   │ No
                                             ↓       ↓
                              ┌────────────────┐  ┌────────────────┐
                              │  任务入队等待   │  │ 线程数 < max ? │
                              └────────────────┘  └──┬─────────┬───┘
                                                     │ Yes     │ No
                                                     ↓         ↓
                                        ┌───────────────┐  ┌────────────┐
                                        │ 创建非核心线程 │  │  拒绝策略   │
                                        │  执行任务     │  └────────────┘
                                        └───────────────┘

常见工作队列

队列类型	特点
ArrayBlockingQueue	有界队列，基于数组
LinkedBlockingQueue	可选有界，基于链表（默认无界，慎用）
SynchronousQueue	不存储元素，直接传递给线程
PriorityBlockingQueue	支持优先级排序的无界队列

拒绝策略

// 1. 抛出异常（默认）
new ThreadPoolExecutor.AbortPolicy()

// 2. 由调用线程执行任务
new ThreadPoolExecutor.CallerRunsPolicy()

// 3. 丢弃任务，不抛异常
new ThreadPoolExecutor.DiscardPolicy()

// 4. 丢弃队列中最老的任务
new ThreadPoolExecutor.DiscardOldestPolicy()

线程池最佳实践

1. 不要使用 Executors 创建

// ❌ 不推荐：使用无界队列，可能 OOM
ExecutorService executor = Executors.newFixedThreadPool(10);
ExecutorService executor = Executors.newCachedThreadPool();

// ✅ 推荐：手动创建，明确参数
ThreadPoolExecutor executor = new ThreadPoolExecutor(
    10,                                     // 核心线程数
    20,                                     // 最大线程数
    60, TimeUnit.SECONDS,                   // 空闲存活时间
    new ArrayBlockingQueue<>(100),          // 有界队列
    new ThreadFactoryBuilder()
        .setNameFormat("business-pool-%d")
        .build(),
    new ThreadPoolExecutor.CallerRunsPolicy()  // 拒绝策略
);

2. 合理设置线程数

// CPU 密集型：核心数 + 1
int cpuIntensive = Runtime.getRuntime().availableProcessors() + 1;

// IO 密集型：核心数 * 2 或更高
int ioIntensive = Runtime.getRuntime().availableProcessors() * 2;

// 更精确的计算公式
// 线程数 = 核心数 * 期望 CPU 利用率 * (1 + 等待时间/计算时间)

3. 线程池监控

/**
 * 线程池监控示例
 * @author yjhu
 */
public void monitor(ThreadPoolExecutor executor) {
    System.out.println("核心线程数: " + executor.getCorePoolSize());
    System.out.println("当前线程数: " + executor.getPoolSize());
    System.out.println("活跃线程数: " + executor.getActiveCount());
    System.out.println("队列任务数: " + executor.getQueue().size());
    System.out.println("已完成任务: " + executor.getCompletedTaskCount());
    System.out.println("历史最大线程数: " + executor.getLargestPoolSize());
}

4. 优雅关闭

/**
 * 优雅关闭线程池
 * @author yjhu
 */
public void shutdown(ExecutorService executor) {
    // 不再接收新任务
    executor.shutdown();
    
    try {
        // 等待现有任务完成
        if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
            // 超时后强制关闭
            executor.shutdownNow();
            
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                System.err.println("线程池未能正常关闭");
            }
        }
    } catch (InterruptedException e) {
        executor.shutdownNow();
        Thread.currentThread().interrupt();
    }
}

常见问题

线程池异常处理

// 方式 1：在任务内部 try-catch
executor.execute(() -> {
    try {
        // 业务逻辑
    } catch (Exception e) {
        log.error("任务执行异常", e);
    }
});

// 方式 2：使用 submit + Future
Future<?> future = executor.submit(() -> {
    // 业务逻辑
});
try {
    future.get();
} catch (ExecutionException e) {
    log.error("任务执行异常", e.getCause());
}

// 方式 3：设置 UncaughtExceptionHandler
Thread.setDefaultUncaughtExceptionHandler((t, e) -> {
    log.error("线程 {} 异常", t.getName(), e);
});