lusiqi

java并发系列-线程池，简单介绍线程池的几种实现方式，以及如何动态设置线程池参数。

线程池简介

线程池是一种基于池化思想管理线程的工具，经常出现在多线程服务器中。

线程池可以看作是线程的集合，在没有任务时，线程处于空闲状态，当请求来时，线程池给这个请求分配一个空闲的线程，任务完成后，回到线程池中等待下次任务。

使用线程池优点

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

Executor框架

JDK给我们提供了Excutor框架，来使用线程池，Ex cutor是线程池的基础。Executor提供了一种将任务提交与任务执行分离开来的机制。

jdk线程池的总体api架构：

Executor组成

组成部分

  1. 任务（Runnable/Callable）

    执行任务需要实现Runnable接口或者Callable接口，Runnable/Callable实现类都可以被ThreadPoolExecutor 或ScheduledThreadPoolExecutor执行。

  2. 任务的执行（Executor）

    任务执行机制的核心接口Executor，继承自Exector接口的ExecutorService接口，ThreadPoolExecutor和ScheduledThreadPoolExecutor实现了ExecutorService接口。

  3. 异步计算结果（Future）

    Future接口以及Future接口的实现类FutureTask类都可以代表一步计算的结果。当我们提交Runnable接口或者Callable接口的实现类给ThreadPoolExecutor或者ScheduledThreadPoolExecutor执行，调用Submit方法时会返回一个FutureTask对象。

执行步骤

  1. 主线程首先要创建实现Runnable或者Callable接口的任务对象。
  2. 把创建完成的实现Runnable、Callable接口的对象直接交给ExecutorService执行（ExecutorService.execute(Runnable command)）或者也可以把Runnable对象、Callable对象提交给ExecutorService执行（ExecutorService.submit(Runnable task) 或者ExecutorService.submit(Callable task))。
  3. 如果执行ExecutorService.submit(),ExecutorService将返回一个实现Future接口的对象，（execute()执行没有返回值，submit()会返回一个FutureTask对象），由于FutureTask实现了Runnable，我们也可以创建FutureTask，然后直接交给ExectorService执行。
  4. 最后，主线程可以执行FutureTask.get()方法来等待任务执行完成，主线程也可以执行FutureTask.cancel(boolean mayInterruptIfRunning)来取消此任务的执行。

Executor工具类实现线程池

常见的线程池实现方式有：Executor工具类可以实现三种类型的线程池（newFixedThreadPool、newCachedThreadPool、newSingleThreadExecutor）；

newFixedThreadPool

FixedThreadPool被称为可重用固定线程数的线程池，源码实现：

1
2
3
4
5
public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

从源码可看出，创建的FixedThreadPool的corePoolSize和maximumPoolSize都被设置为nThreads，这个nThreads参数是使用时传入的。

不推荐使用的原因

FixedThreadPool使用无界队列 linkedBlockingQueue,队列的容量为Integer.MAX_VALUE,将对线程池影响：

  1. 线程池中线程数达到corePoolSize后，先任务将在无界队列中等待，因此线程池中的线程数不会超过corePoolSize
  2. maximumPoolsize将是一个无效参数，因为不可能存在队列满的情况
  3. keepAliveTime将是一个无效参数
  4. 运行中的FixedThreadPool不会拒绝任务，在任务比较多的时候会导致OOM内存溢出

newSingleThreadExecutor

SingleThreadExecutor是只有一个线程的线程池，源码实现：

1
2
3
4
5
6
public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

从源码可以看出，新创建的SingleThreadExecutor的corePoolSize和maximumPoolSize都设置成了1，其他参数和FixedThreadPool相同，也即SingleThreadPool是nThreads=1时的FixedThreadPool。

不推荐使用的原因

由于SingleThreadPool是nThreads=1时的FixedThreadPool，所以原因和FixedThreadPool的原因相同，主要问题是会导致OOM。

newCachedThreadPool

CachedThreadPool是一个根据需要创建新的线程的线程池，源码实现：

1
2
3
4
5
public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

根据源码，CachedThreadPool的corePoolSize被设置为0,maximumPoolSize被设置为无界Integer.MAX_VALUE,这样的话，如果主线程提交任务的速度大于正在处理任务的线程的速度时，cachedThreadPool会不断的创建新的线程，极端情况下，会打导致耗尽CPU和内存资源。

不推荐使用原因

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

ThreadPoolExecutor构造方法实现线程池

可以通过ThreadPoolExecutor构造方法实现自定义参数的线程池。

线程池实现类ThreadPoolExecutor是 Executor框架最核心的类，ThreadPoolExecutor类中提供了四个构造方法，我们来看最长的那个，其余三个都是在这个构造方法的基础上产生的，其他几个构造方法说白点，都是给定某些默认参数的构造方法，比如默认制定拒绝策略等等。

构造方法源码

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
/**
 * 用给定的初始参数创建一个新的ThreadPoolExecutor。
 */
public ThreadPoolExecutor(int corePoolSize,//线程池的核心线程数量
                          int maximumPoolSize,//线程池的最大线程数
                          long keepAliveTime,//当线程数大于核心线程数时，多余的空闲线程存活的最长时间
                          TimeUnit unit,//时间单位
                          BlockingQueue<Runnable> workQueue,//任务队列，用来储存等待执行任务的队列
                          ThreadFactory threadFactory,//线程工厂，用来创建线程，一般默认即可
                          RejectedExecutionHandler handler//拒绝策略，当提交的任务过多而不能及时处理时，我们可以定制策略来处理任务
                           ) {
    if (corePoolSize < 0 ||
        maximumPoolSize <= 0 ||
        maximumPoolSize < corePoolSize ||
        keepAliveTime < 0)
        throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

参数

ThreadPoolExecutor 构造方法的参数（源码注释）：

  • corePoolSize:the number of threads to keep in the pool, even if they are idle, unless {@code allowCoreThreadTimeOut} is set

    核心线程数：不管它们创建以后是不是空闲的，线程池需要保持corePoolSize数量的线程，除非设置了allowCoreThreadTimeOut。

  • maximumPoolSize：the maximum number of threads to allow in the pool。

    最大线程数：线程池中最多允许创建maximumPoolSize个线程。

  • keepAliveTime：when the number of threads is greater than the core, this is the maximum time that excess idle threads will wait for new tasks before terminating。

    存活时间：如果经过keepAliveTime时间后，超过核心线程数的线程还没有接收到新的任务，那就回收。

  • unit：the time unit for the {@code keepAliveTime} argument

    keepAliveTime的时间单位。

  • workQueue：the queue to use for holding tasks before they are executed. This queue will hold only the {@code Runnable} tasks submitted by the {@code execute} method。

    存放待执行任务的队列：当提交的任务数超过核心线程数大小后，在提交的任务就存放在这里，它仅仅用来存放被execute方法提交的Runnable任务。

  • threadFactory:the factory to use when the executor creates a new thread。

    线程工厂：用来创建线程的工厂，比如这里可以定义线程的名称，在进行虚拟机栈分析时，看名字就知道线程来自哪里。

  • hander：the handler to use when execution is blocked because the thread bounds and queue capacities are reached。

    拒绝策略：当队列里面放满了任务，最大线程数的线程都在工作时，这是继续提交的任务线程池就处理不了了，应该按照给定的拒绝策略处理这些线程。

生命周期

线程池内部使用一个变量维护两个值：运行状态runState和线程数量workerCount，具体实现中，线程池将运行状态runState、线程数量workerCount两个参数维护放在了一起

1
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

ctl 定义为AtomicInteger，记录了线程池中的任务数量，和线程池的状态，两个信息。高3位保存runState，低29位保存workerCount。线程池也提供了若干方法获得线程池的当前运行状态、线程个数，这里使用的是位运算，相比基本运算，速度会快很多。

1
2
3
private static int runStateOf(int c)     { return c & ~CAPACITY; } //计算当前运行状态
private static int workerCountOf(int c)  { return c & CAPACITY; }  //计算当前线程数量
private static int ctlOf(int rs, int wc) { return rs | wc; }   //通过状态和线程数生成ctl

ThreadPoolExecutor的运行状态有5种：

运行状态 状态描述
RUNNING 能接受新提交的任务，并且也能处理阻塞队列中的任务。
SHUTDOWN 关闭状态，不在接受新的提交任务，但却可以继续处理阻塞队列中已保存的任务。
STOP 不能接受新的任务，也不处理队列中的任务，会中断正在处理任务的线程。
TIDYING 所有的任务都已经终止了，workerCount有效线程数为0。
TERMINATED 在terminated()方法执行完后进入改状态。

生命周期转换如下：

线程执行流程

所有任务调度都是由execute方法完成的，这部分完成的工作是：检查现在线程池的运行状态、运行线程数、运行策略，决定接下来执行的流程，是直接申请或者直接拒绝。执行过程如下：

  1. 首先检测线程池运训状态，如果不是RUNNING，则直接拒绝，线程池要保证在RUNNING状态下执行任务。
  2. 如果workerCount < corePoolSize,则创建并起动一个线程来执行新提交的任务。
  3. 如果workerCount >= corePoolSize,且线程池内的阻塞队列未满，则将任务添加到该阻塞对了中。
  4. 如果workerCount >= corePoolSize && workerCount < maximumPoolSize,且线程池内的阻塞队列已满，则创建并启动一个线程来执行新提交的任务。
  5. 如果workerCount >= maximumPoolSize ,并且线程池内的阻塞队列已满，则根据拒绝策略来处理该任务，默认处理方式是直接抛异常。

流程图：

任务阻塞

使用不同的队列可以实现不同热任务存取策略，常用的阻塞队列有：

名称 描述
ArrayBlockingQueue 一个用数组实现的有界阻塞队列，此队列按照先进先出FIFO的原则对元素进行排序，支持公平锁和非公平锁。
LinkedBlockingQueue 一个由链表结构组成的有界队列，此队列按照先进先出FIFO的原则对元素进行排序，此队列默认长度为Integer.MAX_VALUE,所以默认创建的队列有容量危险。
PriorityBlockingQueue 一个支持线程优先级排序的无界队列，默认自然序进行排序，也可以自定义compareTo方法来指定排序规则，不能保证同优先级元素的顺序。
DelayQueue 一个实现PriorityBlickingQueue实现延迟获取的无界队列，在创建元素时，可以指定多久才能从队列中获取当前元素，只有延时期满后才能从队列中获取元素。
SynchronousQueue 一个不储存元素的阻塞队列，每一个put操作必须等待taker操作，否则不能添加元素，支持公平锁和非公平锁，SynchronousQueue的一个使用场景时在线程池里，Executors.newCachedThreadPool()就使用了这个队列，这个线程池根据需要，新任务到来时，创建线程，如果有空闲线程则会重复使用，线程空闲了60秒后会被回收。
LinedTransferQueue 一个由链表结构组成的无界阻塞队列，相当于其他队列，LinedTransferQueue队列多了transfer和tryTransfer方法。
LinedBlockingDeque 一个由链表结构组成的双向阻塞队列，队列头部和尾部都可以添加和移除元素，多线程并发时，可以将锁的竞争最多降到一半。

排队策略要点：

  • 同步移交：不会放到队列中，而是等待线程执行它，如果当前线程没有执行，很可能会新开一个线程执行。
  • 无界限策略：如果核心线程都在工作，该线程会放在队列中，所以线程数不会超过核心线程数。
  • 有界限策略：可以避免资源耗尽，但是一定程度减低了吞吐量。

拒绝策略

拒绝策略是一个接口，设计如下：

1
2
3
public interface RejectedExecutionHandler {
    void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}

用户可以通过实现这个接口来定制拒绝策略，也可以选择jdk提供的四种拒绝策略：

名称 描述
ThreadPoolExecutor.AbortPolicy 丢弃任务并抛出RejectedExecutionException异常，这个是线程池默认的拒绝策略，在任务不能在提交的时候，抛出异常，及时反馈程序运行状态，如果是比较关键的业务，推荐使用此拒绝策略，这样子在系统不能承载更大的并发量的时候，能够及时通过异常发现。
ThreadPoolExecutor.DiscardPolicy 丢弃任务，但是不抛出异常，使用此策略，可能会使我们无法发现系统的异常状态。建议是一些无关紧要的业务采用此策略。
ThreadPoolExecutor.DiscardOldestPolisy 丢弃队列最前面的任务，然后重新提交被拒绝的任务，是否要采用此中拒绝策略，还得根据实际业务是否允许丢弃老的任务来认真衡量。
ThreadPoolExecutor.CallerRunsPolicy 由调用线程提交任务的线程处理该任务，这种情况是需要让所有任务都执行完毕，那么就适合大量计算的任务类型去执行，多线程仅仅是增大吞吐量的手段，最终必须要让每个任务都执行完毕。如果你的应用程序可以成熟延迟，并且不能丢弃任何一个任务请求的话可以选择这个策略。

动态设置线程池参数

自定义线程池参数，也即设置主要参数：corePoolSize、maximumPoolSize、workQueue（队列长度）。

正常的业务环境下，根据当前的需求，更改参数后，要重启服务。如果可以动态修改参数，就可以不用重启服务了（参考美团的实现方案）。

市面上大多数的答案都是要先区分线程池中的任务是IO密集型还是CPU密集型。

如果是CPU密集型，可以把核心线程数设置为服务器核心数+1。

《java并发编程实战》一书中说：即使当计算CPU密集型的线程偶尔由于页缺失故障，或者其他原因而暂停时，这个额外的线程也能保证cpu的时钟周期不会被浪费。

如果是IO密集型，这种任务应用起来，系统会大部分时间处理IO交互，我们可以多配一些线程，通常2N。

参数动态化

动态化线程池的核心设计：

  1. 简化线程池配置：线程池构造参数有8个，但是最核心的是三个：corePoolsize、maximumPoolSize、workQueue，它们最大程度的决定了线程池任务分配和线程分配策略。考虑实际应用中我们获取并发性场景：

    • 并行执行子任务，提高相应速度，这种情况下，应该使用同步队列，没有什么什么任务应该呗缓存下来，应该立即执行
    • 并行执行大批次任务，提升吞吐量。这种情况下。应该使用有界队列。使用队列去缓存大批量的任务，队列容量必须声明，防止任务无限堆积。

    所以线程池只需要提供这三个关键参数的配置，并且提供两种队列的选择，就可以满足大部分的业务需求了。

  2. 参数可动态修改：为了解决参数不好配，修改参数成本高问题。在java线程池留有高扩展性的基础上，封装线程池，允许线程池监听外部消息，根据消息进行修改配置。

  3. 增加线程池的监控：在线程池执行的生命周期添加监控。

JDK原生线程池ThreadPoolExecutor提供了参数的public的setter方法，Jdk允许线程池使用方通过ThreadPoolExecutor的实例来动态设置线程池的核心策略。

设置CorePoolSize

以setCorePoolSize为例在运行期线程池使用方调用此方法设置corePoolSize之后，线程池会直接覆盖原来的corePoolSize的值，并且基于当前值和原始值的比较结果采取不同的处理策略。

对于当前值小于当前工作线程数的情况，说明有多余的woker线程，此时会向当前idle的worker线程发送中断请求以实现回收，多余的worker在下次idle的时候，也会被回收。

对于当前值大于原始值且当前队列中有等待的任务时，线程池会创建新的worker来执行队列任务。

setCorePoolSize具体流程如下：

设置maximumPoolSize同corePoolSize。

设置队列长度

JDK原生线程池ThreadPoolExecutor的setter方法中没有设置队列长度的setter方法。队列的长度capacity在源码中被定义为了final。

设置长度的思路是：自定义一个队列，将队列的长度capacity的final去掉，并增加setter方法即可。

实例代码

Runnable+ThreadPoolExecutor

MyRunnable.java

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
import java.util.Date;

/**
 * 这是一个简单的Runnable类，需要大约5秒钟来执行其任务。
 */
publicclass MyRunnable implements Runnable {

    private String command;

    public MyRunnable(String s) {
        this.command = s;
    }

    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + " Start. Time = " + new Date());
        processCommand();
        System.out.println(Thread.currentThread().getName() + " End. Time = " + new Date());
    }

    private void processCommand() {
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public String toString() {
        returnthis.command;
    }
}

ThreadPoolExecutorDemo.java

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
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

publicclass ThreadPoolExecutorDemo {

    privatestaticfinalint CORE_POOL_SIZE = 5;
    privatestaticfinalint MAX_POOL_SIZE = 10;
    privatestaticfinalint QUEUE_CAPACITY = 100;
    privatestaticfinal Long KEEP_ALIVE_TIME = 1L;
    public static void main(String[] args) {

        //使用阿里巴巴推荐的创建线程池的方式
        //通过ThreadPoolExecutor构造函数自定义参数创建
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                CORE_POOL_SIZE,
                MAX_POOL_SIZE,
                KEEP_ALIVE_TIME,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(QUEUE_CAPACITY),
                new ThreadPoolExecutor.CallerRunsPolicy());

        for (int i = 0; i < 10; i++) {
            //创建WorkerThread对象（WorkerThread类实现了Runnable 接口）
            Runnable worker = new MyRunnable("" + i);
            //执行Runnable
            executor.execute(worker);
        }
        //终止线程池
        executor.shutdown();
        while (!executor.isTerminated()) {
        }
        System.out.println("Finished all threads");
    }
}
Callable+ThreadPoolExecutor

MyCallable.java

1
2
3
4
5
6
7
8
9
10
import java.util.concurrent.Callable;

publicclass MyCallable implements Callable<String> {
    @Override
    public String call() throws Exception {
        Thread.sleep(1000);
        //返回执行当前 Callable 的线程名字
        return Thread.currentThread().getName();
    }
}

ThreadPoolExecutorDemo.java

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
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

publicclass ThreadPoolExecutorDemo {

    privatestaticfinalint CORE_POOL_SIZE = 5;
    privatestaticfinalint MAX_POOL_SIZE = 10;
    privatestaticfinalint QUEUE_CAPACITY = 100;
    privatestaticfinal Long KEEP_ALIVE_TIME = 1L;

    public static void main(String[] args) {

        //使用阿里巴巴推荐的创建线程池的方式
        //通过ThreadPoolExecutor构造函数自定义参数创建
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                CORE_POOL_SIZE,
                MAX_POOL_SIZE,
                KEEP_ALIVE_TIME,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(QUEUE_CAPACITY),
                new ThreadPoolExecutor.CallerRunsPolicy());

        List<Future<String>> futureList = new ArrayList<>();
        Callable<String> callable = new MyCallable();
        for (int i = 0; i < 10; i++) {
            //提交任务到线程池
            Future<String> future = executor.submit(callable);
            //将返回值 future 添加到 list，我们可以通过 future 获得 执行 Callable 得到的返回值
            futureList.add(future);
        }
        for (Future<String> fut : futureList) {
            try {
                System.out.println(new Date() + "::" + fut.get());
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        }
        //关闭线程池
        executor.shutdown();
    }
}

常见问题

  1. 线程池被创建后里面有线程吗？如果没有的话，你知道有什么方法对线程池进行预热吗？

    线程池被创建后如果没有线程进来，里面是不会有线程的，如果需要预热可以调用方法：prestartAllCoreThreads全部启动、prestartCoreThread启动一个。

  2. 核心线程会被回收吗？需要怎么设置？

    核心线程默认是不会被回收的，如果需要回收核心线程数，需要调用allowCoreThreadTimeOut(true),allowCoreThreadTimeOut默认值为false。

 上一篇

 评论