Java线程池详解

2023-01-31 00:00:00 创建 执行 线程 队列 状态

一、线程池有哪些优点

  1. 减少资源开销,不用频繁的创建和销毁线程
  2. 提高响应时长,有任务时可直接执行。
  3. 提高线程的可管理性,所有线程资源都由线程池统一管理。

二、线程池的主要参数

线程池ThreadPoolExecutor的继承关系:


想知道有哪些参数,先看参数多的构造函数:

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    if (corePoolSize <  ||
        maximumPoolSize <=  ||
        maximumPoolSize < corePoolSize ||
        keepAliveTime < )
        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;
}

可以看到创建一个线程池需要七个参数。

  1. corePoolSize:线程池核心线程数量大小,有新任务进来时,如果线程池中的线程数小于这个值,则会创建新的线程来执行任务(不管有没有空闲线程,都会创建),直到线程数量大于等于这个数量。
  2. maximumPoolSize:线程池大线程数量大小(当达到核心线程数,且队列任务已满,会增加线程至大线程数)。
  3. keepAliveTime:线程的大空余时间,大于这个时间将被回收(线程数大于核心线程数时,多余的线程空闲时长达到这个值就会被回收)。
  4. unit:空余时长的单位。
  5. workQueue:阻塞队列,当核心线程数已满,任务会被放到这个队列中。
  6. threadFactory:线程工厂,线程池中的线程都是由这个线程工厂创建的,线程池提供了默认的线程工厂。
  7. handler:拒绝策略,当队列任务已满,且线程数量达到大线程数,新进入的任务会执行这个拒绝策略来选择丢弃哪个任务,线程池提供了四种默认的拒绝策略。

线程工厂ThreadFactor,线程池中默认给定了一个线程工厂DefaultThreadFactory,线程工厂的作用是用于创建线程,自己创建线程工厂时需要实现ThreadFactor接口,该接口中只有一个方法Thread newThread(Runnable r);,即创建线程的方法,自定义线程工厂的好处是可以自定义线程名称。

阻塞队列BlockingQueue,阻塞队列是在核心线程满了以后存放任务使用,常用的有LinkedBlockingQueueArrayBlockingQueueSynchronousQueueDelayedWorkQueue等,当阻塞队列也满时,会创建线程至大线程数,如果队列已满,也达到大线程数,则执行拒绝策略。

拒绝策略RejectedExecutionHandler

  • DiscardPolicy:丢弃该任务,不抛异常。
  • DiscardOldestPolicy:丢弃早加入队列的任务,不抛异常。
  • AbortPolicy:丢弃该任务并抛出异常RejectedExecutionException,默认使用该策略。
  • CallerRunsPolicy:由调用线程池的线程来执行当前任务。

查看源码:

public static class DiscardPolicy implements RejectedExecutionHandler {
    public DiscardPolicy() { }
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    }
}
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
    public DiscardOldestPolicy() { }
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            e.getQueue().poll();
            e.execute(r);
        }
    }
}
public static class AbortPolicy implements RejectedExecutionHandler {
    public AbortPolicy() { }
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        throw new RejectedExecutionException("Task " + r.toString() +
                                             " rejected from " +
                                             e.toString());
    }
}
public static class CallerRunsPolicy implements RejectedExecutionHandler {
    public CallerRunsPolicy() { }
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            r.run();
        }
    }
}

如果想要实现自己的拒绝策略,那么实现RejectedExecutionHandler接口即可。

三、线程池的执行流程

如下图:

查看源码执行流程,execute()方法:

public void execute(Runnable command) {
    //传入的任务是否为空
    if (command == null)
        throw new NullPointerException();
    /*
     * Proceed in 3 steps:
     *
     * 1. 如果运行的线程数小于corePoolSize,则创建新的线程(线程池状态为正在运行的状态)。
     *
     * 2. 如果任务排队成功,仍然需要检查线程池的状态,如果不是可运行的状态则回滚刚刚的操作。
     *
     * 3. 如果排队失败,则尝试添加新的线程,如果已经达到线程池大数量,则执行拒绝策略。
     */
     //判断线程数是否小于核心线程数
    int c = ctl.get();
    if (workerCountOf(c) < corePoolSize) {
        //如果小于,新增一个线程来执行
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    //如果核心线程数已满,则向阻塞队列中添加任务
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            reject(command);
        else if (workerCountOf(recheck) == )
            addWorker(null, false);
    }
    //如果添加失败,则创建线程至大值
    else if (!addWorker(command, false))
        //如果创建失败,则执行拒绝策略
        reject(command);
}

四、线程池的状态

查看源码,发现线程池有五种状态,如下:

// runState is stored in the high-order bits
private static final int RUNNING    = -1 << COUNT_BITS;
private static final int SHUTDOWN   =   << COUNT_BITS;
private static final int STOP       =  1 << COUNT_BITS;
private static final int TIDYING    =  2 << COUNT_BITS;
private static final int TERMINATED =  3 << COUNT_BITS;
  1. RUNNING,运行状态,线程池创建完成后就是运行状态。
  2. SHUTDOWN,关闭状态,执行shutdown()方法后进入此状态,继续处理队列中的任务,但是不再接收新的任务。
  3. STOP,停止状态,shutdownNow()方法后进入此状态,不处理队列中的任务,也不接收新的任务。
  4. TIDYING,整理状态,运行的线程数为0,队列中任务为空时,则进入此状态,进入此状态后会执行terminated()方法,进入销毁状态。
  5. TERMINATED,销毁状态,执行terminated()方法,进入此状态。

状态转换如下图:

五、Java提供的快速创建的线程池

Executors提供了几种快速创建的线程池:

  • newSingleThreadExecutor,只有一个线程的线程池
public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}
  • newFixedThreadPool,固定线程数量的线程池
public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}
  • newCachedThreadPool,可缓存的线程池,即不限制线程数量的线程池。
public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}
  • newScheduledThreadPool,定时线程池,可周期性或延迟执行任务的线程池(使用延时队列)。
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE, , NANOSECONDS,
          new DelayedWorkQueue());
}
  • newSingleThreadScheduledExecutor,单个线程的定时线程池,功能和上面一样。
public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
    return new DelegatedScheduledExecutorService
        (new ScheduledThreadPoolExecutor(1));
}

相关文章