线程池框架.png
线程池源码解析
核心的构造方法
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;
}
核心参数说明
int corePoolSize, 池中保持的最大线程数，包括空线程
int maximumPoolSize, 池中允许最大线程数
long keepAliveTime, 当线程大于核心线程时，此为终止前多余的线程等待新任务的最长时间
TimeUnit unit, keepAliveTime 参数的时长单位
BlockingQueue<Runnable> workQueue, 执行用于保持任务的队列。此队列仅由execute方法提交的Runnable任务
ThreadFactory threadFactory, 执行程序创建线程时使用的工厂
RejectedExecutionHandler handler 由于超出线程范围和队列容量而执行的拒绝策略。
runnableTaskQueue（任务队列）：用于保存等待执行的任务的阻塞队列。可以选择以下几
个阻塞队列。
ArrayBlockingQueue:是一个基于数组结构的有界阻塞队列，此队列按FIFO（先进先出）原则对元素进行排序。
LinkedBlockingQueue:一个基于链表结构的阻塞队列，此队列按FIFO排序元素，吞吐量通常要高于ArrayBlockingQueue。静态工厂方法Executors.newFixedThreadPool()使用了这个队列。
SynchronousQueue:一个不存储元素的阻塞队列。每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于Linked-BlockingQueue，静态工厂方法Executors.newCachedThreadPool使用了这个队列。
PriorityBlockingQueue:一个具有优先级的无限阻塞队列。
RejectedExecutionHandler（饱和策略）：当队列和线程池都满了，说明线程池处于饱和状态，那么必须采取一种策略处理提交的新任务。这个策略默认情况下是AbortPolic
线程池运行核心代码
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
/*
* Proceed in 3 steps:
*
* 1. If fewer than corePoolSize threads are running, try to
* start a new thread with the given command as its first
* task. The call to addWorker atomically checks runState and
* workerCount, and so prevents false alarms that would add
* threads when it shouldn't, by returning false.
*
* 2. If a task can be successfully queued, then we still need
* to double-check whether we should have added a thread
* (because existing ones died since last checking) or that
* the pool shut down since entry into this method. So we
* recheck state and if necessary roll back the enqueuing if
* stopped, or start a new thread if there are none.
*
* 3. If we cannot queue task, then we try to add a new
* thread. If it fails, we know we are shut down or saturated
* and so reject the task.
*/
int c = ctl.get();
//判断是否小于核心数量，是直接新增work成功后直接退出
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
// 增加失败后继续获取标记
c = ctl.get();
}
//判断是运行状态并且扔到workQueue里成功后
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
//再次check判断运行状态如果是非运行状态就移除出去&reject掉
if (! isRunning(recheck) && remove(command))
reject(command);
//否则发现可能运行线程数是0那么增加一个null的worker。
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
//直接增加worker如果不成功直接reject
else if (!addWorker(command, false))
reject(command);
}
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;// 两种情况1.如果非运行状态 2.不是这种情况（停止状态并且是null对象并且workQueue不等于null）
for (;;) {
int wc = workerCountOf(c);
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false; // 判断是否饱和容量了
if (compareAndIncrementWorkerCount(c)) //增加一个work数量 然后跳出去
break retry;
c = ctl.get(); // 增加work失败后继续递归
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
w = new Worker(firstTask);//增加一个worker
final Thread t = w.thread;
if (t != null) {//判断是否 为null
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
//锁定后并重新检查下 是否存在线程工厂的失败或者锁定前的关闭
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {{ //本次要是新增加work成功就调用start运行
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();//1.取到当前线程
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
while (task != null || (task = getTask()) != null) {//获取任务 看看是否能拿到
w.lock();
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();// 确保线程是能中断的
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);
}
}
/**
* Performs cleanup and bookkeeping for a dying worker. Called
* only from worker threads. Unless completedAbruptly is set,
* assumes that workerCount has already been adjusted to account
* for exit. This method removes thread from worker set, and
* possibly terminates the pool or replaces the worker if either
* it exited due to user task exception or if fewer than
* corePoolSize workers are running or queue is non-empty but
* there are no workers.
*
* @param w the worker
* @param completedAbruptly if the worker died due to user exception
*/
private void processWorkerExit(Worker w, boolean completedAbruptly) {
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
workers.remove(w); //移除work
} finally {
mainLock.unlock();
}
tryTerminate();
int c = ctl.get();
if (runStateLessThan(c, STOP)) { //判断是否还有任务
if (!completedAbruptly) {
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return; // replacement not needed
}
addWorker(null, false);
}
}
1、如果当前池大小 poolSize 小于 corePoolSize ，则创建新线程执行任务
2、如果当前池大小poolSize 大于cprePoolSize 且等待队列未满，则进入等待队列
3、如果当前池大小 poolSize 大于 corePoolSize 且小于 maximumPoolSize ，且等待队列已满，则创建新线程执行任务
4、如果当前池大小 poolSize 大于 corePoolSize 且大于 maximumPoolSize ，且等待队列已满，则调用拒绝策
略来处理该任务
拒绝策略
AbortPolicy：抛出异常， 默认
CallerRunsPolicy：不使用线程池执行
DiscardPolicy：直接丢弃任务
DiscardOldestPolicy：丢弃队列中最旧的任务对 于 线 程 池 选 择 的 拒 绝 策 略 可 以 通 过 RejectedExecutionHandler handler = new
ThreadPoolExecutor.CallerRunsPolicy();来设置。
线程池执行流程.png
调度线程池原理
public static void main(String[] args) {
ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(10);
scheduledExecutorService.schedule(new Runnable() {
public void run() {
System.out.println("5秒之后执行");
}
}, 5, TimeUnit.SECONDS);
}
public ScheduledFuture<?> schedule(Runnable command,
long delay,
TimeUnit unit) {
if (command == null || unit == null)
throw new NullPointerException();
RunnableScheduledFuture<?> t = decorateTask(command,
new ScheduledFutureTask<Void>(command, null,
triggerTime(delay, unit)));
delayedExecute(t);
return t;
}
/**
* Main execution method for delayed or periodic tasks. If pool
* is shut down, rejects the task. Otherwise adds task to queue
* and starts a thread, if necessary, to run it. (We cannot
* prestart the thread to run the task because the task (probably)
* shouldn't be run yet.) If the pool is shut down while the task
* is being added, cancel and remove it if required by state and
* run-after-shutdown parameters.
*
* @param task the task
*/
private void delayedExecute(RunnableScheduledFuture<?> task) {
if (isShutdown())
reject(task);
else {
super.getQueue().add(task); //获取延迟队列（DelayedWorkQueue）添加任务，通过DelayedWorkQueue 延迟队列实现 offer获取对象的延迟
if (isShutdown() &&
!canRunInCurrentRunState(task.isPeriodic()) &&
remove(task))
task.cancel(false);
else
ensurePrestart();
}
}
/**
确保有work执行
* Same as prestartCoreThread except arranges that at least one
* thread is started even if corePoolSize is 0.
*/
void ensurePrestart() {
int wc = workerCountOf(ctl.get());
if (wc < corePoolSize)
addWorker(null, true);
else if (wc == 0)
addWorker(n ull, false);
}
//add 方法内部
public boolean offer(Runnable x) {
if (x == null)
throw new NullPointerException();
RunnableScheduledFuture<?> e = (RunnableScheduledFuture<?>)x;
final ReentrantLock lock = this.lock;
lock.lock(); //加锁。此处会阻塞
try {
int i = size;
if (i >= queue.length) //扩容
grow();
size = i + 1;
if (i == 0) {
queue[0] = e;
setIndex(e, 0);//第一个直接设置索引和下标0
} else {
siftUp(i, e); //排序算法对任务进行排序
}
if (queue[0] == e) {
leader = null;
available.signal(); //唤醒所有的被挤压的wait线程
}
} finally {
lock.unlock();
}
return true;
}
/**
* Sifts element added at bottom up to its heap-ordered spot.
* Call only when holding lock.
二叉树的堆排序算法
*/
private void siftUp(int k, RunnableScheduledFuture<?> key) {
while (k > 0) {
int parent = (k - 1) >>> 1;
RunnableScheduledFuture<?> e = queue[parent];
if (key.compareTo(e) >= 0)
break;
queue[k] = e;
setIndex(e, k);
k = parent;
}
queue[k] = key;
setIndex(key, k);
}
//work运行的时候调用queue的take方法
public RunnableScheduledFuture<?> take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
for (;;) {
RunnableScheduledFuture<?> first = queue[0]; //获取第一个对象
if (first == null)
available.await();
else {
long delay = first.getDelay(NANOSECONDS); //延迟时间
if (delay <= 0)
return finishPoll(first);
first = null; // don't retain ref while waiting
if (leader != null)
available.await(); ///因为没有执行线程初始化，所以等等什么时候有了自己被他人唤醒
else {
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
available.awaitNanos(delay);
} finally {
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
if (leader == null && queue[0] != null)
available.signal();
lock.unlock();
}
}
/**
* Performs common bookkeeping for poll and take: Replaces
* first element with last and sifts it down. Call only when
* holding lock.
* @param f the task to remove and return
*/
private RunnableScheduledFuture<?> finishPoll(RunnableScheduledFuture<?> f) {
int s = --size;
RunnableScheduledFuture<?> x = queue[s]; //重排序队列
queue[s] = null;
if (s != 0)
siftDown(0, x);
setIndex(f, -1);
return f;
}
scheduleAtFixedRate 与 scheduleWithFixedDelay 的区别
scheduleAtFixedRate ，是以上一个任务开始的时间计时，period时间过去后，检测上一个任务是否执行完毕，如果上一个任务执行完毕，则当前任务立即执行，如果上一个任务没有执行完毕，则需要等上一个任务执行完毕后立即执行。
scheduleWithFixedDelay，是以上一个任务结束时开始计时，period时间过去后，立即执行。
合理配置线程池
要想合理地配置线程池，就必须首先分析任务特性，可以从以下几个角度来分析。
任务的性质：CPU密集型任务、IO密集型任务和混合型任务
任务的优先级：高中低
任务的执行时间：长中短
任务 的依赖性;是否依赖其他系统资源
CPU密集型任务应配置尽可能小的线程，如配置NCPU+1个线程的线程池，
IO密集型任务线程并不是一直在执行任务 ，则应配置尽可能多的线程，如2*NCPU 。
可以通过Runtime.getRuntime().availableProcessors()方法获得当前设备的CPU个数
优先级不同的任务可以使用优先级队列PriorityBlockingQueue来处理，它可以让优先级高的任务先执行。
