ThreadPoolTaskExecutor介绍

ThreadPoolTaskExecutor是Spring框架提供的线程池管理器，它继承自ThreadPoolExecutor类，并对其进行了一些扩展和封装。下面是ThreadPoolTaskExecutor的常用参数：

1. corePoolSize（核心线程数）：线程池中保持活动状态的最小线程数。
2. maxPoolSize（最大线程数）：线程池中允许存在的最大线程数，超过该数量的任务会放入等待队列。
3. keepAliveSeconds（线程活动时间）：当线程池中线程数量超过核心线程数时，多余的空闲线程的存活时间。
4. queueCapacity（等待队列容量）：等待队列的容量，用于存放等待执行的任务。
5. threadNamePrefix（线程名前缀）：新创建的线程的名称前缀，默认为task-。
6. rejectedExecutionHandler（拒绝策略）：当线程池和等待队列都满了，无法处理新的任务时的处理策略，可选的策略有：

1、 AbortPolicy：直接抛出RejectedExecutionException异常，默认策略。
2、CallerRunsPolicy：在调用者的线程中执行该任务。
3、 DiscardPolicy：丢弃任务，不抛出异常。
4、DiscardOldestPolicy：丢弃队列中最旧的任务，然后重新尝试执行新的任务。

除了以上常用参数外，ThreadPoolTaskExecutor还提供了一些其他可配置属性和方法来进一步控制线程池的行为。

例如，你可以修改核心线程数、最大线程数和等待队列容量：

executor.setAllowCoreThreadTimeOut(true); // 允许核心线程超时关闭
executor.setWaitForTasksToCompleteOnShutdown(true); // 等待所有任务完成后再关闭线程池
executor.setAwaitTerminationSeconds(60); // 等待终止的时间

实现步骤

配置管理器

@Slf4j
@EnableAsync
@Configuration
public class AsyncConfiguration implements AsyncConfigurer {

     @Primary
     @Bean(name = "threadPoolTaskExecutor")
    public ThreadPoolTaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        // 获取电脑的处理器数量，一般电脑一个处理器有两个逻辑线程
        int num = Runtime.getRuntime().availableProcessors();
        // 核心线程数目 核心线程数最好设置为电脑的逻辑线程总数 + 1，达到最大化利用处理器
        executor.setCorePoolSize(num * 2 + 1);
        // 指定最大线程数
        executor.setMaxPoolSize(200);
        // 队列中最大的数目
        executor.setQueueCapacity(100);
        // 线程名称前缀
        executor.setThreadNamePrefix("taskExecutor-");
        // 拒绝策略：不在新线程中执行任务，而是由调用者所在的线程来执行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        // 当调度器shutdown被调用时等待当前被调度的任务完成
        executor.setWaitForTasksToCompleteOnShutdown(true);
        // 线程空闲后的最大存活时间 秒
        executor.setKeepAliveSeconds(60);
        // 线程池初始化
        executor.initialize();
        return executor;
    }

    @Override
    public Executor getAsyncExecutor() {
        return taskExecutor();
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return new CustomAsyncUncaughtExceptionHandler();
    }

    static class CustomAsyncUncaughtExceptionHandler implements AsyncUncaughtExceptionHandler {
        @Override
        public void handleUncaughtException(Throwable ex, Method method, Object... params) {
            log.error(">>> CustomAsyncUncaughtExceptionHandler,class:{}, method: {}, params: {}, error: {}",
                    method.getDeclaringClass().getSimpleName(), method.getName(), Arrays.toString(params),
                    ex.getMessage());
        }
    }

}

测试方式一：

第一步：首先在application启动类添加@EnableAsync

@SpringBootApplication
@EnableAsync
public class ThreadpoolApplication {
    public static void main(String[] args) {
        SpringApplication.run(ThreadpoolApplication.class, args);
    }
}

第二步：在需要异步执行的方法上加上@Async注解

@Service
public class AsyncTest {
    protected final Logger logger = LoggerFactory.getLogger(this.getClass());
    @Async("threadPoolTaskExecutor")
    public void hello(String name){
    	//这里使用logger 方便查看执行的线程是什么
        logger.info("异步线程启动 started."+name);  
    }
}

第三步：测试类进行测试验证

    @Autowired
    AsyncTest asyncTest;
    @Test
    void contextLoads() throws InterruptedException {
        asyncTest.hello("afsasfasf");
        //一定要休眠 不然主线程关闭了，子线程还没有启动
        Thread.sleep(1000);
    }

通过在方法上添加@Async(“threadPoolTaskExecutor”)注解，指定了使用名为"threadPoolTaskExecutor"的线程池来执行异步任务。
这样，你就可以在整个应用程序中使用同一个全局线程池管理器来执行异步任务。确保在需要使用的类上添加@Component、@Service或其他适当的注解，以便Spring Boot能够正确地扫描和管理这些组件。

测试方式二：

直接调用ThreadPoolTaskExecutor，修改下测试类，直接注入ThreadPoolTaskExecutor

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

@Service
public class MyService {

    @Autowired
    private ThreadPoolTaskExecutor taskExecutor;

    public void executeTasks() {
        for (int i = 0; i < 10; i++) {
            int taskId = i;
            taskExecutor.execute(() -> {
                System.out.println("Executing task: " + taskId + " on thread: " + Thread.currentThread().getName());
                // 执行具体任务逻辑
            });
        }
    }
}

获取ThreadPoolTaskExecutor线程池状态和参数

ThreadPoolTaskExecutor 提供了多个方法来获取线程池的状态和参数。以下是一些常用的监控方法：

• getActiveCount(): 获取当前活跃的线程数。
• getThreadPoolExecutor(): 返回内部封装的 ThreadPoolExecutor 实例，可以通过该实例访问更多详细信息。
• getCompletedTaskCount(): 获取已完成任务的数量（仅当 ThreadPoolTaskExecutor 使用的是ThreadPoolExecutor 的扩展时可用）。
• getPoolSize(): 获取当前线程池的大小（包括活跃和空闲线程）。
• getQueue().size(): 获取等待队列中任务的数量。

这里是一个简单的示例，展示如何获取上述的一些参数：

ThreadPoolTaskExecutor taskExecutor = ... // 假设已经配置并初始化了你的 taskExecutor

// 获取当前活跃的线程数
int activeCount = taskExecutor.getActiveCount();
System.out.println("Active threads: " + activeCount);

// 获取线程池大小
int poolSize = taskExecutor.getPoolSize();
System.out.println("Pool size: " + poolSize);

// 获取等待队列中任务的数量
int queueSize = taskExecutor.getThreadPoolExecutor().getQueue().size();
System.out.println("Queue size: " + queueSize);

// 获取已完成任务的数量
long completedTasks = taskExecutor.getThreadPoolExecutor().getCompletedTaskCount();
System.out.println("Completed tasks: " + completedTasks);

关闭 ThreadPoolTaskExecutor

为了确保资源被正确释放，你应该在不再需要 ThreadPoolTaskExecutor 的时候调用其 shutdown() 方法。这个方法会启动一个平滑的关闭过程：停止接受新的任务，但允许已提交的任务执行完成。

taskExecutor.shutdown();

如果你希望立即关闭线程池，并尝试中断正在执行的任务，可以使用 shutdownNow() 方法：

taskExecutor.shutdownNow();

注意，这些方法通常应该在应用关闭时调用。如果 ThreadPoolTaskExecutor 是通过 Spring 配置管理的 Bean，它会在 Spring 应用上下文关闭时自动调用 shutdown() 方法进行清理，因此你不需要手动调用它们。

直接使用TaskExecutor

     @Autowired
    private TaskExecutor taskExecutor;
 
    public void TestTaskExecutor() {
        taskExecutor.execute(() -> {
            try {
                //TODO: 业务代码 
            }catch (Exception e){
                e.printStackTrace();
            }

        });
    }

线程池使用最佳实践

Spring Boot配置示例

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;

import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

@Configuration
@EnableAsync
public class ThreadPoolConfig {
    
    /**
     * 核心业务线程池
     */
    @Bean(name = "coreThreadPool")
    public ThreadPoolTaskExecutor coreThreadPool() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(20);
        executor.setQueueCapacity(200);
        executor.setKeepAliveSeconds(60);
        executor.setThreadNamePrefix("core-thread-");
        executor.setAllowCoreThreadTimeOut(false);
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(30);
        executor.initialize();
        return executor;
    }
    
    /**
     * 异步任务线程池
     */
    @Bean(name = "asyncThreadPool")
    public ThreadPoolTaskExecutor asyncThreadPool() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("async-thread-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());
        executor.initialize();
        return executor;
    }
    
    /**
     * 定时任务线程池
     */
    @Bean(name = "scheduledThreadPool")
    public ThreadPoolTaskScheduler scheduledThreadPool() {
        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
        scheduler.setPoolSize(5);
        scheduler.setThreadNamePrefix("scheduled-thread-");
        scheduler.setAwaitTerminationSeconds(30);
        scheduler.setWaitForTasksToCompleteOnShutdown(true);
        scheduler.setRemoveOnCancelPolicy(true);
        scheduler.initialize();
        return scheduler;
    }
    
    /**
     * SimpleThreadPoolTaskExecutor（适用于Quartz）
     */
    @Bean(name = "simpleThreadPool")
    public SimpleThreadPoolTaskExecutor simpleThreadPool() {
        SimpleThreadPoolTaskExecutor executor = new SimpleThreadPoolTaskExecutor();
        executor.setThreadCount(10);
        executor.setThreadNamePrefix("simple-thread-");
        executor.setThreadPriority(Thread.NORM_PRIORITY);
        executor.initialize();
        return executor;
    }
}

使用

import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.util.concurrent.*;

@Service
public class ThreadPoolService {
    
    @Resource
    @Qualifier("coreThreadPool")
    private ThreadPoolTaskExecutor coreExecutor;
    
    @Resource
    @Qualifier("asyncThreadPool")
    private ThreadPoolTaskExecutor asyncExecutor;
    
    /**
     * 方式1：直接使用ExecutorService
     */
    public void executeTask() {
        ExecutorService executor = Executors.newFixedThreadPool(5);
        executor.execute(() -> {
            // 业务逻辑
        });
        executor.shutdown();
    }
    
    /**
     * 方式2：使用Spring注入的线程池
     */
    public void executeWithSpringPool() {
        coreExecutor.execute(() -> {
            // 核心业务逻辑
        });
    }
    
    /**
     * 方式3：使用@Async注解
     */
    @Async("asyncThreadPool")
    public CompletableFuture<String> asyncMethod() {
        // 异步执行的方法
        return CompletableFuture.completedFuture("result");
    }
    
    /**
     * 方式4：使用CompletableFuture
     */
    public CompletableFuture<Void> completableFutureDemo() {
        return CompletableFuture.runAsync(() -> {
            // 异步任务
        }, coreExecutor);
    }
    
    /**
     * 方式5：使用ThreadPoolExecutor监控
     */
    public void monitorThreadPool() {
        ThreadPoolExecutor executor = coreExecutor.getThreadPoolExecutor();
        
        System.out.println("核心线程数: " + executor.getCorePoolSize());
        System.out.println("最大线程数: " + executor.getMaximumPoolSize());
        System.out.println("活跃线程数: " + executor.getActiveCount());
        System.out.println("池中线程数: " + executor.getPoolSize());
        System.out.println("队列大小: " + executor.getQueue().size());
        System.out.println("完成任务数: " + executor.getCompletedTaskCount());
        System.out.println("总任务数: " + executor.getTaskCount());
        
        // 动态调整
        executor.setCorePoolSize(15);
        executor.setMaximumPoolSize(30);
    }
}