【架构实战】线程池设计：高并发系统的资源管理艺术

一、一个newFixedThreadPool差点拖垮整个系统2021年我们一个新人写了一段代码ExecutorServiceexecutorExecutors.newFixedThreadPool(50);for(Orderorder:orders){executor.submit(()-processOrder(order));}看起来没问题对吧但他没考虑一点orders有10万条每条任务要处理3秒。50个线程处理10万条需要6000秒。更关键的是每个任务还要往另一个队列里塞数据那个队列也是无界的。内存直接被撑爆OOM崩溃。从那以后我们对线程池的使用有了严格的规范。二、线程池核心参数2.1 ThreadPoolExecutor七大参数/** * ThreadPoolExecutor七大参数 */publicThreadPoolExecutor(intcorePoolSize,// 核心线程数intmaximumPoolSize,// 最大线程数longkeepAliveTime,// 非核心线程空闲存活时间TimeUnitunit,// 时间单位BlockingQueueRunnableworkQueue,// 工作队列ThreadFactorythreadFactory,// 线程工厂RejectedExecutionHandlerhandler// 拒绝策略)2.2 参数配置原则┌─────────────────────────────────────────────────────────────────┐ │ 线程池参数配置原则 │ │ │ │ CPU密集型任务 │ │ - 核心线程数 CPU核数 1 │ │ - 最大线程数 CPU核数 1 │ │ - 队列有界队列 │ │ │ │ IO密集型任务 │ │ - 核心线程数 CPU核数 × 2 │ │ - 最大线程数 CPU核数 × 4 │ │ - 队列有界队列 │ │ │ │ 混合型任务 │ │ - 核心线程数 CPU核数 × 2 │ │ - 最大线程数 CPU核数 × 4~8 │ │ - 队列有界队列 │ │ │ │ ⚠️ 关键原则 │ │ 1. 必须使用有界队列防止OOM │ │ 2. 必须设置合理的拒绝策略 │ │ 3. 线程命名要可识别 │ │ │ └──────────────────────────────────────────────────────────────────┘三、线程池配置最佳实践3.1 基础配置/** * 线程池配置 */ConfigurationpublicclassThreadPoolConfig{Bean(orderProcessPool)publicThreadPoolExecutororderProcessPool(){returnnewThreadPoolExecutor(10,// 核心线程数50,// 最大线程数60,TimeUnit.SECONDS,// 空闲存活时间newLinkedBlockingQueue(1000),// 有界队列容量1000newNamedThreadFactory(order-process),// 命名线程工厂newCustomRejectedExecutionHandler()// 自定义拒绝策略);}Bean(asyncTaskPool)publicThreadPoolExecutorasyncTaskPool(){returnnewThreadPoolExecutor(5,20,60,TimeUnit.SECONDS,newLinkedBlockingQueue(500),newNamedThreadFactory(async-task),newThreadPoolExecutor.CallerRunsPolicy()// 调用者执行);}}/** * 命名线程工厂 */publicclassNamedThreadFactoryimplementsThreadFactory{privatefinalAtomicIntegerthreadNumbernewAtomicInteger(1);privatefinalStringnamePrefix;publicNamedThreadFactory(StringnamePrefix){this.namePrefixnamePrefix-thread-;}OverridepublicThreadnewThread(Runnabler){ThreadtnewThread(r,namePrefixthreadNumber.getAndIncrement());t.setDaemon(false);t.setPriority(Thread.NORM_PRIORITY);returnt;}}3.2 自定义拒绝策略/** * 自定义拒绝策略记录日志 降级处理 */Slf4jpublicclassCustomRejectedExecutionHandlerimplementsRejectedExecutionHandler{OverridepublicvoidrejectedExecution(Runnabler,ThreadPoolExecutorexecutor){log.error(线程池拒绝执行: pool{}, active{}, queue{}, completed{},executor,executor.getActiveCount(),executor.getQueue().size(),executor.getCompletedTaskCount());// 降级方案1尝试再次入队等待1秒try{if(!executor.getQueue().offer(r,1,TimeUnit.SECONDS)){log.error(再次入队失败执行降级处理);// 降级方案2记录到数据库后续重试saveToRetryQueue(r);}}catch(InterruptedExceptione){Thread.currentThread().interrupt();log.error(等待入队被中断);}}privatevoidsaveToRetryQueue(Runnabler){// 将任务保存到数据库或消息队列后续重试}}3.3 线程池监控/** * 线程池监控 */ComponentSlf4jpublicclassThreadPoolMonitor{AutowiredprivateMapString,ThreadPoolExecutorthreadPools;AutowiredprivateMeterRegistrymeterRegistry;/** * 定时采集线程池指标 */Scheduled(fixedRate5000)publicvoidmonitor(){threadPools.forEach((name,pool)-{// 核心指标gauge(name,active,pool.getActiveCount());gauge(name,pool_size,pool.getPoolSize());gauge(name,queue_size,pool.getQueue().size());gauge(name,completed,pool.getCompletedTaskCount());gauge(name,task_count,pool.getTaskCount());gauge(name,largest_pool_size,pool.getLargestPoolSize());// 告警if(pool.getQueue().size()pool.getQueue().size()*0.8){log.warn(线程池队列使用率超过80%: name{}, queue{}/{},name,pool.getQueue().size(),((LinkedBlockingQueue?)pool.getQueue()).remainingCapacity()pool.getQueue().size());}if(pool.getActiveCount()pool.getMaximumPoolSize()){log.warn(线程池已满: name{}, active{}/{},name,pool.getActiveCount(),pool.getMaximumPoolSize());}});}privatevoidgauge(StringpoolName,Stringmetric,doublevalue){meterRegistry.gauge(thread.pool.metric,Tags.of(pool,poolName),value);}}四、Spring异步线程池4.1 配置/** * Spring异步线程池配置 */ConfigurationEnableAsyncpublicclassAsyncConfigimplementsAsyncConfigurer{OverridepublicExecutorgetAsyncExecutor(){ThreadPoolTaskExecutorexecutornewThreadPoolTaskExecutor();executor.setCorePoolSize(10);executor.setMaxPoolSize(50);executor.setQueueCapacity(500);executor.setKeepAliveSeconds(60);executor.setThreadNamePrefix(async-);executor.setRejectedExecutionHandler(newThreadPoolExecutor.CallerRunsPolicy());executor.setWaitForTasksToCompleteOnShutdown(true);executor.setAwaitTerminationSeconds(60);executor.initialize();returnexecutor;}OverridepublicAsyncUncaughtExceptionHandlergetAsyncUncaughtExceptionHandler(){return(ex,method,params)-{log.error(异步任务执行失败: method{}, params{},method.getName(),Arrays.toString(params),ex);};}}/** * 异步任务使用 */ServiceSlf4jpublicclassAsyncTaskService{/** * 异步发送邮件 */AsyncpublicCompletableFutureVoidsendEmailAsync(Stringto,Stringsubject,Stringcontent){emailService.send(to,subject,content);returnCompletableFuture.completedFuture(null);}/** * 异步处理订单 */Async(orderProcessPool)// 指定线程池publicCompletableFutureOrderResultprocessOrderAsync(Orderorder){OrderResultresultprocessOrder(order);returnCompletableFuture.completedFuture(result);}}五、踩坑实录坑1使用Executors创建线程池newFixedThreadPool和newSingleThreadPool使用无界队列任务堆积导致OOM。解决永远不要用Executors创建线程池用ThreadPoolExecutor。坑2队列太大队列设了10000任务堆积在队列里响应时间越来越长。解决队列容量要适中满了就创建新线程或拒绝。坑3拒绝策略选错用了AbortPolicy任务被拒绝后直接抛异常导致业务中断。解决根据业务选择合适的拒绝策略推荐CallerRunsPolicy或自定义。坑4线程池没有关闭应用关闭时线程池还在运行任务丢失。解决设置waitForTasksToCompleteOnShutdowntrue优雅关闭。坑5ThreadLocal泄漏线程池中的线程是复用的ThreadLocal的值可能残留。解决任务执行前后清理ThreadLocal或使用TransmittableThreadLocal。六、总结线程池使用规范原则说明不用Executors手动创建ThreadPoolExecutor有界队列防止OOM命名线程便于排查问题拒绝策略根据业务选择监控告警及时发现问题优雅关闭防止任务丢失最佳实践线程池参数可配置通过配置中心每个业务使用独立的线程池线程池要有监控和告警压测验证线程池参数定期Review线程池配置血的教训线程池不是配好就完事了。它需要持续监控、持续调优。一个配置不当的线程池比没有线程池更可怕。思考题你的系统线程池是怎么配置的有没有出过问题个人观点仅供参考