Java 线程池使用小结

我们在使用多线程编程的时候，为何要用线程池呢？使用线程池的好处是什么呢？线程池有哪些使用场景？

为何使用线程池？

因为线程资源宝贵，不论创建新的线程还是销毁线程，都有相应的资源开销，比如在数据库连接方面，每个请求过来都是新建连接数据库的线程，请求少，资源开销总体不大，但也架不住请求一直来，线程一直频繁新建和销毁。所以为了缓解这种=资源开销，我们就有了线程的管理工具-线程池。

当然，这种池化的思想，通常我们都用在数据库的连接上，比如数据库的连接池，redis连接池等。

这里总结下使用线程池的好处[1]：

• 1.降低资源消耗，通过池化思想，减少创建线程和销毁线程的消耗，控制资源
• 2.提高响应速度，任务到达时，无需创建线程即可运行
• 3.提供更多更强大的功能，可扩展性高

准备

线程池的五种状态[3]：

• 1.RUNNING：线程池一旦被创建，就处于 RUNNING 状态，任务数为 0，能够接收新任务，对已排队的任务进行处理。

• 2.SHUTDOWN：不接收新任务，但能处理已排队的任务。调用线程池的 shutdown() 方法，线程池由 RUNNING 转变为 SHUTDOWN 状态。

• 3.STOP：不接收新任务，不处理已排队的任务，并且会中断正在处理的任务。调用线程池的 shutdownNow() 方法，线程池由(RUNNING 或 SHUTDOWN ) 转变为 STOP 状态。

• 4.TIDYING：

  • SHUTDOWN 状态下，任务数为0， 其他所有任务已终止，线程池会变为 TIDYING 状态，会执行 terminated() 方法。线程池中的 terminated() 方法是空实现，可以重写该方法进行相应的处理。
  • 线程池在 SHUTDOWN 状态，任务队列为空且执行中任务为空，线程池就会由 SHUTDOWN 转变为 TIDYING 状态。
  • 线程池在 STOP 状态，线程池中执行中任务为空时，就会由 STOP 转变为 TIDYING 状态。

• 5.TERMINATED：线程池彻底终止。线程池在 TIDYING 状态执行完 terminated() 方法就会由 TIDYING 转变为 TERMINATED 状态。

状态流转：

1.线程池使用示例

demo

package org.example;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class ThreadPool {
    private static int taskCount = 20; // 任务数量

    private static AtomicInteger taskCountExec; // 实际完成的任务数

    public static void main(String[] args) {
        taskCountExec = new AtomicInteger();

        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                2,
                5,
                1,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(10));

        System.out.println("总任务数：" + taskCount);
        long startTime = System.currentTimeMillis();

        // 任务提交
        for (int i = 0; i < taskCount; i++) {
            // 定义任务线程
            Thread thread = new Thread(() -> {
                try {
                    TimeUnit.MILLISECONDS.sleep(500L); // 任务耗时
                    System.out.println("已执行" + taskCountExec.addAndGet(1) + "个任务");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });

            // 提交任务
            try {
                executor.execute(thread);
            } catch (RejectedExecutionException e) {
                // e.printStackTrace(); // 拒绝策略抛错
                taskCount = executor.getActiveCount() + executor.getQueue().size();
            }
        }

        // mark endTime
        long endTime = 0;
        while (executor.getCompletedTaskCount() < taskCount) {
            endTime = System.currentTimeMillis();
        }

        System.out.println(taskCountExec + "个任务已执行，总耗时： " + (endTime-startTime) + "ms");

        // 优雅关闭线程池
        executor.shutdown();
    }
}

示例参考[2]做了参数上的改动，上面的示例中，我们定义了任务的总数量，构造线程池时，定义了 corePoolSize/maximumPoolSize，以及数组形式的阻塞队列，接着构造一个任务线程，这个线程主要工作就是休眠500ms，然后统计已执行的任务数量，最后任务全部执行结束耗费的时间情况。

所以从上面的示例可以总结到，线程池的应用实际简单：

// 1.定义线程池
ThreadPoolExecutor executor = new ThreadPoolExecutor(...);

// 2.定义任务，提交任务
Thread task = new Thread(...);
executor.execute(thread);

// 3.关闭线程池
executor.shutdown();

2.ThreadPoolExecutor类说明

先看看这个类相关的类关系图：

我们主要关注：

• Executor接口
• ExecutorService接口
• ThreadPoolExecutor类

Executor接口

public interface Executor {

    /**
     * Executes the given command at some time in the future.  The command
     * may execute in a new thread, in a pooled thread, or in the calling
     * thread, at the discretion of the {@code Executor} implementation.
     *
     * @param command the runnable task
     * @throws RejectedExecutionException if this task cannot be
     * accepted for execution
     * @throws NullPointerException if command is null
     */
    void execute(Runnable command);
}

在这个接口中声明了 void execute(Runnable command)方法，从库的注释来看，就是通过这个方法执行传入的 Runnable 的task，如下面这个官方注释 simple case：

 class DirectExecutor implements Executor {    
     public void execute(Runnable r) {      
         r.run();  
         // 或者下面通过Thread调用start
         // new Thread(r).start();
     }  
 }

ExecutorService接口

通过IDE查看接口的 Structure:

可以看到 ExecutorService接口 在继承接口的同时，扩展了很多声明：

• void shutdown()，关闭线程池，不接受新任务，任务队列的任务将被执行完
• List shutdownNow()，停止所有执行的任务，返回尚待执行的任务列表
• boolean isShutdown()，如果执行器关闭返回true
• boolean isTerminated()，如果关闭后所有任务都已完成，则为True
• Future submint(Callable task)，提交一个待执行的任务，返回表示这个task的Future类型，前面在学习线程创建的三种类型中学习了 Callable接口实现的 线程，后面可以通过调用 get() 获取执行结果

ThreadPoolExecutor类

从类中的结构来看，主要就是：

• 1.内部拒绝策略
• 2.类的构造方法
• 3.线程池的调度执行
• 4.内部属性的设置与获取

构造方法

在线程池的构造方法有四种：

// 1.除基本参数，加入任务阻塞队列    
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), defaultHandler);
}

// 2.在上面基础上加上ThreadFactory    
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         threadFactory, defaultHandler);
}

// 3.在1的基础上加上拒绝策略     
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          RejectedExecutionHandler handler) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), handler);
}

// 4.在1的基础上加上2/3的多添加的参数    
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;
}    

2.1 构造参数

在构造方法中的主要参数如下[4]：

• corePoolSize（必需）：核心线程数。默认情况下，核心线程会一直存活，但是当将allowCoreThreadTimeout设置为true时，核心线程也会超时回收。

• maximumPoolSize（必需）：线程池所能容纳的最大线程数。当活跃线程数达到该数值后，后续的新任务将会阻塞。

• keepAliveTime（必需）：线程闲置超时时长。如果超过该时长，非核心线程就会被回收。如果将allowCoreThreadTimeout设置为true时，核心线程也会超时回收。

• unit（必需）：指定keepAliveTime参数的时间单位。常用的有：TimeUnit.MILLISECONDS（毫秒）、TimeUnit.SECONDS（秒）、TimeUnit.MINUTES（分）。

• workQueue（必需）：任务队列。通过线程池的execute()方法提交的Runnable对象将存储在该参数中。

• threadFactory（可选）：线程工厂。用于指定为线程池创建新线程的方式。

• handler（可选）：拒绝策略。当达到最大线程数时需要执行的饱和策略。

2.2 任务队列-BlockingQueue

任务队列是基于阻塞队列实现的，即采用生产者消费者模式，在Java中需要实现BlockingQueue接口。但Java已经为我们提供了7种阻塞队列的实现[4]：

1. ArrayBlockingQueue：一个由数组结构组成的有界阻塞队列（数组结构可配合指针实现一个环形队列）。
2. LinkedBlockingQueue： 一个由链表结构组成的有界阻塞队列，在未指明容量时，容量默认为Integer.MAX_VALUE。
3. PriorityBlockingQueue： 一个支持优先级排序的无界阻塞队列，对元素没有要求，可以实现Comparable接口也可以提供Comparator来对队列中的元素进行比较。跟时间没有任何关系，仅仅是按照优先级取任务。
4. DelayQueue：类似于PriorityBlockingQueue，是二叉堆实现的无界优先级阻塞队列。要求元素都实现Delayed接口，通过执行时延从队列中提取任务，时间没到任务取不出来。
5. SynchronousQueue： 一个不存储元素的阻塞队列，消费者线程调用take()方法的时候就会发生阻塞，直到有一个生产者线程生产了一个元素，消费者线程就可以拿到这个元素并返回；生产者线程调用put()方法的时候也会发生阻塞，直到有一个消费者线程消费了一个元素，生产者才会返回。
6. LinkedBlockingDeque： 使用双向队列实现的有界双端阻塞队列。双端意味着可以像普通队列一样FIFO（先进先出），也可以像栈一样FILO（先进后出）。
7. LinkedTransferQueue： 它是ConcurrentLinkedQueue、LinkedBlockingQueue和SynchronousQueue的结合体，但是把它用在ThreadPoolExecutor中，和LinkedBlockingQueue行为一致，但是是无界的阻塞队列。

注意有界队列和无界队列的区别：如果使用有界队列，当队列饱和时并超过最大线程数时就会执行拒绝策略；而如果使用无界队列，因为任务队列永远都可以添加任务，所以设置maximumPoolSize没有任何意义。

2.3 线程工厂-ThreadFactory

该参数非必需，Executors框架已经为我们实现了一个默认的线程工厂：

private static class DefaultThreadFactory implements ThreadFactory {
    private static final AtomicInteger poolNumber = new AtomicInteger(1);
    private final ThreadGroup group;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;

    DefaultThreadFactory() {
        @SuppressWarnings("removal")
        SecurityManager s = System.getSecurityManager();
        group = (s != null) ? s.getThreadGroup() :
                              Thread.currentThread().getThreadGroup();
        namePrefix = "pool-" +
                      poolNumber.getAndIncrement() +
                     "-thread-";
    }

    public Thread newThread(Runnable r) {
        Thread t = new Thread(group, r,
                              namePrefix + threadNumber.getAndIncrement(),
                              0);
        if (t.isDaemon())
            t.setDaemon(false);
        if (t.getPriority() != Thread.NORM_PRIORITY)
            t.setPriority(Thread.NORM_PRIORITY);
        return t;
    }
}

2.4 拒绝策略

拒绝策略就是当队列满时，线程如何处理新来的任务，在 ThreadPoolExecutor类 内部实现了以下四种策略[3]：

AbortPolicy（中止策略）-默认

• 功能：当触发拒绝策略时，直接抛出拒绝执行的异常
• 使用场景：ThreadPoolExecutor中默认的策略就是AbortPolicy，由于ExecutorService接口的系列ThreadPoolExecutor都没有显示的设置拒绝策略，所以默认的都是这个。

CallerRunsPolicy（调用者运行策略）

• 功能：只要线程池没有关闭，就由提交任务的当前线程处理。
• 使用场景：一般在不允许失败、对性能要求不高、并发量较小的场景下使用。

DiscardPolicy（丢弃策略）

• 功能：直接丢弃这个任务，不触发任何动作
• 使用场景： 提交的任务无关紧要，一般用的少。

DiscardOldestPolicy（弃老策略）

• 功能：抛弃下一个将要被执行的任务，相当于排队的时候把第一个人打死，然后自己代替
• 使用场景：发布消息、修改消息类似场景。当老消息还未执行，此时新的消息又来了，这时未执行的消息的版本比现在提交的消息版本要低就可以被丢弃了。

当然如果我们实现下面的接口，也可以用自定义的拒绝策略：

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

2.5 执行流程与原理

首先看看该类的 void execute(Runnable command)源码[5]：

public void execute(Runnable command) {
    	// 如果当前任务为空，抛出空指针异常
        if (command == null)
            throw new NullPointerException();
    	// 获取当前线程池的状态+线程个数变量的组合值
        int c = ctl.get();
    	// 如果当前线程数小于核心线程池大小，那么就创建线程并执行当前任务
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
    	// 如果线程池处于RUNNING状态，把任务添加到阻塞队列
        if (isRunning(c) && workQueue.offer(command)) {
            // 二次检查
            int recheck = ctl.get();
            // 如果当前线程池状态不是RUNNING则从队列中删除任务，并且执行线程池的拒绝策略
            if (! isRunning(recheck) && remove(command))
                reject(command);
            // 如果当前线程池为空，则创建一个线程
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
    	// 如果队列满了，则新增线程，新增线程失败则执行线程池的拒绝策略
        else if (!addWorker(command, false))
            reject(command);
    }
    

当提交新任务到线程池时，线程池的处理如下：

• 1、线程池判断核心线程池里的线程是否都在执行任务。如果不是，则创建一个新的工作线程来执行任务。如果核心线程池里的线程都在执行任务，则进入下个流程。

• 2、线程池判断工作队列是否已经满。如果工作队列没有满，则将新提交的任务存储在这个工作队列里。如果工作队列满了，则进入下个流程。

• 3、线程池判断线程池的线程是否都处于工作状态。如果没有，则创建一个新的工作线程来执行任务。如果已经满了，则交给饱和策略来处理这个任务。

ThreadPoolExecutor类的execute()方法的设计思路：

• 1、如果当前线程池运行的线程少于corePoolSize，那么就创建一个线程来执行提交过来的任务。(此操作需要获取全局锁)

• 2、如果当前线程池运行的线程等于或者大于corePoolSize，那么提交过来任务会被放置在BlockingQueue(阻塞队列)中。

• 3、如果BlockingQueue(阻塞队列)已满，那么就创建新的线程来处任务(此操作需要获取全局锁)

• 4、如果撞见新的线程之后，线程池中的线程数大于maximumPoolSize，提交过来的任务会被拒绝执行，并执行拒绝策略

也可以理解为以下执行流程：

执行的先后顺序：
corePool -> BlockingQueue -> maximumPool -> RejectedPolicy

3.功能线程池

除了上面的 ThreadPoolExecutor类，JUC 也为我们封装好了几种常见的功能线程池[4]：

• 定长线程池（FixedThreadPool）
• 定时线程池（ScheduledThreadPool ）
• 可缓存线程池（CachedThreadPool）
• 单线程化线程池（SingleThreadExecutor）

FixedThreadPool

创建方法的源码：

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}
public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>(),
                                  threadFactory);
}
    

• 特点：只有核心线程，线程数量固定，执行完立即回收，任务队列为链表结构的有界队列。
• 应用场景：控制线程最大并发数。

使用示例：

// 1. 创建定长线程池对象 & 设置线程池线程数量固定为3
ExecutorService fixedThreadPool = Executors.newFixedThreadPool(3);
// 2. 创建好Runnable类线程对象 & 需执行的任务
Runnable task =new Runnable(){
  public void run() {
     System.out.println("执行任务啦");
  }
};
// 3. 向线程池提交任务
fixedThreadPool.execute(task);
    

ScheduledThreadPool

创建方法的源码：

   private static final long DEFAULT_KEEPALIVE_MILLIS = 10L;

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}
public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE,
          DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
          new DelayedWorkQueue());
}

public static ScheduledExecutorService newScheduledThreadPool(
        int corePoolSize, ThreadFactory threadFactory) {
    return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
}
public ScheduledThreadPoolExecutor(int corePoolSize,
                                   ThreadFactory threadFactory) {
    super(corePoolSize, Integer.MAX_VALUE,
          DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
          new DelayedWorkQueue(), threadFactory);
}
 

• 特点：核心线程数量固定，非核心线程数量无限，执行完闲置10ms后回收，任务队列为延时阻塞队列。
• 应用场景：执行定时或周期性的任务。

使用示例：

// 1. 创建 定时线程池对象 & 设置线程池线程数量固定为5
ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);
// 2. 创建好Runnable类线程对象 & 需执行的任务
Runnable task =new Runnable(){
  public void run() {
     System.out.println("执行任务啦");
  }
};
// 3. 向线程池提交任务
scheduledThreadPool.schedule(task, 1, TimeUnit.SECONDS); // 延迟1s后执行任务
scheduledThreadPool.scheduleAtFixedRate(task,10,1000,TimeUnit.MILLISECONDS);// 延迟10ms后、每隔1000ms执行任务

CachedThreadPool

创建方法的源码：

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}
public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>(),
                                  threadFactory);
}    

• 特点：无核心线程，非核心线程数量无限，执行完闲置60s后回收，任务队列为不存储元素的阻塞队列。
• 应用场景：执行大量、耗时少的任务。

使用示例：

// 1. 创建可缓存线程池对象
ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
// 2. 创建好Runnable类线程对象 & 需执行的任务
Runnable task =new Runnable(){
  public void run() {
     System.out.println("执行任务啦");
  }
};
// 3. 向线程池提交任务
cachedThreadPool.execute(task);    

SingleThreadExecutor

创建方法的源码：

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}
public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>(),
                                threadFactory));
}   

• 特点：只有1个核心线程，无非核心线程，执行完立即回收，任务队列为链表结构的有界队列。
• 应用场景：不适合并发但可能引起IO阻塞性及影响UI线程响应的操作，如数据库操作、文件操作等。

使用示例：

// 1. 创建单线程化线程池
ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
// 2. 创建好Runnable类线程对象 & 需执行的任务
Runnable task =new Runnable(){
  public void run() {
     System.out.println("执行任务啦");
  }
};
// 3. 向线程池提交任务
singleThreadExecutor.execute(task);

参考：

• 1.图解Java多线程
• 2.一个demo让你彻底理解线程池工作流程
• 3.一文搞懂线程池
• 4.Java多线程：彻底搞懂线程池
• 5.ThreadPoolExecutor的执行流程和原理

延伸阅读：

• java线程池ThreadPoolExecutor类使用详解

Queue任务队列/ThreadFactory等

• 线程池大小/线程数量

本栏目推荐文章
• 深入探索JVM：理解Java程序在虚拟机中的存储和管理
• 深入理解spring框架：剖析多线程模式下数据库连接
• 用jacoco统计JAVA项目测试代码覆盖率
• Java实现在线编辑预览office文档
• Java小细节之缩进中if 和else的对应关系
• 春眠不觉晓，Java数据类型知多少？基础牢不牢看完本文就有数了
• idea 中java代码修改后运行代码不生效
• Java日期时间处理详解
• java项目(jar包)配置为开机自启 jar包bat脚本启动和配置为开机自启
• Java中锁的解决方案

线程 小结 Java线程 小结java 知识点 线程 小结 知识 java小结 容器 基础 小结junit5 junit java 小结spring java boot 下标 数组 小结java 小结 程序java 运算符 小结java 线程java threadpoolexecutor线程java