Spring线程池ThreadPoolTaskExecutor_threadpooltaskexecutor定义线程中使用的任务队列

一、为什么使用线程池

• 减少在创建和销毁线程上所花的时间以及系统资源的开销
• 如果不使用线程池，有可能造成系统创建大量线程（线程池有最大线程限制）而导致消耗完系统内存。
• 好处：降低资源消耗、提高响应速度、提高线程的可管理性、提高程序的高可用

二、线程池工作流程

• 首先，判断核心线程数 corePoolSize 是否已满，没满，直接创建新的线程执行此任务。满了，进入下个流程。
• 其次，核心线程满后，会把新任务放入阻塞队列，判断阻塞队列 workQueue 是否已满，没满，将任务添加到阻塞队列。满了，进入下个流程。
• 最后，判断整个线程池是否已满，没满（此时需要maximumPoolSize > corePoolSize），创建新的非核心线程来执行此请求。满了，交给饱和策略处理。

默认情况下，线程池创建后，池中是没有线程的，需要提交任务才会创建线程，也可以通过preStartAllCoreThreads()设置，在线程池创建时就初始化所有核心线程。

三、线程池的配置

一般我们都是使用配置文件定义线程池属性，如下：

#异步线程池配置
async:
  executor:
    thread:
      core_pool_size: 20
      max_pool_size: 20
      queue_capacity: 100
      keep_alive_seconds: 60
      thread_name_prefix: AsyncExecutorThread-
1
2
3
4
5
6
7
8
9

线程池配置类：

package com.utour.config;

import org.springframework.beans.factory.annotation.Value;
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 java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

@Configuration
@EnableAsync//启用异步任务
public class AsyncConfig {

    @Value("${async.executor.thread.core_pool_size}")
    private int corePoolSize;

    @Value("${async.executor.thread.max_pool_size}")
    private int maxPoolSize;

    @Value("${async.executor.thread.queue_capacity}")
    private int queueCapacity;
    
	@Value("${async.executor.thread.keep_alive_seconds}")
    private int keepAliveSeconds;

	@Value("${async.executor.thread.thread_name_prefix}")
    private String threadNamePrefix;

    @Bean(name = "asyncExecutor")
    public Executor asyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(corePoolSize);//核心线程数
        executor.setMaxPoolSize(maxPoolSize);//最大线程数
        executor.setQueueCapacity(queueCapacity);//线程池队列容量
       	executor.setKeepAliveSeconds(keepAliveSeconds);//空闲的线程多久后被销毁，只对超出corePoolSize的线程起作用(如果核心线程数和最大线程数一样，一般不配置这个属性)
       	executor.setAllowCoreThreadTimeOut(true);//核心线程在规定时间内会被回收，默认为false
       	executor.setThreadNamePrefix(threadNamePrefix);//线程池中线程的名称前缀
       	//rejection-policy：饱和策略，当pool已经达到max size的时候，如何处理新任务？
       	//CallerRuns：不在新线程中执行任务，而是由调用者所在的线程来执行
       	executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
       	executor.initialize();//初始化
       	return executor;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

使用@Async注解，想要使其生效，必须满足以下条件：

• 在 @SpringBootApplication 启动类或者多线程配置类上加注解@EnableAsync，启用异步任务

• 异步方法上使用@Async，返回值为 void 或 Future。

• 切记切记切记 ，异步方法和调用方法一定要写在不同类中，写在同一个类中是没有效果的，建议建一个 AsyncService 异步处理类，类上加@Service注解，专门处理异步方法。

四、线程池配置详解

1、线程池阻塞队列：

• LinkedBlockingQueue：无界队列，不建议使用。当任务耗时较长时可能导致大量新任务在队列中堆积导致OOM，此时 maxPoolSize 已经不起作用。
• ArrayBlockingQueue：有界队列，创建时必须指定容量大小，可以指定公平与非公平，默认非公平（不保证等待最长时间的队列优先访问）。
• SynchronousQueue：直接提交，不存储任务，只有使用无界线程池或者有饱和策略时才建议使用该队列。

Executors.newFixedThreadPool采用的就是无界队列，spring的线程池也是采用无界队列。

2、线程池饱和策略：

当阻塞队列满了，且没有空闲的工作线程，如果继续提交任务，必须采取一种策略处理任务。

• AbortPolicy：直接抛出异常，默认策略
• CallerRunsPolicy：用调用者所在线程来执行任务
• DiscardOldestPolicy：丢弃阻塞队列中靠最前的任务，并执行当前任务
• DiscardPolicy：直接丢弃当前任务

3、线程池设置大小

任务类型不同，设置方式不一样。

• CPU密集型：尽量使用较小的线程池，一般为CPU核数+1、因为CPU密集型任务对CPU的使用率很高，若开过多的线程，只能增加线程的切换次数，带来额外开销。
• IO密集型：使用较大的线程数，一般为CPU核数 * 2。因为对CPU使用率不高，可以让CPU等待IO的时候去处理别的任务，充分利用CPU时间。
• 混合型：可以将任务分为CPU密集型和IO密集型，分别使用不同的线程池去处理。

最佳线程数 = （线程等待时间/线程CPU时间 + 1）* CPU核数

4、线程池的关闭

• shutdown()：不会立即终止线程池，而是等队列中的任务都执行完毕后才终止，但是再也不会接受新的任务。
• shutdownNow()：立即终止线程池，打断正在执行的任务，并且清空缓存队列，返回尚未执行的任务。

5、线程池中所有线程是否都执行完毕

• CountDownLatch：java.util.concurrent.CountDownLatch，同步辅助类，在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待。一个线程执行完将 CountDownLatch 设置的数 -1，知道 CountDownLatch 的数为0，才能继续往下执行，可以判断所有子线程是否执行完毕。
• countDown()：递减计数器的方法，如果计数到达0，则释放所有等待的线程。
• await(Long timeout, TimeUnit unit)：使当前线程在计数器减至0之前一直等待，除非线程被中断或超出了指定等待时间，如果当前计数为0，此方法立刻返回true。

一个CountDownLatch实例是不能重复使用的，可以重复使用的是CyclicBarrier。

import com.example.zzz.service.AsyncTaskService;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import javax.annotation.Resource;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;

@RunWith(SpringRunner.class)
@SpringBootTest
public class ZzzApplicationTests {

    @Resource(name = "asyncExecutor")
    private Executor taskExecutor;//这个是springboot的线程池配置类ThreadPoolTaskExecutor

    @Test
    public void contextLoads() {
        CountDownLatch latch = new CountDownLatch(50);//这个初始化计数要和线程数一样

        for (int i = 0; i < 50; i++) {
            final int j = i;
            taskExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        System.out.println(Thread.currentThread().getName() + "---" + j);
                    } catch (Exception e) {
                        e.printStackTrace();
                    } finally {
                        latch.countDown();//不管线程是否异常都需要计数器减少
                    }
                }
            });
        }
        try{
            latch.await(20L, TimeUnit.SECONDS);//保证之前的所有线程都执行完，才会执行下面的
            System.out.println("the race end");
        }catch(Exception e){
            e.printStackTrace();
        }
        System.out.println("bingo");
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48