线程池

参考：

• Java并发编程：线程池的使用
• Java线程池的分析和使用

基础概念

线程池：线程池就是创建大量空闲的线程并存入一个容器中；程序将一个任务传给线程池，线程池就会启动一条线程来执行这个任务，任务执行完成后，该线程又返回线程池中成为空闲线程，等待执行下一个任务。

• 根据系统的环境情况，可以自动或手动设置线程数量，达到运行的最佳效果。少了浪费系统资源，多了造成系统拥挤效率不高。
• 可以通过线程池控制线程数量，若待执行的任务数超过线程池的核心线程数，则排队等候。
• 一个任务执行完毕，再从队列的中取最前面的任务开始执行。若队列中没有等待进程，线程池就一直等待。
• 当一个新任务传给线程池时，如果线程池中有等待的工作线程，就立刻使用等待中的工作线程执行该任务；否则任务进入等待队列。

线程池的特点：线程复用；控制最大并发数；管理线程

线程池的作用：避免频繁的创建线程和销毁线程，提高程序的效率

• 减少了创建和销毁线程的次数，每个工作线程都可以被重复利用，可执行多个任务。
• 可以根据系统的承受能力，调整线程池中工作线线程的数目，防止因为消耗过多的内存，而把服务器累趴下（每个线程需要大约1MB内存，线程开的越多，消耗的内存也就越大，最后死机）。

ThreadPoolExecutor（线程池）的执行顺序

• 当线程数小于核心线程数时，会一直创建线程直到线程数等于核心线程数；
• 当线程数等于核心线程数时，新加入的任务会被放到任务队列等待执行；
• 当任务队列已满，又有新的任务时，会创建线程直到线程数量等于最大线程数；
• 当线程数等于最大线程数，且任务队列已满时，新加入任务会被拒绝。

一个线程池包括以下四个基本组成部分：

• 线程池管理器（ThreadPool）：用于创建并管理线程池，包括 创建线程池，销毁线程池，添加新任务；
• 工作线程（PoolWorker）：线程池中线程，在没有任务时处于等待状态，可以循环的执行任务；
• 任务接口（Task）：每个任务必须实现的接口，以供工作线程调度任务的执行，它主要规定了任务的入口，任务执行完后的收尾工作，任务的执行状态等；
• 任务队列（taskQueue）：用于存放没有处理的任务。提供一种缓冲机制。

线程池接口的体系结构

线程池的体系结构：

java.util.concurrent.Executor：负责线程的使用和调度的根接口

​ |--ExecutorService 子接口： 线程池的主要接口，提供了线程池生命周期方法

​ |--ThreadPoolExecutor： 线程池的实现类，提供了线程池的维护操作等相关方法

​ |--ScheduledThreadPoolExecutor ： 继承 ThreadPoolExecutor，实现了 ScheduledExecutorService

​ 周期性任务调度的类实现

线程池的常用方法

ExecutorService：JDK提供的线程池

java.util.concurrent.Executor：描述线程池的顶级接口

唯一方法：

void execute(Runnable command)    // 提交一个 Runnable 任务用于执行

java.util.concurrent.ExecutorService：描述线程池的接口

常用方法：

Future<?> submit(Runnable task)        // 提交一个 Runnable 任务用于执行，并返回一个表示该任务的 Future
<T> Future<T> submit(Callable<T> task)    // 提交一个返回值的任务用于执行，返回一个表示任务的未决结果的 Future
/*参数:
    Runnable task：传递Runnable接口的实现类对象(设置的线程任务) ==> 重写run方法，设置线程任务 ==> run方法没有返回值
    Callable<T> task：传递Callable接口的实现类对象(设置线程任务) ==> 重写call方法，设置线程任务 ==> call方法有返回值
返回值：
    Future：用来接收线程任务的返回值 ==> 用来接收call方法的返回值
    
注：Callable 接口类似于 Runnable，两者都是为那些其实例可能被另一个线程执行的类设计的。但是 Runnable 不会返回结果，Callable会返回一个结果。
*/

java.util.concurrent.Future：表示异步计算的结果。用来接收 call 方法的返回值

接口中的方法：

V get()        // 方法可以当任务结束后返回一个结果，如果调用时，工作还没有结束，则会阻塞线程，直到任务执行完毕
V get(long timeout, TimeUnit unit)        // 做多等待timeout的时间就会返回结果
    
boolean isDone()        // 判断当前方法是否完成
boolean isCancel()        // 判断当前方法是否取消

boolean cancel(boolean mayInterruptIfRunning)    // 停止一个任务
/*    如果任务可以停止（通过mayInterruptIfRunning来进行判断），则可以返回true，
    如果任务已经完成或者已经停止，或者这个任务无法停止，则会返回false
*/

submit 和 execute 区别：

• execute() 方法中的是 Runnable 接口的实现，所以只能使用 try-catch 来捕获 Checked Exception，通过实现UncaughtExceptionHande 接口处理 UncheckedException， 即 和普通线程的处理方式完全一致

  submit() 方法中抛出异常不管提交的是 Runnable 还是 Callable 类型的任务，如果不对返回值 Future 调用 get() 方法，都会吃掉异常

• execute() 方法提交的未执行的任务可以通过 remove(Runnable) 方法删除

  submit() 方法提交的任务即使还未执行也不能通过 remove(Runnable) 方法删除

ExecutorService 接口的核心成员变量：

// 核心池的大小（即线程池中的线程数目大于这个参数时，提交的任务会被放进任务缓存队列）
private volatile int corePoolSize;
// 线程池中当前的线程数
private volatile int poolSize;
// 线程池最大能容忍的线程数
private volatile int maximumPoolSize;


// 任务缓存队列，用来存放等待执行的任务
private final BlockingQueue<Runnable> workQueue;
// 线程池的主要状态锁，对线程池状态（比如线程池大小、runState等）的改变都要使用这个锁
private final ReentrantLock mainLock = new ReentrantLock();
// 用来存放工作集
private final HashSet<Worker> workers = new HashSet<Worker>();  
// 线程存货时间
private volatile long keepAliveTime;       
// 是否允许为核心线程设置存活时间
private volatile boolean allowCoreThreadTimeOut;
// 任务拒绝策略
private volatile RejectedExecutionHandler handler; 
// 线程工厂，用来创建线程
private volatile ThreadFactory threadFactory;
// 用来记录线程池中曾经出现过的最大线程数
private int largestPoolSize;
// 用来记录已经执行完毕的任务个数
private long completedTaskCount;

java.util.concurrent.Executors：是一个创建线程池的工具类(工厂类)，专门用来生产线程池，里边的方法都是静态的

常用方法：

// 创建一个可重用固定线程数的线程池，以共享的无界队列方式来运行这些线程
static ExecutorService newFixedThreadPool(int nThreads)
/*参数:
    int nThreads：创建线程池，包含线程的数量  100==>线程池中包含100个线程
*/
static ExecutorService newCachedThreadPool()         // 缓存线程池，线程池的数量不固定，可以根据需求自动的更改数量
static ExecutorService newSingleThreadExecutor()     // 创建单个线程池。 线程池中只有一个线程
static ScheduledExecutorService newScheduledThreadPool()     // 创建固定大小的线程，可以延迟或定时的执行任务

// 注意：线程池一旦销毁，就不能在使用了，会抛出异常
void shutdown()  // 关闭线程池，但不会立即终止，而是要等所有任务缓存队列中的任务都执行完后才终止，但再也不会接受新的任务
void shutdownNow()    // 立即终止线程池，并尝试打断正在执行的任务，并且清空任务缓存队列，返回尚未执行的任务

使用示例

public class Test {
     public static void main(String[] args) {   
         ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 200, TimeUnit.MILLISECONDS,
                 new ArrayBlockingQueue<Runnable>(5));
          
         for(int i=0;i<15;i++){
             MyTask myTask = new MyTask(i);
             executor.execute(myTask);
             System.out.println("线程池中线程数目："+executor.getPoolSize()+"，队列中等待执行的任务数目："+
             executor.getQueue().size()+"，已执行玩别的任务数目："+executor.getCompletedTaskCount());
         }
         executor.shutdown();
     }
}

class MyTask implements Runnable {
    private int taskNum;
     
    public MyTask(int num) {
        this.taskNum = num;
    }
     
    @Override
    public void run() {
        System.out.println("正在执行task "+taskNum);
        try {
            Thread.currentThread().sleep(4000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("task "+taskNum+"执行完毕");
    }
}

ThreadPoolTaskExecutor：Spring 提供的线程池

线程池对象交给 Spring 容器管理

• 配置类的方式配置线程池

  @Configuration
  public class ExecturConfig {
      @Bean("threadPoolTaskExecutor")        // 配置多个线程池时，设置不同的实例名称，注入时变量名为相应的实例名称
      public ThreadPoolTaskExecutor threadPoolTaskExecutor() {
          ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
          int i = Runtime.getRuntime().availableProcessors();    //获取到服务器的cpu内核
          executor.setCorePoolSize(5);    //核心池大小
          executor.setMaxPoolSize(100);    //最大线程数
          executor.setQueueCapacity(1000);    //队列程度
          executor.setKeepAliveSeconds(1000);    //线程空闲时间
          executor.setThreadNamePrefix("tsak-asyn");    //线程前缀名称
          executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());    //配置拒绝策略
          executor.initialize();    //线程初始化
          return executor;
      }

• xml 配置的方式创建

  <!-- spring线程池 -->
  <bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
      <!-- 核心线程数  -->
      <property name="corePoolSize" value="10"/>
      <!-- 最大线程数 -->
      <property name="maxPoolSize" value="200"/>
      <!-- 队列最大长度 >=mainExecutor.maxSize -->
      <property name="queueCapacity" value="10"/>
      <!-- 线程池维护线程所允许的空闲时间 -->
      <property name="keepAliveSeconds" value="20"/>
      <!-- 线程池对拒绝任务(无线程可用)的处理策略 -->
      <property name="rejectedExecutionHandler">
          <bean class="java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy"/>
      </property>
  </bean>

使用时通过自动注入的方式获取线程池对象

// 方式1（推荐）
@AutoWired
private ThreadPoolTaskExecutor threadPoolTaskExecutor;

// 方式2
@Resource(name="taskExecutor")
private ThreadPoolTaskExecutor threadPoolTaskExecutor;

打印线程池的实时参数

public static void printThreadPoolStatic(ThreadPoolTaskExecutor threadPoolTaskExecutor){
    log.info("-----------------------thread pool status start-----------------------");
    log.info("thread core pool size:{}", threadPoolTaskExecutor.getCorePoolSize());    // 核心线程数
    log.info("thread active count:{}", threadPoolTaskExecutor.getActiveCount());     // 正在工作的线程数
    log.info("thread pool max size:{}", threadPoolTaskExecutor.getMaxPoolSize());     // 最大线程数  
    log.info("thread pool size:{}", threadPoolTaskExecutor.getPoolSize());    // 线程池中当前的线程数
    log.info("thread pool wait task:{}", threadPoolTaskExecutor.getThreadPoolExecutor().getQueue().size());                                                                             // 任务队列内等待执行的任务数
    log.info("-----------------------thread pool status end-----------------------");
}

线程池的主要参数

// ThreadPoolExecutor类中提供的构造方法
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, 
        BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler);

• corePoolSize ：核心线程数，也可以理解为最小线程数

  ​ （1）核心线程会一直存在，即使没有任务执行

  ​ （2）当线程数小于核心线程数的时候，即使有空闲线程，也会一直创建线程直到达到核心线程数

  ​ （3）设置 allowCoreThreadTimeout=true（默认 false）时，核心线程会超时关闭

• maxPoolSize（maximumPoolSize）：最大线程数

  ​ （1）线程池里允许存在的最大线程数量

  ​ （2）当任务队列已满，且线程数量大于等于核心线程数时，会创建新的线程执行任务

  ​ （3）线程池里允许存在的最大线程数量。当任务队列已满，且线程数量大于等于核心线程数时，会创建新的线程执行任务

• keepAliveTime ：线程空闲时间

  ​ （1）当线程空闲时间达到 keepAliveTime 时，线程会退出（关闭），直到线程数等于核心线程数

  ​ （2）如果设置了 allowCoreThreadTimeout=true，则线程会退出直到线程数等于零

• （BlockingQueue）workQueue：任务队列，用于传输和保存等待执行任务的阻塞队列。阻塞队列有以下几种选择：

  • LinkedBlockingQueue：基于链表的先进先出队列，如果创建时没有指定此队列大小，则默认为 Integer.MAX_VALUE
  • synchronousQueue：不保存提交的任务，而是将直接新建一个线程来执行新来的任务
  • ArrayBlockingQueue：基于数组的先进先出队列，此队列创建时必须指定大小
  • PriorityBlockingQueue：一个具有优先级的无限阻塞队列
• threadFactory：线程工厂，用于创建新线程。

  threadFactory创建的线程也是采用new Thread()方式，threadFactory创建的线程名都具有统一的风格：pool-m-thread-n（m为线程池的编号，n为线程池内的线程编号）

• （RejectedExecutionHandler）handler：线程饱和策略/任务拒绝处理器，当线程池和队列都满了，再加入线程会执行此策略

  • ThreadPoolExecutor.AbortPolicy
    简单粗暴，丢弃任务并抛出RejectedExecutionException异常，这也是默认的拒绝策略。

    即使是submit提交，也是使用 try-catch 捕捉 任务拒绝异常

    try{
        Future<T> future = threadPoolTaskExecutor.submit(() -> {任务});
        T t = future.get();
    } catch(TaskRejectedException e){
        e.printStackTrace();
    } catch(Exception e){
        e.printStackTrace();
    }

  • ThreadPoolExecutor.CallerRunsPolicy
    如果线程池未关闭，则会在调用者线程中直接执行新任务，这会导致主线程提交线程性能变慢。
  • ThreadPoolExecutor.DiscardPolicy
    丢弃任务，但是不抛出异常
  • ThreadPoolExecutor.DiscardOldestPolicy
    抛弃最老的任务，就是从队列取出最老的任务然后放入新的任务进行执行。

• unit：参数keepAliveTime的时间单位，有7种取值，在TimeUnit类中有7种静态属性：

  TimeUnit.DAYS;            // 天
  TimeUnit.HOURS;           // 小时
  TimeUnit.MINUTES;         // 分钟
  TimeUnit.SECONDS;         // 秒
  TimeUnit.MILLISECONDS;    // 毫秒
  TimeUnit.MICROSECONDS;    // 微妙
  TimeUnit.NANOSECONDS;     // 纳秒

排队的三种通用策略：

• 直接提交

  工作队列的默认选项是 SynchronousQueue，它将任务直接提交给线程而不阻塞它们。

  如果不存在可用于立即运行任务的工作线程时，试图把任务加入队列将失败，因此会构造一个新的工作线程。

  此策略可以避免在处理可能具有内部依赖性的请求集时出现锁。

  直接提交策略通常要求无界的 maximumPoolSizes（最大线程数）以避免拒绝新提交的任务。

• 无界队列（队列大小超级大，例如 Integer.MAX_VALUE）

  使用无界队列（例如，使用默认值的 LinkedBlockingQueue），如果所有的工作线程都忙时，新任务将在队列中等待。

  因为队列超级大，很难队列满，线程池中的线程数一般不会超过 corePoolSize（maximumPoolSize 的值也就用不到了）。

  当每个任务完全独立于其他任务，即任务执行互不影响时，适合于使用无界队列。

  例如，在 Web页服务器中。这种排队可用于处理瞬态突发请求，当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。

• 有界队列

  当使用有界队列（如 ArrayBlockingQueue）以及有限的 maximumPoolSizes（最大线程数）时，有助于防止资源耗尽，但是性能调优可能会比较困难。

  • 使用大型队列和小型池

    可以最大限度地降低 CPU 使用率以及操作系统资源和上下文切换的开销，但是可能导致人为的降低吞吐量。

    如果任务执行过程中可能频繁阻塞（例如，I/O 性能瓶颈），则耗时可能大于节约的线程调度开销。

  • 使用小型队列通常要求较大的池大小，CPU使用率较高，但是可能遇到不可接受的调度开销，这样也会降低吞吐量。

常见线程池

常见线程池（实现 ExecutorService 的四种连接池，使用工厂类 Executors 的静态方法创建）：

• newFixedThreadPool：固定线程池

  核心线程数和最大线程数固定相等，空闲存活时间为 0 毫秒（说明此参数无意义），工作队列为最大为 Integer.MAX_VALUE 大小的阻塞队列。

  当执行任务时，如果线程都很忙，就会丢到工作队列等有空闲线程时再执行，队列满就执行默认的拒绝策略。

• newCachedThreadPool：带缓冲线程池

  核心线程数为 0，最大线程数为 Integer.MAX_VALUE，超过 0 个的空闲线程在 60 秒后销毁，SynchronousQueue（直接提交的队列）

  每个新任务都会有线程来执行，如果线程池有可用线程则执行任务，没有的话就创建一个来执行，线程池中的线程数不确定，一般建议执行速度较快的线程，不然这个最大线程池边界过大容易造成内存溢出。

• newSingleThreadExecutor：单线程线程池

  核心线程数和最大线程数均为1，空闲线程存活 0 毫秒

  每次只执行一个线程，多余的先存储到工作队列，一个一个执行，保证了线程的顺序执行。

  如果这个唯一的线程因为异常结束，那么会有一个新的线程来替代它。

• newScheduledThreadPool：调度线程池

  大小无限制的线程池，支持定时和周期性的执行线程

线程池的合理配置分析

如何配置线程池，可以从以下几个角度来进行分析：

• 任务的性质：CPU 密集型任务，IO 密集型任务和混合型任务

  任务性质不同的任务可以用不同规模的线程池分开处理：

  • CPU 密集型任务：配置尽可能少的线程数量，如配置 Ncpu+1 个线程的线程池
  • IO 密集型任务：由于需要等待 IO 操作，线程并不是一直在执行任务，则配置尽可能多的线程，如 2*Ncpu
  • 混合型的任务：如果可以拆分，则将其拆分成一个 CPU 密集型任务和一个 IO 密集型任务

    只要这两个任务执行的时间相差不是太大，那么分解后执行的吞吐率要高于串行执行的吞吐率

    如果这两个任务执行时间相差太大，则没必要进行分解

  注：可以通过 Runtime.getRuntime().availableProcessors() 方法获得当前设备的CPU个数

• 任务的优先级：高、中和低

  优先级不同的任务：可以使用优先级队列 PriorityBlockingQueue 来处理

  优先级队列可以让优先级高的任务先得到执行

  需要注意的是如果一直有优先级高的任务提交到队列里，那么优先级低的任务可能永远不能执行。

• 任务的执行时间：长、中和短

  执行时间不同的任务：

  • 可以交给不同规模的线程池来处理
  • 也可以使用优先级队列，让执行时间短的任务先执行

• 任务的依赖性：是否依赖其他系统资源，如数据库连接

  由于线程提交 SQL 后需要等待数据库返回结果，等待的时间越长，CPU 空闲时间就越长

  如果等待的时间较长，则线程数应该设置大一点，这样才能更好的利用 CPU

  • 建议使用有界队列，有界队列能增加系统的稳定性和预警能力，可以根据需要设置大一点，比如几千
  • 如果设置成无界队列，当数据库出现了问题，导致执行 SQL 变得非常缓慢，进而导致线程池里的工作线程全部阻塞住，任务积压在线程池里，线程池的队列就会越来越多，有可能会撑满内存，导致整个系统不可用。

线程池的监控

• 可以通过线程池提供的参数进行监控

  线程池里有一些属性在监控线程池的时候可以使用

  • taskCount：线程池需要执行的任务数量。
  • completedTaskCount：线程池在运行过程中已完成的任务数量。小于或等于 taskCount。
  • largestPoolSize：线程池曾经创建过的最大线程数量。通过这个数据可以知道线程池是否满过。如等于线程池的最大大小，则表示线程池曾经满了。
  • getPoolSize：线程池的线程数量。如果线程池不销毁的话，池里的线程不会自动销毁，所以这个大小只增不减。
  • getActiveCount：获取活动的线程数。

• 可以通过扩展线程池进行监控

  通过继承线程池并重写线程池的 beforeExecute，afterExecute 和 terminated 方法

  可以在任务执行前，执行后和线程池关闭前干一些事情。如监控任务的平均执行时间，最大执行时间和最小执行时间等。

  这几个方法在线程池里是空方法。如：

  protected void beforeExecute(Thread t, Runnable r) { }`

定时器

定时器：可以设置线程在某个时间执行某件事情，或者某个时间开始，每间隔指定的时间反复的做某件事情

java.util.Timer 类：

• 是一个描述定时器的类
• 一种工具，线程用其安排以后在后台线程中执行的任务
• 可安排任务执行一次，或者定期重复执行

构造方法：

public Timer()         // 创建一个新计时器

成员方法：

void schedule(TimerTask task, long delay) // 在指定的毫秒值之后，执行指定的任务，只会执行一次
     // 参数:
     //     task - 所要安排的任务。定时器到时间之后要执行的任务
     //     delay - 执行任务前的延迟时间，单位是毫秒。 多少毫秒之后开始执行TimerTask任务
void schedule(TimerTask task, long delay, long period)
    // 在指定的毫秒值之后,执行指定的任务，之后每隔固定的毫秒数重复执行定时任务
    // 参数:
    //     period - 执行各后续任务之间的时间间隔，单位是毫秒。定时器开始执行之后,每隔多少毫秒重复执行
void schedule(TimerTask task, Date time) // 安排在指定的时间执行指定的任务，只会执行一次
    // 参数:
    //     time - 执行任务的时间。从什么日期开始执行任务  20020-07-06 15:25:13
void schedule(TimerTask task, Date firstTime, long period) // 安排指定的任务在指定的时间开始进行重复的固定延迟执行。

void cancel()     // 终止此计时器，丢弃所有当前已安排的任务。
    // 注意，在此计时器调用的计时器任务的 run 方法内调用此方法，就可以绝对确保正在执行的任务是此计时器所执行的最后一个任务。

java.util.TimerTask类 implements Runnable接口：由 Timer 安排为一次执行或重复执行的任务

TimerTask 类是一个抽象类，无法直接创建

void run()        // 此计时器任务要执行的操作。重写run方法,设置线程任务