Java线程池最佳实践与源码分析

// ThreadPoolExecutor的七个核心参数
public ThreadPoolExecutor(
    int corePoolSize,        // 核心线程数：即使空闲也不会被回收（除非设置了allowCoreThreadTimeOut）
    int maximumPoolSize,     // 最大线程数：线程池能创建的最大线程数
    long keepAliveTime,      // 非核心线程空闲存活时间
    TimeUnit unit,           // keepAliveTime的时间单位
    BlockingQueue<Runnable> workQueue,  // 任务等待队列
    ThreadFactory threadFactory,        // 线程工厂：自定义线程名称、优先级等
    RejectedExecutionHandler handler    // 拒绝策略：队列满且线程数达上限时的处理
) { ... }

public class ThreadPoolParameters {

    /**
     * CPU密集型任务：线程数 = CPU核心数 + 1
     * 多出的1个线程用于在某线程因缺页中断等原因暂停时保持CPU利用率
     */
    public static ExecutorService cpuIntensivePool() {
        int cpuCores = Runtime.getRuntime().availableProcessors();
        return new ThreadPoolExecutor(
                cpuCores + 1,
                cpuCores + 1,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(256),
                new CustomThreadFactory("cpu-pool"),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );
    }

    /**
     * IO密集型任务：线程数 = CPU核心数 * 2（或更多）
     * 因为线程大部分时间在等待IO，可以让更多线程交替使用CPU
     */
    public static ExecutorService ioIntensivePool() {
        int cpuCores = Runtime.getRuntime().availableProcessors();
        return new ThreadPoolExecutor(
                cpuCores * 2,
                cpuCores * 4,
                60L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1024),
                new CustomThreadFactory("io-pool"),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );
    }

    /**
     * 更精确的线程数计算公式（参考《Java Concurrency in Practice》）
     * N_threads = N_cpu * U_cpu * (1 + W/C)
     * N_cpu: CPU核心数
     * U_cpu: 目标CPU利用率 (0 < U_cpu <= 1)
     * W/C:  等待时间与计算时间的比值
     */
    public static int optimalThreadCount(double targetUtilization, double waitComputeRatio) {
        int cpuCores = Runtime.getRuntime().availableProcessors();
        return (int) Math.ceil(cpuCores * targetUtilization * (1 + waitComputeRatio));
    }
}

public class WorkQueueComparison {

    // 1. ArrayBlockingQueue - 有界数组队列
    // 适用场景：需要严格控制队列大小，防止OOM
    public static ExecutorService withArrayQueue() {
        return new ThreadPoolExecutor(
                4, 8, 60, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(100), // 固定容量100
                new ThreadPoolExecutor.AbortPolicy()
        );
    }

    // 2. LinkedBlockingQueue - 有界链表队列（推荐指定容量）
    // 适用场景：通用场景，吞吐量高于ArrayBlockingQueue
    public static ExecutorService withLinkedQueue() {
        return new ThreadPoolExecutor(
                4, 8, 60, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(500),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );
    }

    // 3. SynchronousQueue - 不存储元素的队列
    // 适用场景：每个任务必须立即有线程执行，适合短平快的任务
    // 注意：如果maximumPoolSize设置过小，容易触发拒绝策略
    public static ExecutorService withSyncQueue() {
        return new ThreadPoolExecutor(
                0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,
                new SynchronousQueue<>() // Executors.newCachedThreadPool()的配置
        );
    }

    // 4. PriorityBlockingQueue - 优先级队列
    // 适用场景：不同优先级的任务需要区别处理
    public static ExecutorService withPriorityQueue() {
        return new ThreadPoolExecutor(
                4, 4, 0, TimeUnit.MILLISECONDS,
                new PriorityBlockingQueue<>(100)
        );
    }
}

public class RejectionPolicyDemo {

    // JDK内置的四种拒绝策略
    // 1. AbortPolicy（默认）：直接抛出RejectedExecutionException
    // 2. CallerRunsPolicy：由提交任务的线程执行（起到负反馈限流的效果）
    // 3. DiscardPolicy：静默丢弃任务（危险！丢失任务无任何通知）
    // 4. DiscardOldestPolicy：丢弃队列中最旧的任务，然后重试提交

    /**
     * 自定义拒绝策略：记录日志 + 持久化到数据库 + 告警
     * 生产环境推荐，确保任务不会悄无声息地丢失
     */
    public static class LogAndPersistPolicy implements RejectedExecutionHandler {

        private static final Logger log = LoggerFactory.getLogger(LogAndPersistPolicy.class);
        private final TaskPersistenceService persistenceService;
        private final AlertService alertService;

        public LogAndPersistPolicy(TaskPersistenceService persistenceService,
                                   AlertService alertService) {
            this.persistenceService = persistenceService;
            this.alertService = alertService;
        }

        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            // 记录被拒绝任务的信息
            log.error("任务被拒绝! 线程池状态: 活跃线程={}, 队列大小={}, 已完成={}",
                    executor.getActiveCount(),
                    executor.getQueue().size(),
                    executor.getCompletedTaskCount());

            // 持久化到数据库，后续可以重新执行
            if (r instanceof PersistableTask task) {
                persistenceService.save(task);
                log.info("被拒绝的任务已持久化: {}", task.getTaskId());
            }

            // 发送告警通知
            alertService.sendAlert("线程池任务被拒绝，请检查系统负载");

            // 根据业务需求决定是否抛出异常
            throw new RejectedExecutionException("线程池已饱和，任务已持久化等待重试");
        }
    }
}

/**
 * ThreadPoolExecutor.execute() 源码解析
 *
 * ctl是一个AtomicInteger，高3位存储线程池状态，低29位存储线程数量
 * 状态值: RUNNING(-1) < SHUTDOWN(0) < STOP(1) < TIDYING(2) < TERMINATED(3)
 */
public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();

    int c = ctl.get();

    // 步骤1: 如果当前线程数 < corePoolSize，尝试创建核心线程
    if (workerCountOf(c) < corePoolSize) {
        // addWorker的第二个参数true表示以corePoolSize为边界
        if (addWorker(command, true))
            return;
        // 创建失败（可能并发导致），重新获取ctl
        c = ctl.get();
    }

    // 步骤2: 线程池处于RUNNING状态，尝试将任务放入队列
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        // 二次检查：入队后线程池可能已被shutdown
        if (!isRunning(recheck) && remove(command))
            reject(command); // 不在RUNNING状态，移除任务并拒绝
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false); // 没有工作线程了，创建一个（防止任务无人消费）
    }
    // 步骤3: 队列已满，尝试创建非核心线程
    else if (!addWorker(command, false))
        // 创建失败（线程数已达maximumPoolSize），执行拒绝策略
        reject(command);
}

/**
 * addWorker方法核心逻辑简化版
 * 负责创建Worker线程并启动
 */
private boolean addWorker(Runnable firstTask, boolean core) {
    // 外层循环：检查线程池状态
    retry:
    for (int c = ctl.get();;) {
        // 如果线程池已关闭且满足特定条件，拒绝添加
        if (runStateAtLeast(c, SHUTDOWN)
            && (runStateAtLeast(c, STOP) || firstTask != null || workQueue.isEmpty()))
            return false;

        // 内层循环：CAS增加线程计数
        for (;;) {
            // 根据core参数决定边界检查
            if (workerCountOf(c) >= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))
                return false;
            // CAS成功，跳出双重循环
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();
            // 状态变化了，从外层循环重新开始
            if (runStateAtLeast(c, SHUTDOWN))
                continue retry;
        }
    }

    // CAS成功后，创建Worker并启动线程
    boolean workerStarted = false;
    Worker w = new Worker(firstTask);
    Thread t = w.thread;

    // 加锁将Worker添加到workers集合
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        if (isRunning(ctl.get()) || (runStateLessThan(ctl.get(), STOP) && firstTask == null)) {
            workers.add(w);
            workerStarted = true;
        }
    } finally {
        mainLock.unlock();
    }

    if (workerStarted)
        t.start(); // 启动工作线程，线程启动后会执行Worker.run() -> runWorker()

    return workerStarted;
}

public class CustomThreadFactory implements ThreadFactory {

    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;
    private final boolean daemon;
    private final int priority;
    private final Thread.UncaughtExceptionHandler exceptionHandler;

    public CustomThreadFactory(String poolName) {
        this(poolName, false, Thread.NORM_PRIORITY, null);
    }

    public CustomThreadFactory(String poolName, boolean daemon, int priority,
                                Thread.UncaughtExceptionHandler handler) {
        this.namePrefix = poolName + "-thread-";
        this.daemon = daemon;
        this.priority = priority;
        this.exceptionHandler = handler != null ? handler : (t, e) -> {
            LoggerFactory.getLogger(CustomThreadFactory.class)
                    .error("线程 {} 发生未捕获异常", t.getName(), e);
        };
    }

    @Override
    public Thread newThread(Runnable r) {
        Thread thread = new Thread(r, namePrefix + threadNumber.getAndIncrement());
        thread.setDaemon(daemon);
        thread.setPriority(priority);
        thread.setUncaughtExceptionHandler(exceptionHandler);
        return thread;
    }
}

/**
 * 线程池监控组件
 * 定期采集线程池指标并上报
 */
@Component
public class ThreadPoolMonitor {

    private static final Logger log = LoggerFactory.getLogger(ThreadPoolMonitor.class);
    private final List<ThreadPoolExecutor> monitoredPools = new CopyOnWriteArrayList<>();

    public void register(String name, ThreadPoolExecutor pool) {
        monitoredPools.add(pool);

        // 注册到Micrometer指标系统
        Metrics.gauge("thread.pool.active." + name, pool, ThreadPoolExecutor::getActiveCount);
        Metrics.gauge("thread.pool.queue.size." + name, pool,
                p -> p.getQueue().size());
        Metrics.gauge("thread.pool.pool.size." + name, pool,
                ThreadPoolExecutor::getPoolSize);
        Metrics.gauge("thread.pool.core.size." + name, pool,
                ThreadPoolExecutor::getCorePoolSize);
        Metrics.gauge("thread.pool.largest.size." + name, pool,
                ThreadPoolExecutor::getLargestPoolSize);
    }

    /**
     * 定时打印线程池状态（也可对接Prometheus等监控系统）
     */
    @Scheduled(fixedRate = 30000)
    public void reportStatus() {
        for (ThreadPoolExecutor pool : monitoredPools) {
            log.info("线程池状态 - " +
                            "核心线程数: {}, " +
                            "当前线程数: {}, " +
                            "活跃线程数: {}, " +
                            "历史最大线程数: {}, " +
                            "队列大小: {}, " +
                            "已完成任务数: {}",
                    pool.getCorePoolSize(),
                    pool.getPoolSize(),
                    pool.getActiveCount(),
                    pool.getLargestPoolSize(),
                    pool.getQueue().size(),
                    pool.getCompletedTaskCount());

            // 告警：队列使用率超过80%
            if (pool.getQueue() instanceof LinkedBlockingQueue<?> queue) {
                int capacity = queue.remainingCapacity() + queue.size();
                double usage = (double) queue.size() / capacity;
                if (usage > 0.8) {
                    log.warn("线程池队列使用率过高: {:.1f}%", usage * 100);
                }
            }
        }
    }
}

public class ThreadPoolPitfalls {

    // 不推荐：newFixedThreadPool使用无界队列，可能导致OOM
    ExecutorService bad1 = Executors.newFixedThreadPool(10);
    // 内部使用 new LinkedBlockingQueue<>()，默认容量Integer.MAX_VALUE

    // 不推荐：newCachedThreadPool可能创建大量线程
    ExecutorService bad2 = Executors.newCachedThreadPool();
    // maximumPoolSize = Integer.MAX_VALUE，极端情况下创建大量线程

    // 推荐：手动创建，明确所有参数
    ExecutorService good = new ThreadPoolExecutor(
            10, 20,
            60L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(500),
            new CustomThreadFactory("business-pool"),
            new ThreadPoolExecutor.CallerRunsPolicy()
    );
}

public class ExceptionHandling {

    private final ExecutorService pool = new ThreadPoolExecutor(
            4, 8, 60, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(100),
            new CustomThreadFactory("task-pool")
    );

    // 错误做法：execute提交的任务如果抛出异常，线程会终止并创建新线程
    // 异常信息只会输出到stderr，容易被忽略
    public void badSubmit() {
        pool.execute(() -> {
            throw new RuntimeException("出错了"); // 异常被吞没
        });
    }

    // 正确做法1：在任务内部捕获异常
    public void goodSubmit1() {
        pool.execute(() -> {
            try {
                riskyOperation();
            } catch (Exception e) {
                log.error("任务执行失败", e);
            }
        });
    }

    // 正确做法2：使用submit + Future获取异常
    public void goodSubmit2() {
        Future<?> future = pool.submit(() -> {
            riskyOperation();
        });

        try {
            future.get(10, TimeUnit.SECONDS);
        } catch (ExecutionException e) {
            log.error("任务执行异常", e.getCause());
        } catch (TimeoutException e) {
            future.cancel(true);
            log.error("任务执行超时");
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    // 正确做法3：使用自定义ThreadPoolExecutor重写afterExecute
    public static class MonitoredThreadPoolExecutor extends ThreadPoolExecutor {

        public MonitoredThreadPoolExecutor(int corePoolSize, int maxPoolSize,
                                            long keepAlive, TimeUnit unit,
                                            BlockingQueue<Runnable> queue) {
            super(corePoolSize, maxPoolSize, keepAlive, unit, queue);
        }

        @Override
        protected void afterExecute(Runnable r, Throwable t) {
            super.afterExecute(r, t);
            // 处理execute()提交的任务异常
            if (t != null) {
                LoggerFactory.getLogger(getClass())
                        .error("任务执行异常（execute提交）", t);
            }
            // 处理submit()提交的任务异常
            if (t == null && r instanceof Future<?> future) {
                try {
                    if (future.isDone()) {
                        future.get();
                    }
                } catch (ExecutionException e) {
                    LoggerFactory.getLogger(getClass())
                            .error("任务执行异常（submit提交）", e.getCause());
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } catch (CancellationException e) {
                    // 任务被取消，可忽略
                }
            }
        }
    }
}

public class ShutdownBestPractice {

    private final ExecutorService pool;

    public ShutdownBestPractice(ExecutorService pool) {
        this.pool = pool;
    }

    /**
     * 优雅关闭线程池的标准模式
     */
    public void gracefulShutdown() {
        // 第一步：停止接受新任务
        pool.shutdown();

        try {
            // 第二步：等待已提交的任务完成（最多等待30秒）
            if (!pool.awaitTermination(30, TimeUnit.SECONDS)) {
                // 第三步：超时后强制关闭
                List<Runnable> pendingTasks = pool.shutdownNow();
                log.warn("线程池强制关闭，{}个任务未执行", pendingTasks.size());

                // 第四步：再等待一段时间
                if (!pool.awaitTermination(10, TimeUnit.SECONDS)) {
                    log.error("线程池未能完全关闭");
                }
            }
        } catch (InterruptedException e) {
            pool.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }

    /**
     * Spring环境中使用@PreDestroy优雅关闭
     */
    @PreDestroy
    public void destroy() {
        gracefulShutdown();
    }
}

@RestController
@RequestMapping("/threadpool")
public class ThreadPoolController {

    private final ThreadPoolExecutor businessPool;

    public ThreadPoolController(ThreadPoolExecutor businessPool) {
        this.businessPool = businessPool;
    }

    @PostMapping("/update")
    public String updatePoolConfig(@RequestParam int coreSize,
                                    @RequestParam int maxSize,
                                    @RequestParam int queueCapacity) {
        // 校验参数合法性
        if (coreSize <= 0 || maxSize < coreSize) {
            return "参数不合法";
        }

        int oldCore = businessPool.getCorePoolSize();
        int oldMax = businessPool.getMaximumPoolSize();

        // 注意顺序：先调大的，再调小的，避免临时违反 core <= max 约束
        if (coreSize > oldCore) {
            businessPool.setMaximumPoolSize(maxSize);
            businessPool.setCorePoolSize(coreSize);
        } else {
            businessPool.setCorePoolSize(coreSize);
            businessPool.setMaximumPoolSize(maxSize);
        }

        // 动态调整队列容量需要自定义可调容量的队列
        // LinkedBlockingQueue不支持运行时改容量

        return String.format("线程池参数已更新: core %d->%d, max %d->%d",
                oldCore, coreSize, oldMax, maxSize);
    }

    @GetMapping("/status")
    public Map<String, Object> getStatus() {
        return Map.of(
                "corePoolSize", businessPool.getCorePoolSize(),
                "maximumPoolSize", businessPool.getMaximumPoolSize(),
                "activeCount", businessPool.getActiveCount(),
                "poolSize", businessPool.getPoolSize(),
                "queueSize", businessPool.getQueue().size(),
                "completedTaskCount", businessPool.getCompletedTaskCount(),
                "taskCount", businessPool.getTaskCount()
        );
    }
}