Java并发编程札记(六)JUC线程池02ThreadPoolExecutor实现原理

本文通过学习ThreadPoolExecutor源码来学习线程池的实现原理。

简介

为什么要使用线程池
许多服务器都面临着处理大量客户端远程请求的压力&＃xff0c;如果每收到一个请求&＃xff0c;就创建一个线程来处理&＃xff0c;表面看是没有问题的&＃xff0c;但实际上存在着很严重的缺陷。服务器应用程序中经常出现的情况是请求处理的任务很简单但客户端的数目却是庞大的&＃xff0c;这种情况下如果还是每收到一个请求就创建一个线程来处理它&＃xff0c;服务器在创建和销毁线程所花费的时间和资源可能比处理客户端请求处理的任务花费的时间和资源更多。为了缓解服务器压力&＃xff0c;需要解决频繁创建和销毁线程的问题。线程池可以实现这个需求。

什么是线程池
线程池可以看做是许多线程的集合。在没有任务时线程处于空闲状态&＃xff0c;当请求到来&＃xff0c;线程池给这个请求分配一个空闲的线程&＃xff0c;任务完成后回到线程池中等待下次任务。这样就实现了线程的重用。线程池会通过相应的调度策略和拒绝策略&＃xff0c;对添加到线程池中的线程进行管理。

工作模型

工作模型中一共有三种队列&＃xff1a;正在执行的任务队列&＃xff0c;等待被执行的阻塞队列&＃xff0c;等待被commit进阻塞队列中的任务队列。

Java中的线程池
Java中常用的线程池有三个&＃xff0c;最出名的当然是ThreadPoolExecutor&＃xff0c;除此之外还有ScheduledThreadPoolExecutor、ForkJoinPool。本文主要学习ThreadPoolExecutor的实现原理。

创建ThreadPoolExecutor线程池

强烈推荐使用Executors工厂方法创建线程池&＃xff0c;如Executors.newCachedThreadPool()&＃xff08;无界线程池&＃xff0c;可以进行自动线程回收&＃xff09;、Executors.newFixedThreadPool(int)&＃xff08;固定大小线程池&＃xff09;和 Executors.newSingleThreadExecutor()&＃xff08;单个后台线程&＃xff09;&＃xff0c;它们均为大多数使用场景预定义了设置。本文主要学习如何手动配置&＃xff0c;下面是ThreadPoolExecutor的一个构造方法。

public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) {if (corePoolSize <0 ||maximumPoolSize <&＃61; 0 ||maximumPoolSize 0)throw new IllegalArgumentException();if (workQueue &＃61;&＃61; null || threadFactory &＃61;&＃61; null || handler &＃61;&＃61; null)throw new NullPointerException();this.corePoolSize &＃61; corePoolSize;this.maximumPoolSize &＃61; maximumPoolSize;this.workQueue &＃61; workQueue;this.keepAliveTime &＃61; unit.toNanos(keepAliveTime);this.threadFactory &＃61; threadFactory;this.handler &＃61; handler;
}

ThreadPoolExecutor一共有四个构造方法&＃xff0c;其他三个构造方法都是通过上述的构造方法来实现的。毫无疑问手动配置线程池的关键就是学好构造方法中的几个参数如何设置。这几个参数对应着ThreadPoolExecutor中的几个成员属性。

属性

corePoolSize与maximumPoolSize分别是核心池大小与最大池大小。在源码中的声明为
private volatile int corePoolSize;private volatile int maximumPoolSize;
当新任务在方法 execute(java.lang.Runnable) 中提交时&＃xff0c;如果运行的线程少于 corePoolSize&＃xff0c;则创建新线程来处理请求&＃xff0c;即使其他辅助线程是空闲的。如果运行的线程多于 corePoolSize 而少于 maximumPoolSize&＃xff0c;则仅当队列满时才创建新线程。如果设置的 corePoolSize 和 maximumPoolSize 相同&＃xff0c;则创建了固定大小的线程池。如果将 maximumPoolSize 设置为基本的无界值&＃xff08;如 Integer.MAX_VALUE&＃xff09;&＃xff0c;则允许池适应任意数量的并发任务。在大多数情况下&＃xff0c;核心和最大池大小仅基于构造来设置&＃xff0c;不过也可以使用 setCorePoolSize(int) 和 setMaximumPoolSize(int) 进行动态更改。

workQueue是线程池工作模型中的阻塞队列&＃xff0c;用于传输和保持提交的任务。在源码中的声明为private final BlockingQueue workQueue;。

keepAliveTime是池中线程空闲时的活动时间。如果池中当前有多于 corePoolSize 的线程&＃xff0c;则这些多出的线程在空闲时间超过 keepAliveTime 时将会终止&＃xff08;参见 getKeepAliveTime(java.util.concurrent.TimeUnit)&＃xff09;。这提供了当池处于非活动状态时减少资源消耗的方法。如果池后来变得更为活动&＃xff0c;则可以创建新的线程。也可以使用方法 setKeepAliveTime(long, java.util.concurrent.TimeUnit) 动态地更改此参数。使用 Long.MAX_VALUE TimeUnit.NANOSECONDS 的值在关闭前有效地从以前的终止状态禁用空闲线程。默认情况下&＃xff0c;保持活动策略只在有多于 corePoolSizeThreads 的线程时应用。但是只要 keepAliveTime 值非 0&＃xff0c; allowCoreThreadTimeOut(boolean) 方法也可将此超时策略应用于核心线程。

threadFactory是一个线程集合。线程池可以使用ThreadFactory创建新线程。如果没有另外说明&＃xff0c;则在同一个 ThreadGroup 中一律使用 Executors.defaultThreadFactory() 创建线程&＃xff0c;并且这些线程具有相同的 NORM_PRIORITY 优先级和非守护进程状态。通过提供不同的 ThreadFactory&＃xff0c;可以改变线程的名称、线程组、优先级、守护进程状态&＃xff0c;等等。如果从 newThread 返回 null 时 ThreadFactory 未能创建线程&＃xff0c;则执行程序将继续运行&＃xff0c;但不能执行任何任务。

handler是线程池拒绝策略&＃xff0c;RejectedExecutionHandler类型的对象。当 Executor 已经关闭&＃xff0c;并且 Executor 将有限边界用于最大线程和工作队列容量&＃xff0c;且已经饱和时&＃xff0c;在方法 execute(java.lang.Runnable) 中提交的新任务将被拒绝。在以上两种情况下&＃xff0c; execute 方法都将调用其 RejectedExecutionHandler 的 RejectedExecutionHandler.rejectedExecution(java.lang.Runnable, java.util.concurrent.ThreadPoolExecutor) 方法。下面提供了四种预定义的处理程序策略&＃xff1a;

• ThreadPoolExecutor.AbortPolicy &＃xff0c;默认策略&＃xff0c;处理程序遭到拒绝将抛出运行时 RejectedExecutionException。
• ThreadPoolExecutor.CallerRunsPolicy&＃xff0c;线程调用运行该任务的 execute 本身。此策略提供简单的反馈控制机制&＃xff0c;能够减缓新任务的提交速度。
• ThreadPoolExecutor.DiscardPolicy&＃xff0c;不能执行的任务将被删除。
• ThreadPoolExecutor.DiscardOldestPolicy&＃xff0c;如果执行程序尚未关闭&＃xff0c;则位于工作队列头部的任务将被删除&＃xff0c;然后重试执行程序&＃xff08;如果再次失败&＃xff0c;则重复此过程&＃xff09;。

除了上面的几个属性外&＃xff0c;ThreadPoolExecutor还有下面的几个参数。

//ctl是一个AtomicInteger类型的原子对象。ctl记录了线程池中的任务数量和线程池状态。
private final AtomicInteger ctl &＃61; new AtomicInteger(ctlOf(RUNNING, 0));
//线程池的锁
private final ReentrantLock mainLock &＃61; new ReentrantLock();
//所有工作的线程
private final HashSet workers &＃61; new HashSet();
//支持的等待condition
private final Condition termination &＃61; mainLock.newCondition();
//线程池中线程数量曾经达到过的最大值
private int largestPoolSize;
//已完成任务数量
private long completedTaskCount;
//是否允许为核心线程设置存活时间
private volatile boolean allowCoreThreadTimeOut;

排队策略

排队有三种通用策略&＃xff1a;
SynchronousQueue
它将任务直接传输给工作队列workers&＃xff0c;而不保持任务。如果不存在空闲线程&＃xff0c;则会新建一个线程来执行任务。比如&＃xff0c;在Executors.newCachedThreadPool()方法中使用的就是此策略。

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

LinkedBlockingQueue
有界队列&＃xff0c;使用此队列会导致在所有corePoolSize线程都忙时新任务在队列中等待。这样&＃xff0c;创建的线程就不会超过corePoolSize。比如&＃xff0c;在Executors.newFixedThreadPool()和Executors.newSingleThreadExecutor()方法中使用的就是此策略。

public static ExecutorService newFixedThreadPool(int nThreads) {return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue());
}
public static ExecutorService newSingleThreadExecutor() {return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue()));
}

ArrayBlockingQueue
有界队列&＃xff0c;没见到在哪里用到了这种策略。

线程池状态

源码已经告诉了我们线程池有几个状态。

private final AtomicInteger ctl &＃61; new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS &＃61; Integer.SIZE - 3;
private static final int CAPACITY &＃61; (1 <1;// runState is stored in the high-order bits
private static final int RUNNING &＃61; -1 <private static final int SHUTDOWN &＃61; 0 <private static final int STOP &＃61; 1 <private static final int TIDYING &＃61; 2 <private static final int TERMINATED &＃61; 3 <

可以看出&＃xff0c;一共有RUNNING、SHUTDOWN、STOP、TIDYING、TERMINATED五种状态。ctl对象一共32位&＃xff0c;高3位保存线程池状态信息&＃xff0c;后29位保存线程池容量信息。线程池的初始化状态是RUNNING&＃xff0c;在源码中体现为private final AtomicInteger ctl &＃61; new AtomicInteger(ctlOf(RUNNING, 0));。