1. 线程池基础概念

1.1 什么是线程池

线程池是一种多线程处理形式，它预先创建一定数量的线程，并将这些线程放在一个池子中。当有任务需要处理时，不需要创建新线程，而是将任务提交给线程池，由池中的线程来执行。

线程池的核心优势：

• 减少线程创建销毁的开销：避免频繁创建和销毁线程
• 提高响应速度：任务到达时，无需等待线程创建
• 提高线程的可管理性：统一管理、调优和监控
• 控制并发数量：避免无限制创建线程导致系统资源耗尽

1.2 Java线程池核心参数

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

参数详解：

1. corePoolSize（核心线程数）：

   • 线程池中始终保持的线程数量
   • 即使线程空闲也不会被回收
   • 除非设置了allowCoreThreadTimeOut

2. maximumPoolSize（最大线程数）：

   • 线程池中允许的最大线程数
   • 当工作队列满时，会创建新线程直到达到最大线程数

3. keepAliveTime（空闲存活时间）：

   • 非核心线程的空闲存活时间
   • 超过这个时间的空闲线程会被回收

4. workQueue（工作队列）：

   • 存储等待执行任务的队列
   • 常用类型：ArrayBlockingQueue、LinkedBlockingQueue、SynchronousQueue

5. threadFactory（线程工厂）：

   • 用于创建新线程的工厂
   • 可以自定义线程名称、优先级等属性

6. handler（拒绝策略）：

   • 当线程池和队列都满时的处理策略
   • 默认策略：AbortPolicy（抛出异常）

1.3 线程池执行流程

/*** 线程池任务执行流程演示*/
public class ThreadPoolFlowDemo {public static void main(String[] args) {// 创建线程池：核心2个，最大4个，队列容量3ThreadPoolExecutor executor = new ThreadPoolExecutor(2,                              // 核心线程数4,                              // 最大线程数60L,                            // 空闲存活时间TimeUnit.SECONDS,               // 时间单位new ArrayBlockingQueue<>(3),    // 有界队列，容量3new ThreadFactory() {private AtomicInteger threadNumber = new AtomicInteger(1);@Overridepublic Thread newThread(Runnable r) {Thread t = new Thread(r, "自定义线程-" + threadNumber.getAndIncrement());t.setDaemon(false);return t;}},new ThreadPoolExecutor.CallerRunsPolicy()  // 调用者运行策略);// 提交9个任务，观察执行流程for (int i = 1; i <= 9; i++) {final int taskId = i;try {executor.submit(new Runnable() {@Overridepublic void run() {System.out.println("任务" + taskId + " 开始执行，线程：" + Thread.currentThread().getName());try {Thread.sleep(3000); // 模拟任务耗时} catch (InterruptedException e) {Thread.currentThread().interrupt();}System.out.println("任务" + taskId + " 执行完成");}});// 打印当前线程池状态System.out.println("提交任务" + taskId + " 后 - " +"核心线程数:" + executor.getCorePoolSize() + ", 当前线程数:" + executor.getPoolSize() + ", 队列任务数:" + executor.getQueue().size());Thread.sleep(500); // 稍等片刻观察状态变化} catch (Exception e) {System.err.println("任务" + taskId + " 提交失败: " + e.getMessage());}}// 优雅关闭线程池executor.shutdown();try {if (!executor.awaitTermination(10, TimeUnit.SECONDS)) {executor.shutdownNow();}} catch (InterruptedException e) {executor.shutdownNow();}}
}

---

2. SpringBoot中的线程池

2.1 SpringBoot默认线程池

SpringBoot提供了自动配置的任务执行器，默认配置如下：

/*** SpringBoot默认任务执行器配置* 对应配置类：TaskExecutionAutoConfiguration*/
@Configuration
public class DefaultTaskExecutorConfiguration {@Bean@ConditionalOnMissingBeanpublic TaskExecutor taskExecutor(TaskExecutorBuilder builder) {return builder.build();}// 默认配置等价于：public ThreadPoolTaskExecutor defaultTaskExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setCorePoolSize(8);                    // 核心线程数executor.setMaxPoolSize(Integer.MAX_VALUE);     // 最大线程数executor.setQueueCapacity(Integer.MAX_VALUE);   // 队列容量executor.setKeepAliveSeconds(60);               // 空闲存活时间executor.setThreadNamePrefix("task-");          // 线程名前缀executor.setWaitForTasksToCompleteOnShutdown(false);executor.setAwaitTerminationSeconds(0);executor.initialize();return executor;}
}

默认配置的问题：

• 最大线程数和队列容量都是无限的，可能导致OOM
• 无法区分不同业务的线程池
• 缺乏监控和管理能力

2.2 SpringBoot异步任务基础

/*** SpringBoot异步任务基础示例*/
@Service
public class AsyncTaskService {// 使用@Async注解标记异步方法@Asyncpublic void executeAsyncTask(String taskName) {System.out.println("开始执行异步任务: " + taskName + ", 线程: " + Thread.currentThread().getName());try {// 模拟耗时操作Thread.sleep(2000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}System.out.println("异步任务完成: " + taskName);}// 带返回值的异步方法@Asyncpublic CompletableFuture<String> executeAsyncTaskWithResult(String taskName) {System.out.println("开始执行异步任务: " + taskName + ", 线程: " + Thread.currentThread().getName());try {Thread.sleep(2000);String result = "任务 " + taskName + " 的执行结果";return CompletableFuture.completedFuture(result);} catch (InterruptedException e) {Thread.currentThread().interrupt();return CompletableFuture.completedFuture("任务被中断");}}// 指定线程池的异步方法@Async("customTaskExecutor")public void executeAsyncTaskWithCustomExecutor(String taskName) {System.out.println("使用自定义线程池执行任务: " + taskName + ", 线程: " + Thread.currentThread().getName());try {Thread.sleep(1000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}System.out.println("自定义线程池任务完成: " + taskName);}
}/*** 启用异步支持*/
@Configuration
@EnableAsync  // 必须添加此注解启用异步支持
public class AsyncConfiguration {// 异步配置
}/*** 异步任务测试控制器*/
@RestController
public class AsyncTaskController {@Autowiredprivate AsyncTaskService asyncTaskService;@GetMapping("/async/simple")public String executeSimpleAsyncTask() {System.out.println("Controller开始，线程：" + Thread.currentThread().getName());// 执行异步任务asyncTaskService.executeAsyncTask("简单异步任务");System.out.println("Controller结束，线程：" + Thread.currentThread().getName());return "异步任务已提交";}@GetMapping("/async/with-result")public CompletableFuture<String> executeAsyncTaskWithResult() {System.out.println("Controller开始，线程：" + Thread.currentThread().getName());// 执行带返回值的异步任务CompletableFuture<String> future = asyncTaskService.executeAsyncTaskWithResult("带返回值的任务");// 可以组合多个异步操作return future.thenApply(result -> {System.out.println("处理异步结果，线程：" + Thread.currentThread().getName());return "最终结果：" + result;});}@GetMapping("/async/custom")public String executeAsyncTaskWithCustomExecutor() {System.out.println("Controller开始，线程：" + Thread.currentThread().getName());// 使用指定的线程池执行异步任务asyncTaskService.executeAsyncTaskWithCustomExecutor("自定义线程池任务");System.out.println("Controller结束，线程：" + Thread.currentThread().getName());return "自定义线程池异步任务已提交";}
}

---

3. 自定义线程池配置

3.1 配置文件方式

application.yml配置：

# SpringBoot线程池配置
spring:task:execution:pool:core-size: 8              # 核心线程数max-size: 16              # 最大线程数queue-capacity: 100       # 队列容量keep-alive: 60s           # 空闲线程存活时间thread-name-prefix: "async-task-"  # 线程名前缀shutdown:await-termination: true   # 关闭时等待任务完成await-termination-period: 30s  # 等待时间scheduling:pool:size: 4                   # 定时任务线程池大小thread-name-prefix: "scheduled-task-"# 自定义线程池配置
custom:thread-pool:# 用户相关操作线程池user:core-size: 5max-size: 10queue-capacity: 50keep-alive-seconds: 300thread-name-prefix: "user-task-"# 邮件发送线程池email:core-size: 3max-size: 6queue-capacity: 20keep-alive-seconds: 180thread-name-prefix: "email-task-"# 文件处理线程池file:core-size: 2max-size: 4queue-capacity: 30keep-alive-seconds: 600thread-name-prefix: "file-task-"

配置属性类：

/*** 自定义线程池配置属性*/
@ConfigurationProperties(prefix = "custom.thread-pool")
@Data
public class CustomThreadPoolProperties {private UserPoolConfig user = new UserPoolConfig();private EmailPoolConfig email = new EmailPoolConfig();private FilePoolConfig file = new FilePoolConfig();@Datapublic static class UserPoolConfig {private int coreSize = 5;private int maxSize = 10;private int queueCapacity = 50;private int keepAliveSeconds = 300;private String threadNamePrefix = "user-task-";}@Datapublic static class EmailPoolConfig {private int coreSize = 3;private int maxSize = 6;private int queueCapacity = 20;private int keepAliveSeconds = 180;private String threadNamePrefix = "email-task-";}@Datapublic static class FilePoolConfig {private int coreSize = 2;private int maxSize = 4;private int queueCapacity = 30;private int keepAliveSeconds = 600;private String threadNamePrefix = "file-task-";}
}

3.2 Java配置方式

/*** 自定义线程池配置类*/
@Configuration
@EnableAsync
@EnableConfigurationProperties(CustomThreadPoolProperties.class)
public class ThreadPoolConfiguration {@Autowiredprivate CustomThreadPoolProperties threadPoolProperties;/*** 用户操作线程池*/@Bean("userTaskExecutor")public ThreadPoolTaskExecutor userTaskExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();CustomThreadPoolProperties.UserPoolConfig config = threadPoolProperties.getUser();executor.setCorePoolSize(config.getCoreSize());executor.setMaxPoolSize(config.getMaxSize());executor.setQueueCapacity(config.getQueueCapacity());executor.setKeepAliveSeconds(config.getKeepAliveSeconds());executor.setThreadNamePrefix(config.getThreadNamePrefix());// 拒绝策略：调用者运行executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());// 等待所有任务结束后再关闭线程池executor.setWaitForTasksToCompleteOnShutdown(true);executor.setAwaitTerminationSeconds(30);// 线程池初始化executor.initialize();System.out.println("用户任务线程池初始化完成");return executor;}/*** 邮件发送线程池*/@Bean("emailTaskExecutor")public ThreadPoolTaskExecutor emailTaskExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();CustomThreadPoolProperties.EmailPoolConfig config = threadPoolProperties.getEmail();executor.setCorePoolSize(config.getCoreSize());executor.setMaxPoolSize(config.getMaxSize());executor.setQueueCapacity(config.getQueueCapacity());executor.setKeepAliveSeconds(config.getKeepAliveSeconds());executor.setThreadNamePrefix(config.getThreadNamePrefix());// 拒绝策略：丢弃最老的任务executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());executor.setWaitForTasksToCompleteOnShutdown(true);executor.setAwaitTerminationSeconds(30);executor.initialize();System.out.println("邮件任务线程池初始化完成");return executor;}/*** 文件处理线程池*/@Bean("fileTaskExecutor")public ThreadPoolTaskExecutor fileTaskExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();CustomThreadPoolProperties.FilePoolConfig config = threadPoolProperties.getFile();executor.setCorePoolSize(config.getCoreSize());executor.setMaxPoolSize(config.getMaxSize());executor.setQueueCapacity(config.getQueueCapacity());executor.setKeepAliveSeconds(config.getKeepAliveSeconds());executor.setThreadNamePrefix(config.getThreadNamePrefix());// 自定义拒绝策略executor.setRejectedExecutionHandler(new RejectedExecutionHandler() {@Overridepublic void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {System.err.println("文件处理任务被拒绝执行：" + r.toString());// 可以记录日志、发送告警等}});executor.setWaitForTasksToCompleteOnShutdown(true);executor.setAwaitTerminationSeconds(60); // 文件处理可能需要更长时间executor.initialize();System.out.println("文件任务线程池初始化完成");return executor;}/*** 默认异步任务执行器* 覆盖SpringBoot默认配置*/@Primary@Bean("taskExecutor")public ThreadPoolTaskExecutor taskExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setCorePoolSize(8);executor.setMaxPoolSize(16);executor.setQueueCapacity(100);executor.setKeepAliveSeconds(60);executor.setThreadNamePrefix("default-async-");// 拒绝策略：抛出异常executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());executor.setWaitForTasksToCompleteOnShutdown(true);executor.setAwaitTerminationSeconds(30);executor.initialize();System.out.println("默认异步任务线程池初始化完成");return executor;}
}

3.3 线程池工厂配置

/*** 自定义线程工厂*/
public class CustomThreadFactory implements ThreadFactory {private final AtomicInteger threadNumber = new AtomicInteger(1);private final String namePrefix;private final boolean daemon;private final int priority;public CustomThreadFactory(String namePrefix) {this(namePrefix, false, Thread.NORM_PRIORITY);}public CustomThreadFactory(String namePrefix, boolean daemon, int priority) {this.namePrefix = namePrefix;this.daemon = daemon;this.priority = priority;}@Overridepublic Thread newThread(Runnable r) {Thread t = new Thread(r, namePrefix + threadNumber.getAndIncrement());t.setDaemon(daemon);t.setPriority(priority);// 设置未捕获异常处理器t.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {@Overridepublic void uncaughtException(Thread t, Throwable e) {System.err.println("线程 " + t.getName() + " 发生未捕获异常：");e.printStackTrace();// 可以添加日志记录、告警等逻辑// logger.error("线程异常", e);// alertService.sendAlert("线程异常", e.getMessage());}});return t;}
}/*** 使用自定义线程工厂的线程池配置*/
@Configuration
public class AdvancedThreadPoolConfiguration {/*** 高优先级任务线程池*/@Bean("highPriorityTaskExecutor")public ThreadPoolTaskExecutor highPriorityTaskExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setCorePoolSize(4);executor.setMaxPoolSize(8);executor.setQueueCapacity(50);executor.setKeepAliveSeconds(120);// 使用自定义线程工厂executor.setThreadFactory(new CustomThreadFactory("high-priority-", false, Thread.MAX_PRIORITY));// 自定义拒绝策略：记录日志并抛出异常executor.setRejectedExecutionHandler(new RejectedExecutionHandler() {@Overridepublic void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {String msg = String.format("高优先级任务被拒绝执行，当前线程数：%d，队列大小：%d",executor.getPoolSize(), executor.getQueue().size());System.err.println(msg);throw new RejectedExecutionException(msg);}});executor.initialize();return executor;}/*** CPU密集型任务线程池*/@Bean("cpuIntensiveTaskExecutor")public ThreadPoolTaskExecutor cpuIntensiveTaskExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();// CPU密集型任务：线程数通常设置为CPU核心数int cpuCores = Runtime.getRuntime().availableProcessors();executor.setCorePoolSize(cpuCores);executor.setMaxPoolSize(cpuCores);executor.setQueueCapacity(100);executor.setKeepAliveSeconds(60);executor.setThreadFactory(new CustomThreadFactory("cpu-intensive-", false, Thread.NORM_PRIORITY));executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());executor.initialize();return executor;}/*** IO密集型任务线程池*/@Bean("ioIntensiveTaskExecutor")public ThreadPoolTaskExecutor ioIntensiveTaskExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();// IO密集型任务：线程数可以设置为CPU核心数的2倍或更多int cpuCores = Runtime.getRuntime().availableProcessors();executor.setCorePoolSize(cpuCores * 2);executor.setMaxPoolSize(cpuCores * 4);executor.setQueueCapacity(200);executor.setKeepAliveSeconds(300);executor.setThreadFactory(new CustomThreadFactory("io-intensive-", false, Thread.NORM_PRIORITY));executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());executor.initialize();return executor;}
} 

---

4. 异步任务实际应用

4.1 业务服务中的异步任务

/*** 用户服务 - 演示用户相关的异步操作*/
@Service
public class UserService {@Autowiredprivate UserRepository userRepository;@Autowiredprivate NotificationService notificationService;@Autowiredprivate AuditService auditService;/*** 用户注册 - 同步保存用户，异步发送通知和记录审计*/@Transactionalpublic User registerUser(UserRegistrationDto dto) {System.out.println("开始用户注册，线程：" + Thread.currentThread().getName());// 同步保存用户信息User user = new User();user.setUsername(dto.getUsername());user.setEmail(dto.getEmail());user.setCreateTime(LocalDateTime.now());user = userRepository.save(user);System.out.println("用户保存完成：" + user.getId());// 异步发送欢迎邮件notificationService.sendWelcomeEmail(user);// 异步记录审计日志auditService.recordUserRegistration(user);// 异步更新用户统计updateUserStatistics(user);return user;}/*** 异步更新用户统计信息*/@Async("userTaskExecutor")public void updateUserStatistics(User user) {System.out.println("开始更新用户统计，线程：" + Thread.currentThread().getName());try {// 模拟复杂的统计计算Thread.sleep(2000);// 更新统计信息// statisticsService.updateUserCount();// statisticsService.updateRegionStatistics(user.getRegion());System.out.println("用户统计更新完成：" + user.getId());} catch (InterruptedException e) {Thread.currentThread().interrupt();System.err.println("用户统计更新被中断：" + user.getId());} catch (Exception e) {System.err.println("用户统计更新失败：" + user.getId() + ", 错误：" + e.getMessage());}}/*** 批量用户数据处理*/@Async("userTaskExecutor")public CompletableFuture<BatchProcessResult> batchProcessUsers(List<Long> userIds) {System.out.println("开始批量处理用户，线程：" + Thread.currentThread().getName());BatchProcessResult result = new BatchProcessResult();result.setTotalCount(userIds.size());for (Long userId : userIds) {try {// 模拟处理单个用户Thread.sleep(100);// 处理用户逻辑// User user = userRepository.findById(userId);// processUserData(user);result.incrementSuccessCount();} catch (Exception e) {result.incrementFailureCount();result.addError("用户 " + userId + " 处理失败：" + e.getMessage());}}System.out.println("批量处理完成，成功：" + result.getSuccessCount() + "，失败：" + result.getFailureCount());return CompletableFuture.completedFuture(result);}
}/*** 通知服务 - 演示邮件和消息推送的异步处理*/
@Service
public class NotificationService {/*** 异步发送欢迎邮件*/@Async("emailTaskExecutor")public void sendWelcomeEmail(User user) {System.out.println("开始发送欢迎邮件，线程：" + Thread.currentThread().getName());try {// 模拟邮件发送耗时Thread.sleep(3000);// 发送邮件逻辑String subject = "欢迎注册";String content = "亲爱的 " + user.getUsername() + "，欢迎您的加入！";// emailSender.send(user.getEmail(), subject, content);System.out.println("欢迎邮件发送成功：" + user.getEmail());} catch (InterruptedException e) {Thread.currentThread().interrupt();System.err.println("邮件发送被中断：" + user.getEmail());} catch (Exception e) {System.err.println("邮件发送失败：" + user.getEmail() + ", 错误：" + e.getMessage());// 可以加入重试逻辑或放入死信队列}}/*** 异步批量发送邮件*/@Async("emailTaskExecutor")public CompletableFuture<Void> sendBatchEmails(List<EmailTask> emailTasks) {System.out.println("开始批量发送邮件，数量：" + emailTasks.size() + "，线程：" + Thread.currentThread().getName());List<CompletableFuture<Void>> futures = new ArrayList<>();for (EmailTask task : emailTasks) {CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {try {Thread.sleep(1000); // 模拟发送耗时System.out.println("邮件发送成功：" + task.getTo());} catch (InterruptedException e) {Thread.currentThread().interrupt();}});futures.add(future);}// 等待所有邮件发送完成CompletableFuture<Void> allFutures = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));return allFutures.thenRun(() -> {System.out.println("批量邮件发送完成");});}/*** 异步发送推送通知*/@Async("userTaskExecutor")public void sendPushNotification(String userId, String message) {System.out.println("开始发送推送通知，线程：" + Thread.currentThread().getName());try {// 模拟推送耗时Thread.sleep(500);// 推送逻辑// pushService.send(userId, message);System.out.println("推送通知发送成功：" + userId);} catch (Exception e) {System.err.println("推送通知发送失败：" + userId + ", 错误：" + e.getMessage());}}
}/*** 文件处理服务 - 演示文件上传、转换等异步操作*/
@Service
public class FileProcessingService {/*** 异步处理文件上传*/@Async("fileTaskExecutor")public CompletableFuture<FileProcessResult> processUploadedFile(MultipartFile file) {System.out.println("开始处理上传文件，线程：" + Thread.currentThread().getName());FileProcessResult result = new FileProcessResult();result.setOriginalFilename(file.getOriginalFilename());try {// 文件验证validateFile(file);// 保存文件String savedPath = saveFile(file);result.setSavedPath(savedPath);// 生成缩略图（如果是图片）if (isImageFile(file)) {String thumbnailPath = generateThumbnail(savedPath);result.setThumbnailPath(thumbnailPath);}// 提取文件信息extractFileMetadata(savedPath, result);result.setSuccess(true);System.out.println("文件处理完成：" + file.getOriginalFilename());} catch (Exception e) {result.setSuccess(false);result.setErrorMessage(e.getMessage());System.err.println("文件处理失败：" + file.getOriginalFilename() + ", 错误：" + e.getMessage());}return CompletableFuture.completedFuture(result);}/*** 异步批量处理文件*/@Async("fileTaskExecutor")public void batchProcessFiles(List<String> filePaths) {System.out.println("开始批量处理文件，数量：" + filePaths.size() + "，线程：" + Thread.currentThread().getName());for (String filePath : filePaths) {try {// 处理单个文件processFile(filePath);System.out.println("文件处理完成：" + filePath);} catch (Exception e) {System.err.println("文件处理失败：" + filePath + ", 错误：" + e.getMessage());}}System.out.println("批量文件处理完成");}private void validateFile(MultipartFile file) throws Exception {if (file.isEmpty()) {throw new Exception("文件为空");}if (file.getSize() > 10 * 1024 * 1024) { // 10MBthrow new Exception("文件大小超过限制");}}private String saveFile(MultipartFile file) throws Exception {// 模拟文件保存耗时Thread.sleep(1000);return "/uploads/" + System.currentTimeMillis() + "_" + file.getOriginalFilename();}private boolean isImageFile(MultipartFile file) {String contentType = file.getContentType();return contentType != null && contentType.startsWith("image/");}private String generateThumbnail(String imagePath) throws Exception {// 模拟缩略图生成耗时Thread.sleep(2000);return imagePath.replace(".", "_thumb.");}private void extractFileMetadata(String filePath, FileProcessResult result) throws Exception {// 模拟元数据提取耗时Thread.sleep(500);result.setFileSize(1024L * 1024L); // 模拟文件大小result.setFileType("document");}private void processFile(String filePath) throws Exception {// 模拟文件处理耗时Thread.sleep(2000);}
}

4.2 定时任务与异步任务结合

/*** 定时任务服务 - 演示定时任务触发异步操作*/
@Service
public class ScheduledTaskService {@Autowiredprivate UserService userService;@Autowiredprivate NotificationService notificationService;@Autowiredprivate FileProcessingService fileProcessingService;/*** 每天凌晨2点执行用户数据统计*/@Scheduled(cron = "0 0 2 * * ?")public void dailyUserStatistics() {System.out.println("开始执行每日用户统计任务，线程：" + Thread.currentThread().getName());try {// 获取昨天注册的用户List<Long> yesterdayUserIds = getUserIdsRegisteredYesterday();if (!yesterdayUserIds.isEmpty()) {// 异步处理用户统计CompletableFuture<BatchProcessResult> future = userService.batchProcessUsers(yesterdayUserIds);// 可以设置回调处理结果future.thenAccept(result -> {System.out.println("每日用户统计完成，处理用户数：" + result.getTotalCount());// 发送统计报告邮件sendStatisticsReport(result);}).exceptionally(throwable -> {System.err.println("每日用户统计失败：" + throwable.getMessage());return null;});}} catch (Exception e) {System.err.println("定时任务执行失败：" + e.getMessage());}}/*** 每小时清理临时文件*/@Scheduled(fixedRate = 3600000) // 每小时执行一次public void cleanupTempFiles() {System.out.println("开始清理临时文件，线程：" + Thread.currentThread().getName());try {// 获取需要清理的文件列表List<String> tempFiles = getTempFiles();if (!tempFiles.isEmpty()) {// 异步清理文件fileProcessingService.batchProcessFiles(tempFiles);}} catch (Exception e) {System.err.println("清理临时文件失败：" + e.getMessage());}}/*** 每周发送汇总邮件*/@Scheduled(cron = "0 0 9 ? * MON") // 每周一上午9点public void sendWeeklySummary() {System.out.println("开始发送周报邮件，线程：" + Thread.currentThread().getName());try {// 生成周报数据WeeklySummary summary = generateWeeklySummary();// 获取需要接收周报的用户List<EmailTask> emailTasks = createWeeklySummaryEmails(summary);// 异步批量发送邮件CompletableFuture<Void> future = notificationService.sendBatchEmails(emailTasks);future.thenRun(() -> {System.out.println("周报邮件发送完成");}).exceptionally(throwable -> {System.err.println("周报邮件发送失败：" + throwable.getMessage());return null;});} catch (Exception e) {System.err.println("周报邮件任务失败：" + e.getMessage());}}private List<Long> getUserIdsRegisteredYesterday() {// 模拟获取昨天注册的用户IDreturn Arrays.asList(1L, 2L, 3L, 4L, 5L);}private List<String> getTempFiles() {// 模拟获取临时文件列表return Arrays.asList("/tmp/file1.tmp", "/tmp/file2.tmp", "/tmp/file3.tmp");}private WeeklySummary generateWeeklySummary() {// 模拟生成周报数据WeeklySummary summary = new WeeklySummary();summary.setNewUsers(100);summary.setActiveUsers(500);summary.setTotalRevenue(10000.0);return summary;}private List<EmailTask> createWeeklySummaryEmails(WeeklySummary summary) {// 模拟创建邮件任务List<EmailTask> tasks = new ArrayList<>();tasks.add(new EmailTask("admin@example.com", "周报", "周报内容"));tasks.add(new EmailTask("manager@example.com", "周报", "周报内容"));return tasks;}private void sendStatisticsReport(BatchProcessResult result) {try {notificationService.sendPushNotification("admin", "用户统计完成：" + result.getSuccessCount());} catch (Exception e) {System.err.println("发送统计报告失败：" + e.getMessage());}}
}

---

5. 线程池监控和管理

5.1 线程池状态监控

/*** 线程池监控服务*/
@Service
public class ThreadPoolMonitorService {@Autowired@Qualifier("userTaskExecutor")private ThreadPoolTaskExecutor userTaskExecutor;@Autowired@Qualifier("emailTaskExecutor")private ThreadPoolTaskExecutor emailTaskExecutor;@Autowired@Qualifier("fileTaskExecutor")private ThreadPoolTaskExecutor fileTaskExecutor;/*** 获取线程池状态信息*/public ThreadPoolStatus getThreadPoolStatus(String executorName) {ThreadPoolTaskExecutor executor = getExecutor(executorName);if (executor == null) {return null;}ThreadPoolExecutor threadPoolExecutor = executor.getThreadPoolExecutor();ThreadPoolStatus status = new ThreadPoolStatus();status.setExecutorName(executorName);status.setCorePoolSize(threadPoolExecutor.getCorePoolSize());status.setMaximumPoolSize(threadPoolExecutor.getMaximumPoolSize());status.setCurrentPoolSize(threadPoolExecutor.getPoolSize());status.setActiveThreadCount(threadPoolExecutor.getActiveCount());status.setQueueSize(threadPoolExecutor.getQueue().size());status.setQueueCapacity(getQueueCapacity(threadPoolExecutor));status.setCompletedTaskCount(threadPoolExecutor.getCompletedTaskCount());status.setTotalTaskCount(threadPoolExecutor.getTaskCount());status.setKeepAliveTime(threadPoolExecutor.getKeepAliveTime(TimeUnit.SECONDS));status.setRejectedExecutionCount(getRejectedExecutionCount(threadPoolExecutor));// 计算使用率status.setPoolUsageRate(calculatePoolUsageRate(threadPoolExecutor));status.setQueueUsageRate(calculateQueueUsageRate(threadPoolExecutor));return status;}/*** 获取所有线程池状态*/public List<ThreadPoolStatus> getAllThreadPoolStatus() {List<ThreadPoolStatus> statusList = new ArrayList<>();statusList.add(getThreadPoolStatus("userTaskExecutor"));statusList.add(getThreadPoolStatus("emailTaskExecutor"));statusList.add(getThreadPoolStatus("fileTaskExecutor"));return statusList;}/*** 动态调整线程池参数*/public boolean adjustThreadPool(String executorName, ThreadPoolAdjustment adjustment) {ThreadPoolTaskExecutor executor = getExecutor(executorName);if (executor == null) {return false;}ThreadPoolExecutor threadPoolExecutor = executor.getThreadPoolExecutor();try {if (adjustment.getCorePoolSize() != null) {threadPoolExecutor.setCorePoolSize(adjustment.getCorePoolSize());}if (adjustment.getMaximumPoolSize() != null) {threadPoolExecutor.setMaximumPoolSize(adjustment.getMaximumPoolSize());}if (adjustment.getKeepAliveTime() != null) {threadPoolExecutor.setKeepAliveTime(adjustment.getKeepAliveTime(), TimeUnit.SECONDS);}System.out.println("线程池参数调整成功：" + executorName);return true;} catch (Exception e) {System.err.println("线程池参数调整失败：" + executorName + ", 错误：" + e.getMessage());return false;}}/*** 优雅关闭线程池*/public boolean shutdownThreadPool(String executorName) {ThreadPoolTaskExecutor executor = getExecutor(executorName);if (executor == null) {return false;}try {executor.shutdown();// 等待任务完成if (!executor.getThreadPoolExecutor().awaitTermination(30, TimeUnit.SECONDS)) {executor.getThreadPoolExecutor().shutdownNow();}System.out.println("线程池优雅关闭成功：" + executorName);return true;} catch (InterruptedException e) {executor.getThreadPoolExecutor().shutdownNow();Thread.currentThread().interrupt();System.err.println("线程池关闭被中断：" + executorName);return false;}}private ThreadPoolTaskExecutor getExecutor(String executorName) {switch (executorName) {case "userTaskExecutor":return userTaskExecutor;case "emailTaskExecutor":return emailTaskExecutor;case "fileTaskExecutor":return fileTaskExecutor;default:return null;}}private int getQueueCapacity(ThreadPoolExecutor executor) {if (executor.getQueue() instanceof ArrayBlockingQueue) {return ((ArrayBlockingQueue<?>) executor.getQueue()).size() + ((ArrayBlockingQueue<?>) executor.getQueue()).remainingCapacity();}return -1; // 无界队列}private long getRejectedExecutionCount(ThreadPoolExecutor executor) {// 这里需要自定义RejectedExecutionHandler来统计拒绝次数// 或者使用JMX监控return 0;}private double calculatePoolUsageRate(ThreadPoolExecutor executor) {return (double) executor.getActiveCount() / executor.getMaximumPoolSize() * 100;}private double calculateQueueUsageRate(ThreadPoolExecutor executor) {int capacity = getQueueCapacity(executor);if (capacity <= 0) {return 0;}return (double) executor.getQueue().size() / capacity * 100;}
}/*** 线程池状态数据类*/
@Data
public class ThreadPoolStatus {private String executorName;private int corePoolSize;private int maximumPoolSize;private int currentPoolSize;private int activeThreadCount;private int queueSize;private int queueCapacity;private long completedTaskCount;private long totalTaskCount;private long keepAliveTime;private long rejectedExecutionCount;private double poolUsageRate;private double queueUsageRate;private LocalDateTime updateTime = LocalDateTime.now();
}/*** 线程池调整参数类*/
@Data
public class ThreadPoolAdjustment {private Integer corePoolSize;private Integer maximumPoolSize;private Long keepAliveTime;
}

5.2 监控API接口

/*** 线程池监控控制器*/
@RestController
@RequestMapping("/api/threadpool")
public class ThreadPoolMonitorController {@Autowiredprivate ThreadPoolMonitorService monitorService;/*** 获取所有线程池状态*/@GetMapping("/status")public ResponseEntity<List<ThreadPoolStatus>> getAllThreadPoolStatus() {List<ThreadPoolStatus> statusList = monitorService.getAllThreadPoolStatus();return ResponseEntity.ok(statusList);}/*** 获取指定线程池状态*/@GetMapping("/status/{executorName}")public ResponseEntity<ThreadPoolStatus> getThreadPoolStatus(@PathVariable String executorName) {ThreadPoolStatus status = monitorService.getThreadPoolStatus(executorName);if (status == null) {return ResponseEntity.notFound().build();}return ResponseEntity.ok(status);}/*** 调整线程池参数*/@PostMapping("/adjust/{executorName}")public ResponseEntity<String> adjustThreadPool(@PathVariable String executorName,@RequestBody ThreadPoolAdjustment adjustment) {boolean success = monitorService.adjustThreadPool(executorName, adjustment);if (success) {return ResponseEntity.ok("线程池参数调整成功");} else {return ResponseEntity.badRequest().body("线程池参数调整失败");}}/*** 关闭线程池*/@PostMapping("/shutdown/{executorName}")public ResponseEntity<String> shutdownThreadPool(@PathVariable String executorName) {boolean success = monitorService.shutdownThreadPool(executorName);if (success) {return ResponseEntity.ok("线程池关闭成功");} else {return ResponseEntity.badRequest().body("线程池关闭失败");}}/*** 线程池健康检查*/@GetMapping("/health")public ResponseEntity<Map<String, Object>> healthCheck() {List<ThreadPoolStatus> statusList = monitorService.getAllThreadPoolStatus();Map<String, Object> healthInfo = new HashMap<>();boolean allHealthy = true;for (ThreadPoolStatus status : statusList) {boolean healthy = isThreadPoolHealthy(status);healthInfo.put(status.getExecutorName(), healthy ? "健康" : "异常");if (!healthy) {allHealthy = false;}}healthInfo.put("overall", allHealthy ? "健康" : "异常");return ResponseEntity.ok(healthInfo);}private boolean isThreadPoolHealthy(ThreadPoolStatus status) {// 定义健康标准return status.getPoolUsageRate() < 90 &&  // 线程池使用率不超过90%status.getQueueUsageRate() < 80 &&  // 队列使用率不超过80%status.getRejectedExecutionCount() == 0;  // 没有任务被拒绝}
}

5.3 线程池告警机制

/*** 线程池告警服务*/
@Service
public class ThreadPoolAlertService {@Autowiredprivate ThreadPoolMonitorService monitorService;@Autowiredprivate NotificationService notificationService;// 告警阈值配置private static final double POOL_USAGE_THRESHOLD = 85.0;private static final double QUEUE_USAGE_THRESHOLD = 75.0;private static final long REJECTED_TASK_THRESHOLD = 10;/*** 定期检查线程池状态并发送告警*/@Scheduled(fixedRate = 60000) // 每分钟检查一次public void checkThreadPoolStatus() {List<ThreadPoolStatus> statusList = monitorService.getAllThreadPoolStatus();for (ThreadPoolStatus status : statusList) {checkAndAlert(status);}}private void checkAndAlert(ThreadPoolStatus status) {List<String> alerts = new ArrayList<>();// 检查线程池使用率if (status.getPoolUsageRate() > POOL_USAGE_THRESHOLD) {alerts.add(String.format("线程池使用率过高：%.2f%%", status.getPoolUsageRate()));}// 检查队列使用率if (status.getQueueUsageRate() > QUEUE_USAGE_THRESHOLD) {alerts.add(String.format("队列使用率过高：%.2f%%", status.getQueueUsageRate()));}// 检查拒绝任务数if (status.getRejectedExecutionCount() > REJECTED_TASK_THRESHOLD) {alerts.add(String.format("拒绝任务数过多：%d", status.getRejectedExecutionCount()));}// 检查活跃线程数是否接近最大值if (status.getActiveThreadCount() >= status.getMaximumPoolSize() * 0.9) {alerts.add("活跃线程数接近最大值");}// 发送告警if (!alerts.isEmpty()) {sendAlert(status.getExecutorName(), alerts);}}private void sendAlert(String executorName, List<String> alerts) {String alertMessage = String.format("线程池告警 [%s]:\n%s", executorName, String.join("\n", alerts));System.err.println(alertMessage);// 发送告警通知try {notificationService.sendPushNotification("admin", alertMessage);// 也可以发送邮件告警// notificationService.sendAlertEmail("admin@example.com", "线程池告警", alertMessage);} catch (Exception e) {System.err.println("发送告警失败：" + e.getMessage());}}
}/*** 线程池性能统计服务*/
@Service
public class ThreadPoolMetricsService {private final Map<String, List<ThreadPoolMetrics>> metricsHistory = new ConcurrentHashMap<>();private final int MAX_HISTORY_SIZE = 100; // 保留最近100条记录@Autowiredprivate ThreadPoolMonitorService monitorService;/*** 定期收集线程池指标*/@Scheduled(fixedRate = 30000) // 每30秒收集一次public void collectMetrics() {List<ThreadPoolStatus> statusList = monitorService.getAllThreadPoolStatus();for (ThreadPoolStatus status : statusList) {ThreadPoolMetrics metrics = new ThreadPoolMetrics();metrics.setTimestamp(LocalDateTime.now());metrics.setPoolUsageRate(status.getPoolUsageRate());metrics.setQueueUsageRate(status.getQueueUsageRate());metrics.setActiveThreadCount(status.getActiveThreadCount());metrics.setCompletedTaskCount(status.getCompletedTaskCount());addMetrics(status.getExecutorName(), metrics);}}private void addMetrics(String executorName, ThreadPoolMetrics metrics) {metricsHistory.computeIfAbsent(executorName, k -> new ArrayList<>()).add(metrics);// 保持历史记录在限制范围内List<ThreadPoolMetrics> history = metricsHistory.get(executorName);if (history.size() > MAX_HISTORY_SIZE) {history.remove(0);}}/*** 获取线程池历史指标*/public List<ThreadPoolMetrics> getMetricsHistory(String executorName) {return metricsHistory.getOrDefault(executorName, new ArrayList<>());}/*** 获取线程池性能报告*/public ThreadPoolPerformanceReport getPerformanceReport(String executorName) {List<ThreadPoolMetrics> history = getMetricsHistory(executorName);if (history.isEmpty()) {return null;}ThreadPoolPerformanceReport report = new ThreadPoolPerformanceReport();report.setExecutorName(executorName);report.setReportTime(LocalDateTime.now());// 计算平均值double avgPoolUsage = history.stream().mapToDouble(ThreadPoolMetrics::getPoolUsageRate).average().orElse(0);double avgQueueUsage = history.stream().mapToDouble(ThreadPoolMetrics::getQueueUsageRate).average().orElse(0);double avgActiveThreads = history.stream().mapToDouble(ThreadPoolMetrics::getActiveThreadCount).average().orElse(0);report.setAvgPoolUsageRate(avgPoolUsage);report.setAvgQueueUsageRate(avgQueueUsage);report.setAvgActiveThreadCount(avgActiveThreads);// 计算峰值double maxPoolUsage = history.stream().mapToDouble(ThreadPoolMetrics::getPoolUsageRate).max().orElse(0);double maxQueueUsage = history.stream().mapToDouble(ThreadPoolMetrics::getQueueUsageRate).max().orElse(0);int maxActiveThreads = history.stream().mapToInt(ThreadPoolMetrics::getActiveThreadCount).max().orElse(0);report.setMaxPoolUsageRate(maxPoolUsage);report.setMaxQueueUsageRate(maxQueueUsage);report.setMaxActiveThreadCount(maxActiveThreads);// 计算任务完成速率if (history.size() >= 2) {ThreadPoolMetrics first = history.get(0);ThreadPoolMetrics last = history.get(history.size() - 1);long timeDiff = Duration.between(first.getTimestamp(), last.getTimestamp()).getSeconds();long taskDiff = last.getCompletedTaskCount() - first.getCompletedTaskCount();if (timeDiff > 0) {report.setTaskCompletionRate(taskDiff / (double) timeDiff);}}return report;}
}/*** 线程池指标数据类*/
@Data
public class ThreadPoolMetrics {private LocalDateTime timestamp;private double poolUsageRate;private double queueUsageRate;private int activeThreadCount;private long completedTaskCount;
}/*** 线程池性能报告类*/
@Data
public class ThreadPoolPerformanceReport {private String executorName;private LocalDateTime reportTime;private double avgPoolUsageRate;private double avgQueueUsageRate;private double avgActiveThreadCount;private double maxPoolUsageRate;private double maxQueueUsageRate;private int maxActiveThreadCount;private double taskCompletionRate; // 任务/秒
} 

---

6. 最佳实践和性能优化

6.1 线程池参数设置最佳实践

/*** 线程池参数优化指导*/
@Component
public class ThreadPoolOptimizationGuide {/*** CPU密集型任务线程池配置* 特点：主要进行计算操作，很少阻塞* 推荐线程数：CPU核心数 或 CPU核心数 + 1*/@Bean("cpuIntensiveExecutor")public ThreadPoolTaskExecutor createCpuIntensiveExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();int cpuCores = Runtime.getRuntime().availableProcessors();executor.setCorePoolSize(cpuCores);executor.setMaxPoolSize(cpuCores);executor.setQueueCapacity(50);executor.setKeepAliveSeconds(60);executor.setThreadNamePrefix("cpu-task-");executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());System.out.println("CPU密集型线程池配置 - 核心线程数: " + cpuCores);return executor;}/*** IO密集型任务线程池配置* 特点：大量IO操作，线程经常阻塞等待* 推荐线程数：CPU核心数 * 2 到 CPU核心数 * 4*/@Bean("ioIntensiveExecutor")public ThreadPoolTaskExecutor createIoIntensiveExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();int cpuCores = Runtime.getRuntime().availableProcessors();int corePoolSize = cpuCores * 2;int maxPoolSize = cpuCores * 4;executor.setCorePoolSize(corePoolSize);executor.setMaxPoolSize(maxPoolSize);executor.setQueueCapacity(200);executor.setKeepAliveSeconds(300);executor.setThreadNamePrefix("io-task-");executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());System.out.println("IO密集型线程池配置 - 核心线程数: " + corePoolSize + ", 最大线程数: " + maxPoolSize);return executor;}/*** 混合型任务线程池配置* 特点：既有计算又有IO操作* 推荐线程数：根据IO等待时间和CPU使用时间的比例调整*/@Bean("mixedTaskExecutor")public ThreadPoolTaskExecutor createMixedTaskExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();int cpuCores = Runtime.getRuntime().availableProcessors();// 假设IO等待时间占80%，CPU使用时间占20%// 线程数 = CPU核心数 / (1 - 阻塞系数) = CPU核心数 / (1 - 0.8) = CPU核心数 * 5int optimalThreads = Math.min(cpuCores * 5, 50); // 设置上限避免过多线程executor.setCorePoolSize(cpuCores * 2);executor.setMaxPoolSize(optimalThreads);executor.setQueueCapacity(100);executor.setKeepAliveSeconds(180);executor.setThreadNamePrefix("mixed-task-");executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());System.out.println("混合型线程池配置 - 核心线程数: " + (cpuCores * 2) + ", 最大线程数: " + optimalThreads);return executor;}/*** 动态线程池配置* 根据系统负载自动调整参数*/@PostConstructpublic void setupDynamicAdjustment() {// 定期检查系统负载并调整线程池参数ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);scheduler.scheduleAtFixedRate(this::adjustThreadPoolBasedOnLoad, 5, 30, TimeUnit.MINUTES);}private void adjustThreadPoolBasedOnLoad() {try {// 获取系统负载OperatingSystemMXBean osBean = ManagementFactory.getOperatingSystemMXBean();double systemLoad = osBean.getSystemLoadAverage();int availableProcessors = osBean.getAvailableProcessors();// 获取JVM内存使用情况MemoryMXBean memoryBean = ManagementFactory.getMemoryMXBean();MemoryUsage heapUsage = memoryBean.getHeapMemoryUsage();double memoryUsageRate = (double) heapUsage.getUsed() / heapUsage.getMax();System.out.println("系统负载: " + systemLoad + ", 内存使用率: " + (memoryUsageRate * 100) + "%");// 根据负载调整线程池参数的逻辑if (systemLoad > availableProcessors * 0.8) {System.out.println("系统负载过高，建议减少线程池大小");} else if (systemLoad < availableProcessors * 0.3) {System.out.println("系统负载较低，可以适当增加线程池大小");}if (memoryUsageRate > 0.85) {System.out.println("内存使用率过高，建议减少线程池大小或队列容量");}} catch (Exception e) {System.err.println("动态调整线程池参数失败: " + e.getMessage());}}
}

6.2 性能优化策略

/*** 线程池性能优化策略*/
@Service
public class ThreadPoolPerformanceOptimizer {/*** 预热线程池* 在应用启动时预先创建核心线程*/@EventListener(ApplicationReadyEvent.class)public void preWarmThreadPools() {System.out.println("开始预热线程池...");List<ThreadPoolTaskExecutor> executors = getAll ThreadPoolExecutors();for (ThreadPoolTaskExecutor executor : executors) {ThreadPoolExecutor threadPoolExecutor = executor.getThreadPoolExecutor();// 预启动所有核心线程int preStarted = threadPoolExecutor.prestartAllCoreThreads();System.out.println("线程池 " + executor.getThreadNamePrefix() + " 预启动了 " + preStarted + " 个核心线程");}System.out.println("线程池预热完成");}/*** 任务分片处理* 将大任务分解为小任务，提高并行度*/@Async("ioIntensiveExecutor")public CompletableFuture<List<String>> processBatchDataWithSharding(List<String> data) {int shardSize = 100; // 每个分片处理100条数据List<List<String>> shards = createShards(data, shardSize);List<CompletableFuture<List<String>>> futures = new ArrayList<>();for (List<String> shard : shards) {CompletableFuture<List<String>> future = CompletableFuture.supplyAsync(() -> {return processShard(shard);});futures.add(future);}// 等待所有分片处理完成CompletableFuture<Void> allFutures = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));return allFutures.thenApply(v -> {List<String> result = new ArrayList<>();for (CompletableFuture<List<String>> future : futures) {try {result.addAll(future.get());} catch (Exception e) {System.err.println("分片处理失败: " + e.getMessage());}}return result;});}/*** 任务优先级队列* 使用PriorityBlockingQueue实现任务优先级*/@Bean("priorityTaskExecutor")public ThreadPoolTaskExecutor createPriorityTaskExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setCorePoolSize(4);executor.setMaxPoolSize(8);executor.setKeepAliveSeconds(60);executor.setThreadNamePrefix("priority-task-");// 使用优先级队列PriorityBlockingQueue<Runnable> priorityQueue = new PriorityBlockingQueue<>(50, new TaskPriorityComparator());// 需要自定义ThreadPoolExecutor来使用优先级队列ThreadPoolExecutor customExecutor = new ThreadPoolExecutor(4, 8, 60L, TimeUnit.SECONDS, priorityQueue);// 这里需要通过反射或其他方式设置自定义executor// 实际应用中可能需要完全自定义ThreadPoolTaskExecutorreturn executor;}/*** 任务执行时间监控* 监控任务执行时间，识别慢任务*/@Async("monitoredTaskExecutor")public void executeMonitoredTask(String taskName, Runnable task) {long startTime = System.currentTimeMillis();String threadName = Thread.currentThread().getName();try {System.out.println("任务开始: " + taskName + ", 线程: " + threadName);task.run();} catch (Exception e) {System.err.println("任务执行失败: " + taskName + ", 错误: " + e.getMessage());throw e;} finally {long endTime = System.currentTimeMillis();long duration = endTime - startTime;System.out.println("任务完成: " + taskName + ", 耗时: " + duration + "ms, 线程: " + threadName);// 记录慢任务if (duration > 5000) { // 超过5秒的任务System.out.println("检测到慢任务: " + taskName + ", 耗时: " + duration + "ms");// 可以发送告警或记录日志}}}private List<ThreadPoolTaskExecutor> getAllThreadPoolExecutors() {// 这里需要获取所有的ThreadPoolTaskExecutor实例// 可以通过ApplicationContext获取return new ArrayList<>();}private List<List<String>> createShards(List<String> data, int shardSize) {List<List<String>> shards = new ArrayList<>();for (int i = 0; i < data.size(); i += shardSize) {int endIndex = Math.min(i + shardSize, data.size());shards.add(data.subList(i, endIndex));}return shards;}private List<String> processShard(List<String> shard) {// 模拟分片处理逻辑List<String> result = new ArrayList<>();for (String item : shard) {result.add("processed-" + item);}return result;}
}/*** 任务优先级比较器*/
public class TaskPriorityComparator implements Comparator<Runnable> {@Overridepublic int compare(Runnable r1, Runnable r2) {if (r1 instanceof PriorityTask && r2 instanceof PriorityTask) {PriorityTask task1 = (PriorityTask) r1;PriorityTask task2 = (PriorityTask) r2;return Integer.compare(task2.getPriority(), task1.getPriority()); // 高优先级在前}return 0;}
}/*** 带优先级的任务*/
public class PriorityTask implements Runnable {private final Runnable task;private final int priority;public PriorityTask(Runnable task, int priority) {this.task = task;this.priority = priority;}@Overridepublic void run() {task.run();}public int getPriority() {return priority;}
}

6.3 错误处理和恢复机制

/*** 线程池错误处理和恢复机制*/
@Service
public class ThreadPoolErrorHandler {@Autowiredprivate NotificationService notificationService;/*** 异步任务异常处理器*/@Beanpublic AsyncUncaughtExceptionHandler asyncExceptionHandler() {return new CustomAsyncExceptionHandler();}/*** 自定义异步异常处理器*/public class CustomAsyncExceptionHandler implements AsyncUncaughtExceptionHandler {@Overridepublic void handleUncaughtException(Throwable throwable, Method method, Object... params) {String errorMessage = String.format("异步任务执行失败 - 方法: %s.%s, 参数: %s, 错误: %s",method.getDeclaringClass().getSimpleName(),method.getName(),Arrays.toString(params),throwable.getMessage());System.err.println(errorMessage);throwable.printStackTrace();// 发送告警通知try {notificationService.sendPushNotification("admin", errorMessage);} catch (Exception e) {System.err.println("发送异常告警失败: " + e.getMessage());}// 根据异常类型决定是否重试if (isRetryableException(throwable)) {scheduleRetry(method, params);}}private boolean isRetryableException(Throwable throwable) {// 定义可重试的异常类型return throwable instanceof ConnectException ||throwable instanceof TimeoutException ||throwable instanceof SocketException;}private void scheduleRetry(Method method, Object... params) {// 实现重试逻辑，可以使用定时器延迟重试System.out.println("任务将在5秒后重试: " + method.getName());}}/*** 带重试机制的异步任务执行*/@Async("reliableTaskExecutor")@Retryable(value = {Exception.class}, maxAttempts = 3, backoff = @Backoff(delay = 2000))public CompletableFuture<String> executeTaskWithRetry(String taskName) {System.out.println("执行任务: " + taskName + ", 线程: " + Thread.currentThread().getName());try {// 模拟可能失败的操作if (Math.random() < 0.7) { // 70%的几率失败throw new RuntimeException("模拟任务执行失败");}Thread.sleep(1000);String result = "任务 " + taskName + " 执行成功";System.out.println(result);return CompletableFuture.completedFuture(result);} catch (InterruptedException e) {Thread.currentThread().interrupt();return CompletableFuture.completedFuture("任务被中断");}}/*** 重试失败后的回调方法*/@Recoverpublic CompletableFuture<String> recoverFromFailure(Exception ex, String taskName) {String errorMessage = "任务 " + taskName + " 重试失败，执行回退逻辑: " + ex.getMessage();System.err.println(errorMessage);// 执行回退逻辑，比如记录到数据库、发送通知等try {notificationService.sendPushNotification("admin", "任务执行失败: " + taskName);} catch (Exception e) {System.err.println("发送失败通知失败: " + e.getMessage());}return CompletableFuture.completedFuture("任务执行失败，已记录错误");}/*** 线程池健康检查和自动恢复*/@Scheduled(fixedRate = 60000) // 每分钟检查一次public void healthCheckAndRecover() {// 这里需要注入所有的线程池// checkAndRecoverThreadPool(userTaskExecutor);// checkAndRecoverThreadPool(emailTaskExecutor);// checkAndRecoverThreadPool(fileTaskExecutor);}private void checkAndRecoverThreadPool(ThreadPoolTaskExecutor executor) {ThreadPoolExecutor threadPoolExecutor = executor.getThreadPoolExecutor();// 检查线程池是否关闭if (threadPoolExecutor.isShutdown() || threadPoolExecutor.isTerminated()) {System.err.println("检测到线程池已关闭: " + executor.getThreadNamePrefix());// 尝试重新初始化线程池try {executor.initialize();System.out.println("线程池重新初始化成功: " + executor.getThreadNamePrefix());} catch (Exception e) {System.err.println("线程池重新初始化失败: " + e.getMessage());}}// 检查是否有太多拒绝的任务// 这需要自定义RejectedExecutionHandler来统计}
}/*** 线程池断路器模式* 当线程池过载时自动熔断，避免系统崩溃*/
@Service
public class ThreadPoolCircuitBreaker {private enum State {CLOSED,    // 正常状态OPEN,      // 熔断状态HALF_OPEN  // 半开状态}private volatile State state = State.CLOSED;private volatile long lastFailureTime = 0;private volatile int failureCount = 0;private final int failureThreshold = 5;           // 失败阈值private final long timeout = 60000;               // 熔断超时时间（毫秒）@Autowired@Qualifier("userTaskExecutor")private ThreadPoolTaskExecutor userTaskExecutor;/*** 带断路器的任务执行*/public CompletableFuture<String> executeWithCircuitBreaker(Callable<String> task) {if (state == State.OPEN) {// 检查是否可以转为半开状态if (System.currentTimeMillis() - lastFailureTime > timeout) {state = State.HALF_OPEN;System.out.println("断路器状态变更: OPEN -> HALF_OPEN");} else {return CompletableFuture.completedFuture("断路器开启，任务被拒绝");}}return CompletableFuture.supplyAsync(() -> {try {String result = task.call();onSuccess();return result;} catch (Exception e) {onFailure();throw new RuntimeException(e);}}, userTaskExecutor);}private void onSuccess() {failureCount = 0;if (state == State.HALF_OPEN) {state = State.CLOSED;System.out.println("断路器状态变更: HALF_OPEN -> CLOSED");}}private void onFailure() {failureCount++;lastFailureTime = System.currentTimeMillis();if (failureCount >= failureThreshold) {if (state == State.CLOSED) {state = State.OPEN;System.out.println("断路器状态变更: CLOSED -> OPEN");} else if (state == State.HALF_OPEN) {state = State.OPEN;System.out.println("断路器状态变更: HALF_OPEN -> OPEN");}}}public State getState() {return state;}
}

---

7. 常见问题和解决方案

7.1 常见配置错误

/*** 常见线程池配置问题示例和解决方案*/
@Configuration
public class ThreadPoolCommonIssues {/*** ❌ 错误配置1：无界队列 + 有限最大线程数* 问题：最大线程数永远不会生效，因为队列永远不会满*/public ThreadPoolTaskExecutor badConfiguration1() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setCorePoolSize(2);executor.setMaxPoolSize(10);        // 这个设置没有意义executor.setQueueCapacity(Integer.MAX_VALUE);  // 无界队列executor.setKeepAliveSeconds(60);executor.setThreadNamePrefix("bad-config-1-");return executor;}/*** ✅ 正确配置1：有界队列*/@Bean("goodConfiguration1")public ThreadPoolTaskExecutor goodConfiguration1() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setCorePoolSize(2);executor.setMaxPoolSize(10);executor.setQueueCapacity(50);     // 有界队列executor.setKeepAliveSeconds(60);executor.setThreadNamePrefix("good-config-1-");executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());executor.initialize();return executor;}/*** ❌ 错误配置2：核心线程数大于最大线程数* 问题：会抛出IllegalArgumentException*/public ThreadPoolTaskExecutor badConfiguration2() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setCorePoolSize(10);     // 核心线程数executor.setMaxPoolSize(5);      // 最大线程数 < 核心线程数，会报错executor.setQueueCapacity(50);executor.setThreadNamePrefix("bad-config-2-");return executor;}/*** ✅ 正确配置2：最大线程数 >= 核心线程数*/@Bean("goodConfiguration2")public ThreadPoolTaskExecutor goodConfiguration2() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setCorePoolSize(5);executor.setMaxPoolSize(10);     // 最大线程数 >= 核心线程数executor.setQueueCapacity(50);executor.setKeepAliveSeconds(60);executor.setThreadNamePrefix("good-config-2-");executor.initialize();return executor;}/*** ❌ 错误配置3：没有设置拒绝策略* 问题：默认AbortPolicy会抛出异常，可能导致任务丢失*/public ThreadPoolTaskExecutor badConfiguration3() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setCorePoolSize(2);executor.setMaxPoolSize(4);executor.setQueueCapacity(10);    // 小队列容易满executor.setThreadNamePrefix("bad-config-3-");// 没有设置拒绝策略，使用默认的AbortPolicyreturn executor;}/*** ✅ 正确配置3：设置合适的拒绝策略*/@Bean("goodConfiguration3")public ThreadPoolTaskExecutor goodConfiguration3() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setCorePoolSize(2);executor.setMaxPoolSize(4);executor.setQueueCapacity(10);executor.setThreadNamePrefix("good-config-3-");// 设置调用者运行策略，避免任务丢失executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());executor.initialize();return executor;}
}

7.2 内存泄漏问题

/*** 线程池内存泄漏问题演示和解决方案*/
@Service
public class MemoryLeakPrevention {/*** ❌ 可能导致内存泄漏的代码* 问题：ThreadLocal没有清理，导致内存泄漏*/private static final ThreadLocal<List<String>> THREAD_LOCAL_DATA = new ThreadLocal<>();@Async("memoryLeakTaskExecutor")public void problematicTask(String data) {// 在ThreadLocal中存储大量数据List<String> largeList = new ArrayList<>();for (int i = 0; i < 10000; i++) {largeList.add(data + "-" + i);}THREAD_LOCAL_DATA.set(largeList);// 执行业务逻辑processData(THREAD_LOCAL_DATA.get());// ❌ 忘记清理ThreadLocal，导致内存泄漏// THREAD_LOCAL_DATA.remove();}/*** ✅ 正确处理ThreadLocal的代码*/@Async("memoryLeakTaskExecutor")public void safeTask(String data) {try {// 在ThreadLocal中存储数据List<String> dataList = new ArrayList<>();for (int i = 0; i < 10000; i++) {dataList.add(data + "-" + i);}THREAD_LOCAL_DATA.set(dataList);// 执行业务逻辑processData(THREAD_LOCAL_DATA.get());} finally {// ✅ 确保清理ThreadLocalTHREAD_LOCAL_DATA.remove();}}/*** ❌ 可能导致内存泄漏的线程池配置* 问题：没有设置线程空闲时间，线程永不销毁*/public ThreadPoolTaskExecutor createLeakyExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setCorePoolSize(10);executor.setMaxPoolSize(50);executor.setQueueCapacity(100);// ❌ 没有设置keepAliveSeconds，非核心线程不会被回收executor.setThreadNamePrefix("leaky-");return executor;}/*** ✅ 防止内存泄漏的线程池配置*/@Bean("memoryLeakTaskExecutor")public ThreadPoolTaskExecutor createSafeExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();executor.setCorePoolSize(5);executor.setMaxPoolSize(20);executor.setQueueCapacity(100);executor.setKeepAliveSeconds(60);  // ✅ 设置空闲线程回收时间executor.setThreadNamePrefix("safe-");// ✅ 允许核心线程超时executor.setAllowCoreThreadTimeOut(true);// ✅ 设置线程工厂，确保创建的线程不是守护线程executor.setThreadFactory(new ThreadFactory() {private final AtomicInteger threadNumber = new AtomicInteger(1);@Overridepublic Thread newThread(Runnable r) {Thread t = new Thread(r, "safe-thread-" + threadNumber.getAndIncrement());t.setDaemon(false);  // 非守护线程return t;}});executor.setWaitForTasksToCompleteOnShutdown(true);executor.setAwaitTerminationSeconds(30);executor.initialize();return executor;}/*** 内存使用监控*/@Scheduled(fixedRate = 30000)public void monitorMemoryUsage() {MemoryMXBean memoryBean = ManagementFactory.getMemoryMXBean();MemoryUsage heapUsage = memoryBean.getHeapMemoryUsage();long used = heapUsage.getUsed();long max = heapUsage.getMax();double usageRate = (double) used / max * 100;System.out.println(String.format("内存使用情况 - 已用: %d MB, 总计: %d MB, 使用率: %.2f%%",used / 1024 / 1024, max / 1024 / 1024, usageRate));if (usageRate > 85) {System.err.println("内存使用率过高，建议检查是否存在内存泄漏");// 可以触发垃圾回收System.gc();// 发送告警// alertService.sendMemoryAlert(usageRate);}}private void processData(List<String> data) {// 模拟数据处理try {Thread.sleep(100);} catch (InterruptedException e) {Thread.currentThread().interrupt();}}
}

7.3 性能问题诊断

/*** 线程池性能问题诊断工具*/
@Service
public class ThreadPoolDiagnostics {@Autowiredprivate ThreadPoolMonitorService monitorService;/*** 诊断线程池性能问题*/public ThreadPoolDiagnosisReport diagnoseThreadPool(String executorName) {ThreadPoolStatus status = monitorService.getThreadPoolStatus(executorName);if (status == null) {return null;}ThreadPoolDiagnosisReport report = new ThreadPoolDiagnosisReport();report.setExecutorName(executorName);report.setDiagnosisTime(LocalDateTime.now());List<String> issues = new ArrayList<>();List<String> recommendations = new ArrayList<>();// 检查线程池使用率if (status.getPoolUsageRate() > 90) {issues.add("线程池使用率过高: " + status.getPoolUsageRate() + "%");recommendations.add("考虑增加最大线程数或优化任务执行效率");}// 检查队列使用率if (status.getQueueUsageRate() > 80) {issues.add("队列使用率过高: " + status.getQueueUsageRate() + "%");recommendations.add("考虑增加队列容量或增加处理线程数");}// 检查拒绝任务数if (status.getRejectedExecutionCount() > 0) {issues.add("存在任务拒绝: " + status.getRejectedExecutionCount() + " 个");recommendations.add("检查拒绝策略是否合适，考虑增加线程池容量");}// 检查线程创建效率if (status.getCurrentPoolSize() < status.getMaximumPoolSize() && status.getQueueSize() > status.getQueueCapacity() * 0.5) {issues.add("线程创建不及时，队列积压较多");recommendations.add("检查队列类型和容量配置");}// 检查任务完成率if (status.getTotalTaskCount() > 0) {double completionRate = (double) status.getCompletedTaskCount() / status.getTotalTaskCount() * 100;if (completionRate < 50) {issues.add("任务完成率较低: " + String.format("%.2f%%", completionRate));recommendations.add("检查任务是否有长时间阻塞或死锁");}}report.setIssues(issues);report.setRecommendations(recommendations);report.setSeverity(calculateSeverity(issues.size()));return report;}/*** 检查线程死锁*/public List<String> checkDeadlocks() {ThreadMXBean threadBean = ManagementFactory.getThreadMXBean();long[] deadlockedThreadIds = threadBean.findDeadlockedThreads();List<String> deadlockInfo = new ArrayList<>();if (deadlockedThreadIds != null) {ThreadInfo[] threadInfos = threadBean.getThreadInfo(deadlockedThreadIds);for (ThreadInfo threadInfo : threadInfos) {deadlockInfo.add(String.format("死锁线程: %s (ID: %d), 状态: %s, 锁信息: %s",threadInfo.getThreadName(),threadInfo.getThreadId(),threadInfo.getThreadState(),threadInfo.getLockInfo()));}}return deadlockInfo;}/*** 分析线程状态分布*/public Map<Thread.State, Integer> analyzeThreadStates() {ThreadMXBean threadBean = ManagementFactory.getThreadMXBean();ThreadInfo[] threadInfos = threadBean.dumpAllThreads(false, false);Map<Thread.State, Integer> stateCount = new HashMap<>();for (ThreadInfo threadInfo : threadInfos) {Thread.State state = threadInfo.getThreadState();stateCount.put(state, stateCount.getOrDefault(state, 0) + 1);}return stateCount;}/*** 生成性能调优建议*/public List<String> generateTuningRecommendations(ThreadPoolStatus status) {List<String> recommendations = new ArrayList<>();int cpuCores = Runtime.getRuntime().availableProcessors();// 基于CPU核心数的建议if (status.getCorePoolSize() > cpuCores * 4) {recommendations.add("核心线程数过多，建议设置为CPU核心数的1-4倍 (当前CPU核心数: " + cpuCores + ")");}if (status.getMaxPoolSize() > cpuCores * 8) {recommendations.add("最大线程数过多，可能导致上下文切换开销增大");}// 基于队列使用情况的建议if (status.getQueueCapacity() > 0) {double queueUsageRate = status.getQueueUsageRate();if (queueUsageRate < 10) {recommendations.add("队列使用率很低，可以考虑减少队列容量");} else if (queueUsageRate > 90) {recommendations.add("队列使用率很高，建议增加处理能力或队列容量");}}// 基于活跃线程数的建议if (status.getActiveThreadCount() == status.getMaxPoolSize()) {recommendations.add("所有线程都在工作，考虑增加最大线程数");}return recommendations;}private String calculateSeverity(int issueCount) {if (issueCount == 0) {return "正常";} else if (issueCount <= 2) {return "轻微";} else if (issueCount <= 4) {return "中等";} else {return "严重";}}
}/*** 线程池诊断报告*/
@Data
public class ThreadPoolDiagnosisReport {private String executorName;private LocalDateTime diagnosisTime;private List<String> issues;private List<String> recommendations;private String severity;
}

---

8. 总结

8.1 知识点回顾

通过这个详细的SpringBoot自定义线程池教程，我们学习了：

🎯 核心概念：

1. 线程池基础：理解线程池的工作原理、核心参数、执行流程
2. SpringBoot集成：掌握SpringBoot中异步任务的配置和使用
3. 自定义配置：学会通过配置文件和Java代码自定义线程池
4. 实际应用：了解在业务场景中如何合理使用线程池

💡 配置要点：

• 核心线程数：根据任务类型（CPU密集型/IO密集型）合理设置
• 最大线程数：控制并发上限，避免系统资源耗尽
• 队列容量：平衡内存使用和任务处理能力
• 拒绝策略：选择合适的策略处理超载情况
• 线程命名：便于监控和问题排查

🚀 最佳实践：

• 根据业务场景创建不同的线程池
• 设置合理的线程池参数
• 实现完善的监控和告警机制
• 注意内存泄漏和性能优化
• 建立错误处理和恢复机制

8.2 实际应用建议

1. 线程池规划：

# 推荐的线程池配置模板
spring:task:execution:pool:core-size: 8                    # CPU核心数max-size: 16                    # CPU核心数 * 2queue-capacity: 100             # 根据业务量调整keep-alive: 60s                 # 空闲线程回收时间thread-name-prefix: "app-task-"   # 便于识别shutdown:await-termination: true         # 优雅关闭await-termination-period: 30s   # 等待时间# 业务线程池配置
custom:thread-pool:# 用户操作（响应要求高）user:core-size: 4max-size: 8queue-capacity: 50# 邮件发送（IO密集型）email:core-size: 8max-size: 16queue-capacity: 200# 文件处理（资源消耗大）file:core-size: 2max-size: 4queue-capacity: 20

2. 监控指标：

• 线程池使用率（< 90%）
• 队列使用率（< 80%）
• 任务拒绝数量（= 0）
• 任务平均执行时间
• 任务完成率

3. 告警阈值：

• 线程池使用率 > 85%
• 队列使用率 > 75%
• 连续任务拒绝
• 任务执行时间过长
• 线程死锁检测

8.3 进阶学习方向

🌟 深入研究方向：

1. 高级特性：

   • CompletableFuture的高级用法
   • 响应式编程（WebFlux）
   • 异步事务处理
   • 分布式任务调度

2. 性能优化：

   • 线程池参数动态调优
   • 任务优先级队列
   • 批量任务处理优化
   • 内存和CPU使用优化

3. 企业级应用：

   • 微服务中的异步处理
   • 消息队列与线程池结合
   • 容器化环境下的线程池配置
   • 云原生架构中的并发处理

4. 监控运维：

   • APM工具集成
   • 自定义Metrics指标
   • 日志分析和告警
   • 性能问题诊断工具

🔧 实践建议：

• 在实际项目中应用学到的知识
• 建立线程池使用规范和最佳实践
• 定期review和优化线程池配置
• 关注SpringBoot和JVM的版本更新

📚 相关技术栈：

• Spring Cloud：微服务异步处理
• Redis：分布式锁和任务队列
• Kafka：异步消息处理
• Prometheus：监控指标收集
• Grafana：可视化监控面板

掌握SpringBoot自定义线程池是提高应用性能和并发处理能力的重要技能。通过合理的配置、完善的监控和持续的优化，可以显著提升系统的响应能力和用户体验。记住，没有一种配置适用于所有场景，需要根据具体的业务需求和系统环境来调整和优化线程池配置。