线程池原理
线程池,关注如何缩短频繁线程创建和销毁的时间,通过线程复用技术,减少非核心任务的时间损耗(创建和销毁的时间),提高程序性能。它的主要原理是采用阻塞任务队列实现线程复用的方案。ThreadPoolExecutor是线程池的具体实现类,它的继承关系图。
ThreadPoolExecutor继承关系图.jpg线程池类型
源码提供了三种类型的线程池,通过不同的参数设置,缓存线程池、单线程线程池、固定数量线程的线程池。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;
}
这几个参数决定线程池类型,核心参数。
corePoolSize,核心线程数。
maximumPoolSize,线程池的允许的最大线程,超过会报异常。
keepAliveTime,非核心线程允许存留时间,(保持活跃性的时间)。
TimeUnit,时间度量参数。
BlockingQueue<Runnable>,任务队列,可以使配置成无限或有限队列或栈。
ThreadFactory,线程工厂,创建线程优先级以及统计线程数量等。
线程池的对任务的控制流程
- 线程数量<corePoolSize,新建线程,处理任务。
- 线程数量>=corePoolSize,将任务放入workQueue队列,若有核心线程空闲,从workQueue队列取任务处理。
- 当任务队列有限且已满,再次新建线程处理任务,这时,要保证总量不超过最大允许值,否则,导致RejectedExecutionHandler异常。
- 再次新建的线程是非核心线程,空闲时最大存留keepAliveTime时间。
三种线程池是Executors的静态方法创建。
缓存线程池
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
核心线程数是0,允许的最大线程无限,任务队列SynchronousQueue不做具体存储,线程活跃时间60秒。
每个任务先到SynchronousQueue队列,它其实一个管道,不保存,若不存在空闲线程则新建,新建线程数量永远不会超过允许的最大值。
若一开始并发任务较多,会创建不少线程,每个线程任务完成后,变空闲线程,空闲时间未达到60s,可重用空闲线程处理新进任务,线程最大数量不限。即可以保障任务繁重时,空闲线程可复用,又能办证在没有任务时,保持一定时间后消失。
单线程线程池
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
核心线程是1,允许的最大线程是1,任务队列LinkedBlockingQueue,不设空闲时间,无界队列。
只有一个核心线程处理任务,新任务入队列。若队列是有界的,多并发任务状态下队列总有满的时候,若队列满了就得新建临时线程,肯定会超允许的最大线程,报异常,因此,队列必须无界。
固定数量线程的线程池
Executors#newFixedThreadPool方法创建固定数量线程的线程池。
public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory
threadFactory) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(),
threadFactory);
}
核心线程与最大线程数量自己配置,任务队列LinkedBlockingQueue,不设空闲时间,无界队列。
任务队列无界,因此,不会创建临时工线程,只有核心线程工作,keepAlieveTime无需设置。
综上所述
允许的最大线程数量与队列必须保证一个是无界的,否则,在高并发条件下导致异常。根据特定的业务场景灵活配置。
工作原理
当线程池ThreadPoolExecutor创建后,我们一般调用它的execute方法,向线程池派发任务,将任务交给线程池,不需要自己再管理,会自动执行。
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();
}
if (isRunning(c) && workQueue.offer(command)) {//加入队列
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
else if (!addWorker(command, false))//队列满了,创建非核心临时线程
reject(command);
}
首先,workerCountOf(c)方法,返回工作线程数量,若小于核心线程数量,调用addWorker方法,创建核心线程(设标志位参数),该任务将由新建的核心线程处理。
然后,若工作线程数量>=核心线程,isRunning(c)方法,表示c<0,将任务加入队列,offer方法,加入后返回成功标志。
因为offer是非阻塞方法,也就是说,如果队列已满,将直接返回失败,这时,将调用addWorker方法,创建非核心线程(不设标志位参数)。该任务将由新建的非核心线程处理。
以下是上面代码用到的一些变量与方法。
//AtomicInteger确保高并发整型值自增时线程安全。
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;//32-3=29
private static final int CAPACITY = (1 << COUNT_BITS) - 1;//536870911
private static final int RUNNING = -1 << COUNT_BITS;//-536870912
private static final int SHUTDOWN = 0 << COUNT_BITS;//0
private static final int STOP = 1 << COUNT_BITS;//536870912
private static final int TIDYING = 2 << COUNT_BITS;//1073741824
private static final int TERMINATED = 3 << COUNT_BITS;//1610612736
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; }
private static boolean isRunning(int c) {
return c < SHUTDOWN;
}
ctl的初始值与RUNNING相同,值是-536870912,即11100000 00000000 00000000 00000000。
-1存储的补码是1111...(32个1),向左移动29位,即右边加29个0。
workerCountOf方法将c与CAPACITY进行与操作。
runStateOf方法将c与CAPACITY的取反进行与操作。
CAPACITY是536870911,即00011111 11111111 11111111 11111111。
CAPACITY的取反是11100000 00000000 00000000 00000000。
workerCountOf方法,第一次调用时,ctl初始值与CAPACITY与操作结果是0,若ctl不断自增,与CAPACITY操作的值不断自增1,工作线程初始是0,每增加一个工作线程,ctl自增一次。
runStateOf方法,第一次调用时,ctl初始值与反CAPACITY与操作结果为反CAPACITY。
总之,workerCountOf方法返回工作线程数量,runStateOf方法返回ctl的初始值-536870912。
五个运行状态,RUNNING状态<0,isRunning方法,判断c是否<0,每次新建线程后c值自增1,c初始值-536870912,一般情况下,不足以使达到c>=0,因此,isRunning(c)方法一般返回true。
下面分析一下addWorker创建线程方法。
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
...
for (;;) {
int wc = workerCountOf(c);//工作线程数量
if (wc >= CAPACITY ||// 已经等于最大值。
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get();
if (runStateOf(c) != rs)
continue retry;
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//创建线程,绑定任务。
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();//上锁
try {
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive())
throw new IllegalThreadStateException();
workers.add(w);//任务放入集合
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true; //后续根据此标志启动线程。
}
} finally {
mainLock.unlock();//解锁
}
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
两层死循环,若工作线程>CAPACITY容量,或>=允许的最大值(创建核心线程>=核心线程数量),返回失败。自增ctl,跳出循环。创建一个Worker对象,构造方法。
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
它是Runnable类型,封装派发给线程池的Runnable任务,ThreadFactory的newThread方法,创建一个新线程,将Worker类作为任务主体。该类的结构关系图。
线程池的Worker任务关系图.jpg 我们再回到addWorker方法,runStateOf(ctl.get())<0时(一般<0),将Worker任务放入HashSet集合,设置workerAdded标志。然后,根据该标志,调用Thread的start方法,启动线程,设置启动标志workerStarted,表示线程已启动执行任务。最后,返回启动标志。新线程执行Worker任务,触发Worker#run方法。该方法调用runWorker方法。它是外部类ThreadPoolExecutor的方法,入参就是该Worker。
final void runWorker(Worker w) {//处理Worker内的派发任务,在循环中进一步访问任务队列。
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock();
boolean completedAbruptly = true;
try {
while (task != null || (task = getTask()) != null) {
....
try {
beforeExecute(wt, task);
Throwable thrown = null;
try {
task.run();//执行任务
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
afterExecute(task, thrown);
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
该方法中,新建线程借助Worker任务,开始为线程池工作,内部Runnable是用户需要完成的任务。在while循环中,第一个优先处理execute派送,Worker内部的线程池任务,完成后,线程也不会结束,而是getTask方法,继续从任务队列中获取务,如果getTask返回空,结束线程,否则,继续执行,该方法可能会阻塞。
private Runnable getTask() {
boolean timedOut = false;
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
...//队列空时,返回null。
int wc = workerCountOf(c);
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
如果工作线程数量wc>核心线程,设置timed标志。从队列取任务采用阻塞等待一定时间的poll方式,等待时间设置线程存活时间keepAliveTime,因此,即使队列没有任务,线程仍然存活,(保证任务来到可立即开始工作)。
如果当工作线程数量wc<核心线程,采用一直阻塞的take方案,队列是空时线程一直阻塞,核心线程不会死亡。
如果队列是空,poll时间已到,设置timeOut超时标志,进入下次循环,这时,如果再发生工作线程数量wc>核心线程,会使得timed和timedOut标志同时存在,此时,工作线程数量自减,返回空,退出while循环,线程结束。如果工作线程数量wc<核心线程(仅有核心线程),工作线程不会自减,for循环继续,阻塞在take查询任务。
如果设置核心线程TimeOut,也会采用poll方式,存活时间一到,队列无任务,即使wc数量<核心线程,线程也会退出,允许核心线程死亡。
每一个新建的线程在该方法的执行逻辑是相同。根据当前线程数量和超时标志决定从任务队列的获取方法是否阻塞。
总结
线程池的本质,每一个线程在完成派发任务后,并未结束,继续访问任务队列。根据当前线程数量,利用任务队列的阻塞特性,实现线程的存留时间。通过留存工作线程消费,不新建线程,实现任务的线程复用。
任重而道远