SpringBoot使用线程池

2022-06-02 本文已影响0人轻轻敲醒沉睡的心灵

1. 线程池

线程池是一种线程使用模式。线程过多会带来额外的开销，其中包括创建销毁线程的开销、调度线程的开销等等，同时也降低了计算机的整体性能。线程池维护多个线程，等待监督管理者分配可并发执行的任务。这种做法，一方面避免了处理任务时创建销毁线程开销的代价，另一方面避免了线程数量膨胀导致的过分调度问题，保证了对内核的充分利用。
使用线程池，有几点好处：

降低资源消耗：通过池化技术重复利用已创建的线程，降低线程创建和销毁造成的损耗。
提高响应速度：任务到达时，无需等待线程创建即可立即执行。
提高线程的可管理性：线程是稀缺资源，如果无限制创建，不仅会消耗系统资源，还会因为线程的不合理分布导致资源调度失衡，降低系统的稳定性。使用线程池可以进行统一的分配、调优和监控。

2. Java中线程池

Java开发，我们使用JDK环境，开发框架基本都是Spring全家桶。线程池的基础原理都在JDK中，JDK1.5新增线程池相关类，在核心jar包rt.jar中，java.util.concurrent下面。

2.1 JDK中线程池

看一下jdk中线程池继承关系类图：

线程UML.png
可以看出，包含三个线程池：ForkJoinPool，ThreadPoolExecutor，ScheduledThreadPoolExecutor。

ForkJoinPool是Java 1.7 引入的一种新的并发框架，核心思想是将大的任务拆分成多个小任务（fork），然后在将多个小任务处理汇总到一个结果上（join），充分利用多cpu，多核CPU的优势，引入了“work-stealing”机制，更有效的利用线程。
ThreadPoolExecutor是 java常用的线程池，提供基础的线程池功能。初始化传入不同类型的工作队列和拒绝策略参数，可以定制不同类型和功能的线程池，应用最为广泛。
ScheduledThreadPoolExecutor 从类图上可以看出，它继承了ThreadPoolExecutor，并实现了ScheduledExecutorService，是对ThreadPoolExecutor做的功能扩展，本质上是一个使用线程池执行定时任务的类，可以用来在给定延时后执行异步任务或者周期性执行任务，较任务调度的Timer来说，其功能更加强大。

2.2 JDK中提供的几类线程池

JDK中，使用ThreadPoolExecutor给创建了几个不同功能种类的线程池，可以调出来看看：

 //创建一个核心线程数和最大线程数相同的线程池
 ExecutorService executorService = Executors.newFixedThreadPool(4);
 executorService.execute(()->{System.out.println("this is fixed thread pool");});
 //创建一个单线程的线程池
 ExecutorService singleExecutor = Executors.newSingleThreadExecutor();
 singleExecutor.execute(() -> System.out.println("this is single thread pool"));
 //创建一个缓存线程池
ExecutorService cacheExecutor = Executors.newCachedThreadPool();
cacheExecutor.execute(()-> System.out.println("this is cache thread pool"));
//创建一个延时执行任务的线程池
ScheduledExecutorService scheduleExecutor = Executors.newScheduledThreadPool(4);
scheduleExecutor.schedule(()-> System.out.println("this is schedule thread  pool"),2,TimeUnit.SECONDS);

看一下都有什么特点：

线程池名称	使用阻塞队列	特点
newFixedThreadPool	LinkedBlockingQueue()	1、核心线程数和最大线程数相同 2、由于keepAliveTime设置为0，当线程创建后会一直存在 3、由于用的是无界队列所以可能会导致OOM
newSingleThreadExecutor	LinkedBlockingQueue()	1、核心线程数和最大线程数都为1单线程 2、无界队列可能导致OOM
newCachedThreadPool	SynchronousQueue()	1、核心线程数为0，最大线程数为Integer.MAX_VALUE 2、当没任务时线程存活时间为60秒 3、使用的是0大小的队列，所以不存储任务，只做任务转发
newScheduledThreadPool	DelayedWorkQueue()	1、执行周期任务 2、无界队列，可能会导致OOM

从上面可以看出他们各有各的特点，但是阿里巴巴开发守则却不推荐使用以上线程池，因为它们可能会对服务资源的浪费，所以推荐使用通过ThreadPoolExecutor自定线程池。
Spring中将Java中的线程池进行了封装，而且提供了默认实现，也能自定义线程池，我一般都用Spring中的线程池包。

3. Spring中的线程池

Spring中的线程池和JDK中的基本一样，在org.springframework.scheduling.concurrent包下面。
和JDK中对应的，Spring的顶层接口是TaskExecutor，它实现了JDK中的Executor接口。
Spring中常用的线程池是ThreadPoolTaskExecutor，它是是借助于JDK并发包中的java.util.concurrent.ThreadPoolExecutor来实现的。
要想使用线程池，先了解一下线程池中的一些参数：

3.1 线程池用到的一些参数

因为我们常用的是ThreadPoolExecutor线程池，所以去这个类中找。

参数
也就这几个常用参数，在下面使用中再介绍意思。

3.2 使用自定义线程池

3.2.1 线程池配置类

package com.example.demo.threadpool;

import java.util.concurrent.ThreadPoolExecutor;

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

import lombok.Data;

@Configuration
public class ThreadPoolConfig {
    
    ThreadPoolProperties properties = new ThreadPoolProperties();
    
    @Bean(name = "taskExecutor")
    public ThreadPoolTaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(properties.getCorePoolSize());
        executor.setMaxPoolSize(properties.getMaxPoolSize());
        executor.setQueueCapacity(properties.getQueueCapacity());
        executor.setThreadNamePrefix(properties.getThreadNamePrefix());
        // 设置线程保持活跃的时间（默认：60）
        executor.setKeepAliveSeconds(properties.getKeepAliveTime());
        // 当任务完成后，长时间无待处理任务时，销毁线程池
        executor.setWaitForTasksToCompleteOnShutdown(properties.isWaitForTasksToCompleteOnShutdown());
        executor.setAwaitTerminationSeconds(properties.getAwaitTerminationSeconds());
        // 设置任务拒绝策略
        /**
         * 4种
         * ThreadPoolExecutor类有几个内部实现类来处理这类情况：
            - AbortPolicy 丢弃任务，抛RejectedExecutionException
            - CallerRunsPolicy 由该线程调用线程运行。直接调用Runnable的run方法运行。
            - DiscardPolicy  抛弃策略，直接丢弃这个新提交的任务
            - DiscardOldestPolicy 抛弃旧任务策略，从队列中踢出最先进入队列（最后一个执行）的任务
         * 实现RejectedExecutionHandler接口，可自定义处理器
         */
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        return executor;
    }
    
    @Data
    class ThreadPoolProperties {
        
        /**
         * 核心线程数（默认是1）：若是IO密集型，cpu核心数*2，若是cpu密集型，cpu核心数
         * 核心线程会一直存活，及时没有任务需要执行
         * 设置allowCoreThreadTimeout=true（默认false）时，核心线程会超时关闭
         * 注意：当线程数小于核心线程数时，即使有线程空闲，线程池也会优先创建新线程处理
         */
        private int corePoolSize = 8;  
        /** 
         * 最大线程数，系统默认Integer.MAX_VALUE
         */  
        private int maxPoolSize = 20;  
        /** 
         * 允许线程空闲时间（单位：默认为秒，默认60S）
         * 当线程空闲时间达到keepAliveTime时，线程会退出，直到线程数量=corePoolSize
         * 如果allowCoreThreadTimeout=true，则会直到线程数量=0 
         */  
        private int keepAliveTime;  
        /** 
         * 缓冲队列大小，系统默认Integer.MAX_VALUE
         * 注意：这个值肯定要改小，不然任务陡增时，都堆在队列中（队列值大），
         * 核心线程数就那几个，无法很快执行队列中的任务，
         * 就会延长响应时间，阻塞任务
         */  
        private int queueCapacity = 6; 
        /** 
         * 线程池名前缀，用于监控识别
         */  
        private String threadNamePrefix = "test";  
        
        /**
         * 允许核心线程超时（默认false）
         */
        private boolean allowCoreThreadTimeout = false;
        
        /**
         * 当任务完成后，长时间无待处理任务时，销毁线程池
         */
        private boolean waitForTasksToCompleteOnShutdown = false;
        
        private int awaitTerminationSeconds;
        
    }
}

上面配置了线程池，并生成了线程池bean，交给了Spring容器管理，使用时注入即可使用。

3.2.2 使用线程池
SpringBoot使用线程池注意2点：
1. 添加@EnableAsync注解开启异步，即多线程
2. 在要使用的方法上加上@Async("taskExecutor")注解，括号内是线程池名字

package com.example.demo.service;

import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.stereotype.Service;

@Service
@EnableAsync
public class ThreadPoolService {
    
    @Async("taskExecutor")
    public void F1() throws InterruptedException {
        System.out.println(Thread.currentThread().getName());
         for(int i = 0 ; i < 10 ; i++){
             System.out.println("A-"+i);
             Thread.sleep(500);
         }
         
    }
    
    @Async("taskExecutor")
    public void F2() throws InterruptedException {
        System.out.println(Thread.currentThread().getName());
         for(int i = 20 ; i < 30 ; i++){
            System.out.println("B-"+i);
            Thread.sleep(500);
        }
         
    }
}

3.2.3 @Async失效情况：

1. 异步方法使用static修饰
2.异步类没有使用@Component注解（或其他注解）导致spring无法扫描到异步类
1. 异步方法不能与被调用的异步方法在同一个类中
1. 类中需要使用@Autowired或@Resource等注解自动注入，不能自己手动new对象
1. 如果使用SpringBoot框架必须在启动类中增加@EnableAsync注解

3.3 SpringBoot中自动装配的线程池

SpringBoot线程池自动装配在spring-boot-autoconfigure这个jar中，在org.springframework.boot.autoconfigure.task.TaskExecutionAutoConfiguration类中。
自动装配条件：

条件
自动装配的Bean名字给了2个，一个是applicationTaskExecutor，一个是taskExecutor。如果自动装配了，可以直接注入使用。

Bean

4. 线程池处理任务流程：

当往线程池中提交新任务时，线程池主要流程如下：
核心线程数 -> 线程队列 -> 最大线程数 -> 拒绝策略

如果池中任务数 < corePoolSize （核心线程数），创建新线程立即执行任务
如果池中任务数 > corePoolSize，新任务放到缓存队列当中等待执行
队列满，线程数量<maxPoolSize，新建线程立即执行任务
队列满，线程数量>=maxPoolSize，使用拒绝策略拒绝

所以使用线程池，需要注意：

搞懂执行流程，设置合适的corePoolSize 、maxPoolSize和队列容量，避免延时任务，一般队列容量不会像默认值那么大，否则可能等很长时间；
注意使用时一些注解加上
要了解线程的一些基础知识，比如生命周期等。
Java多线程--线程的状态 - 简书 (jianshu.com)
Java多线程--线程常用方法 - 简书 (jianshu.com)