#线程池故障实战

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

// 错误配置1：无界队列
ExecutorService pool1 = Executors.newFixedThreadPool(100);
// 等价于：
new ThreadPoolExecutor(100, 100, 0L, TimeUnit.MILLISECONDS,
    new LinkedBlockingQueue<Runnable>());  // Integer.MAX_VALUE!

// 问题：队列无限膨胀，内存溢出

// 错误配置2：单线程池
ExecutorService pool2 = Executors.newSingleThreadExecutor();
// 问题：所有任务串行执行

// 错误配置3：缓存线程池
ExecutorService pool3 = Executors.newCachedThreadPool();
// 问题：线程数无限增长，高并发时创建大量线程

// 错误配置4：队列容量过大
new ThreadPoolExecutor(10, 20, 0L, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(1000000));  // 100万容量!

// 问题：队列太大，任务堆积严重，响应延迟

// 正确配置：IO密集型任务
public class ThreadPoolConfig {

    @Bean
    public ThreadPoolExecutor orderExecutor() {
        // IO密集型：线程数 = CPU核数 × 2
        int cpuCores = Runtime.getRuntime().availableProcessors();
        int poolSize = cpuCores * 2;

        return new ThreadPoolExecutor(
            poolSize,                     // 核心线程数
            poolSize * 2,                 // 最大线程数
            60L,                          // 空闲存活时间
            TimeUnit.SECONDS,
            // 必须是有限队列，控制任务堆积
            new LinkedBlockingQueue<>(1000),
            // 自定义线程工厂
            new ThreadFactoryBuilder()
                .setNameFormat("order-pool-%d")
                .setDaemon(false)
                .build(),
            // 拒绝策略
            new ThreadPoolExecutor.AbortPolicy()
        );
    }
}

// 正确配置：CPU密集型任务
@Bean
public ThreadPoolExecutor computeExecutor() {
    int cpuCores = Runtime.getRuntime().availableProcessors();

    return new ThreadPoolExecutor(
        cpuCores + 1,
        cpuCores + 1,
        0L, TimeUnit.MILLISECONDS,
        new LinkedBlockingQueue<>(100),
        new ThreadFactoryBuilder()
            .setNameFormat("compute-pool-%d")
            .build(),
        new ThreadPoolExecutor.AbortPolicy()
    );
}

// 队列选择策略
public class QueueSelectionStrategy {

    /**
     * 根据任务特性选择队列
     */
    public BlockingQueue<Runnable> selectQueue(TaskType type, int capacity) {
        switch (type) {
            case URGENT:
                // 紧急任务：SynchronousQueue，必须立即处理
                return new SynchronousQueue<>();

            case IMPORTANT:
                // 重要任务：ArrayBlockingQueue，有界可控
                return new ArrayBlockingQueue<>(capacity);

            case NORMAL:
                // 普通任务：LinkedBlockingQueue，允许一定堆积
                return new LinkedBlockingQueue<>(capacity);

            case BACKGROUND:
                // 后台任务：PriorityBlockingQueue，按优先级
                return new PriorityBlockingQueue<>(capacity,
                    Comparator.comparingInt(Task::getPriority));

            default:
                return new LinkedBlockingQueue<>(capacity);
        }
    }
}

// 自定义拒绝策略：记录+告警+降级
public class LoggingRejectHandler implements RejectedExecutionHandler {

    @Autowired
    private AlertService alertService;

    @Autowired
    private MetricsService metricsService;

    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        // 1. 记录日志
        log.error("任务被拒绝，线程池状态：活跃={}, 队列大小={}, 核心={}, 最大={}",
            executor.getActiveCount(),
            executor.getQueue().size(),
            executor.getCorePoolSize(),
            executor.getMaximumPoolSize());

        // 2. 发送告警
        alertService.sendAlert("线程池拒绝任务，当前队列堆积：" +
            executor.getQueue().size());

        // 3. 记录指标
        metricsService.increment("threadpool.rejected");

        // 4. 降级处理
        if (r instanceof Callable) {
            try {
                // 降级：同步执行，设置超时
                Object result = ((Callable<?>) r).call();
                log.warn("降级执行成功");
            } catch (Exception e) {
                log.error("降级执行失败", e);
            }
        }
    }
}

// 降级策略：写入消息队列
public class MQRejectHandler implements RejectedExecutionHandler {

    @Autowired
    private MessageProducer producer;

    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        if (r instanceof TaskRunnable) {
            TaskRunnable task = (TaskRunnable) r;
            // 写入消息队列，稍后处理
            producer.send("task.retry", task.getTaskData());
            log.warn("任务写入MQ重试队列");
        }
    }
}

// 线程池监控
@Component
@Slf4j
public class ThreadPoolMonitor {

    @Autowired
    private Map<String, ThreadPoolExecutor> threadPools;

    @Scheduled(fixedRate = 10000)
    public void monitor() {
        for (Map.Entry<String, ThreadPoolExecutor> entry : threadPools.entrySet()) {
            String name = entry.getKey();
            ThreadPoolExecutor pool = entry.getValue();

            int activeCount = pool.getActiveCount();
            int poolSize = pool.getPoolSize();
            int queueSize = pool.getQueue().size();
            long completedTaskCount = pool.getCompletedTaskCount();
            long totalTaskCount = pool.getTaskCount();

            // 计算使用率
            double activeRate = (double) activeCount / poolSize;
            double queueUsage = (double) queueSize / 1000;  // 假设队列容量1000

            log.info("线程池[{}]：活跃={}/{}, 队列={}, 完成={}/{}",
                name, activeCount, poolSize, queueSize,
                completedTaskCount, totalTaskCount);

            // 告警条件
            if (activeRate > 0.8) {
                alert("线程池[{}]活跃率超过80%", name);
            }
            if (queueSize > 800) {
                alert("线程池[{}]队列堆积超过80%", name);
            }
        }
    }
}

// 动态调整线程池
@Service
public class DynamicThreadPool {

    @Autowired
    private ConfigService configService;

    public void adjustThreadPool(String poolName, ThreadPoolExecutor pool) {
        // 从配置中心获取新的配置
        int coreSize = configService.getInt(poolName + ".coreSize", 10);
        int maxSize = configService.getInt(poolName + ".maxSize", 20);

        // 调整核心线程数
        pool.setCorePoolSize(coreSize);

        // 调整最大线程数
        pool.setMaximumPoolSize(maxSize);

        // 设置核心线程超时
        pool.allowCoreThreadTimeOut(
            configService.getBoolean(poolName + ".allowTimeout", true));

        log.info("线程池[{}]已调整：核心={}, 最大={}",
            poolName, coreSize, maxSize);
    }
}