线程池

发表于 | 更新于

线程池做的工作主要是控制运行的线程的数量，处理过程中将任务放入队列，然后在线程创建后启动这些任务，如果线程数量超出数量的线程排队等候，等其他线程执行完毕，再从队列中取出来执行。

主要特点：线程复用、控制最大并发数、管理线程。

  1. 降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
  2. 提高响应速度。当任务到达时，任务可以不需要的等到线程创建就能立即执行。
  3. 第三:提高线程的可管理性。线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统的分配， 调优和监控

创建线程池

1
2
3
4
Executors.newFixedThreadPool(int);
Executors.newSingleThreadExecutor();
Executors.newCachedThreadPool();
Executors.newScheduledThreadPool(int);
  1. newCachedThreadPool 是一个可根据需要创建新线程的线程池，但是在以前构造的线程可用时将重用它们。对于执行很多短期异步任务的程序而言，这些线程池通常可提高程序性能。调用 execute() 将重用以前构造的线程（如果线程可用）。如果现有线程没有可用的，则创建一个新线程并添加到池中。终止并从缓存中移除那些已有 60 秒钟未被使用的线程。因此，长时间保持空闲的线程池不会使用任何资源。注意，可以使用 ThreadPoolExecutor 构造方法创建具有类似属性但细节不同（例如超时参数）的线程池。
  2. newSingleThreadExecutor 创建是一个单线程池，也就是该线程池只有一个线程在工作，所有的任务是串行执行的，如果这个唯一的线程因为异常结束，那么会有一个新的线程来替代它，此线程池保证所有任务的执行顺序按照任务的提交顺序执行。
  3. newFixedThreadPool 创建固定大小的线程池，每次提交一个任务就创建一个线程，直到线程达到线程池的最大大小，线程池的大小一旦达到最大值就会保持不变，如果某个线程因为执行异常而结束，那么线程池会补充一个新线程。
  4. newScheduledThreadPool 创建一个大小无限的线程池，此线程池支持定时以及周期性执行任务的需求。
使用哪个

上述创建线程池的方法都不推荐使用

线程池不允许使用 Executors 去创建，而是通过 ThreadPoolExecutor 的方式，这样 的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险。

说明：Executors 返回的线程池对象的弊端如下：

  1. FixedThreadPool 和 SingleThreadPool: 允许的请求队列长度为 Integer.MAX_VALUE，可能会堆积大量的请求，从而导致 OOM。

  2. CachedThreadPool 和 ScheduledThreadPool: 允许的创建线程数量为 Integer.MAX_VALUE，可能会创建大量的线程，从而导致 OOM。

    ​ —来自《阿里巴巴Java手册》

线程池流程

  1. 在创建了线程池后，等待提交过来的任务请求。
  2. 当调用execute()方法添加一一个请求任务时，线程池会做如下判断:
    1. 如果正在运行的线程数量小于corePoolSize,那么马上创建线程运行这个任务;
    2. 如果正在运行的线程数量大于或等于corePoolSize,那么将这个任务放入队列;
    3. 如果这时候队列满了且正在运行的线程数量还小于maximumPoolSize,那么还是要创建非核心线程立刻运行这个任务;
    4. 如果队列满了且正在运行的线程数量大于或等于maximumPoolSize,那么线程池会启动饱和拒绝策略来执行。
  3. 当一个线程完成任务时，它会从队列中取下一个任务来执行。
  4. 当一个线程无事可做超过一定的时间(keepAliveTime) 时，线程池会判断:
    如果当前运行的线程数大于corePoolSize,那么这个线程就被停掉。所以线程池的所有任务完成后它最终会收缩到corePoolSize的大小

线程池参数

1
2
int corePoolSize;
线程池中的常驻核心线程数，allowCoreThreadTimeOut(boolean value) 设置为 true 时，闲置的核心线程会存在超时机制，如果在指定时间没有新任务来时，核心线程也会被终止，而这个时间间隔由第3个属性 keepAliveTime 指定。
1
2
int maximumPoolSize,
线程池中的常驻核心线程数
1
2
long keepAliveTime;
多余的空闲线程的存货时间。当前线程池数量超过corePoolSize时，当空闲时间达到keepAliveTime值时，多余空闲线程会被销毁直到只剩下corePoolSize个线程为止
1
2
TimeUnit unit；
keepAliveTime的单位
1
2
BlockingQueue<Runnable> workQueue;
任务队列，被提交但尚未被执行的任务
1
2
ThreadFactory threadFactory;
表示生产线程池中工作线程的线程工厂，用于创建线程一般用默认的即可
1
2
RejectedExecutionHandler handler;
拒绝策略，表实当前队列满了并且工作线程大于等于线程池的最大线程数

线程池拒绝策略

RejectedExecutionHandler提供了四种方式来处理任务拒绝策略

  1. 直接丢弃（DiscardPolicy）
  2. 丢弃队列中最老的任务(DiscardOldestPolicy)。
  3. 抛异常(AbortPolicy)
  4. 将任务分给调用线程来执行(CallerRunsPolicy)。

如何配置线程

上述提到了几个核心参数应该如何配置呢？

有一点是肯定的，线程池肯定是不是越大越好。

通常我们是需要根据这批任务执行的性质来确定的。

  • CPU 密集型任务（大量复杂的运算）应当分配较少的线程，比如 CPU 个数相当的大小。

  • IO 密集型任务：

    • 由于线程并不是一直在运行，所以可以尽可能的多配置线程，比如 CPU 个数 * 2
    • 该任务需要大量的IO，即大量的阻塞，故需要多配置线程数，如 CPU核数 / (1 - 阻塞系数) 阻塞系数在0.8~0.9之间

线程池的关闭

ThreadPoolExecutor 提供了两个方法，用于线程池的关闭，分别是 shutdown() 和 shutdownNow()。

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

线程池Demo

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
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.*;

public class MyThreadPoolDemo {
    public static void main(String[] args) throws InterruptedException {

        ExecutorService threadPool = new ThreadPoolExecutor(
                2,
                5,
                1L,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(3),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.CallerRunsPolicy());

        List<String> list = Arrays.asList("a", "b", "c");
//        ExecutorService threadPool = Executors.newFixedThreadPool(5); //线程池中5个线程
//        ExecutorService threadPool = Executors.newSingleThreadExecutor(); // 一池一个处理线程
        // ExecutorService threadPool = Executors.newCachedThreadPool();// 一池N个处理线程
        // 模拟10个用户办理业务，每个用户就是一个来自外部的请求线程
        try {
            for (int i = 0; i < 10; i++) {
                threadPool.execute(() -> System.out.println(Thread.currentThread().getName() + "\t 办理业务"));
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            threadPool.shutdown();
        }
    }
}
0%