阿里巴巴代码规范：
【强制】线程池不允许使用Executors去创建，而是通过ThreadPoolExecutor的方式，
这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险。
说明：Executors各个方法的弊端：
1）newFixedThreadPool和newSingleThreadExecutor:
主要问题是堆积的请求处理队列可能会耗费非常大的内存，甚至OOM。
2）newCachedThreadPool和newScheduledThreadPool:
主要问题是线程数最大数是Integer.MAX_VALUE，可能会创建数量非常多的线程，甚至OOM。
execute方法
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(); // 获取当前线程的状态 线程的运行状态+线程的数量
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);
}
execute 是提交线程时会调用的方法，他是ThreadPoolExecutor中最重要的方法。
ctl.get() 获取当前线程的状态 线程的运行状态+线程的数量,这里作者把一个int类型数据的高三位表示为当前线程的运行状态，低29位表示为线程的数量
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// runState is stored in the high-order bits
private static final int RUNNING = -1 << COUNT_BITS;
private static final int SHUTDOWN = 0 << 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;
// Packing and unpacking ctl
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; }
runStateof 获取当前线程的运行状态
workerCountof 获取线程的数量
corePoolSize 实例化线程池时所定义的核心线程的数量
当当前线程的数量小于核心线程的数量时 跳入addWorker方法
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())) //判断当前线程运行的状态是否不为runing状态，如果不是，则跳出循环
return false;
for (;;) {
int wc = workerCountOf(c);
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize)) // 判断当前线程是否大于核心线程数（或大于最大线程数量，根据core判断要和那个变量进行对比）
return false;
if (compareAndIncrementWorkerCount(c)) // 尝试对线程数量变量进行加运算，如果成功，则跳出循环
break retry;
c = ctl.get(); // Re-read ctl
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对象，把提交的线程任务set到worker属性里面
final Thread t = w.thread; // 安全检查，其实还是获取worker自己
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock(); //上锁，因为workers 是个hashset对象，是线程不安全的
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 // 判断worker是否已经在运行
throw new IllegalThreadStateException();
workers.add(w); //向workers添加值
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
t.start(); 调用worker的run方法,run方法又调用了worker里面的runworker方法
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
addWorker方法的大致流程为:
1.获取当前线程的运行状态
2.判断当前线程运行的状态是否不为runing状态，如果不是，则跳出循环
3.判断当前线程是否大于核心线程数（或大于最大线程数量，根据core判断要和那个变量进行对比）
4.尝试对线程数量变量进行加运算，如果成功，则跳出循环
5.实例化一个worker对象，把提交的线程任务set到worker属性里面
6.向workers添加worker对象
7.调用worker的run方法,run方法又调用了worker里面的runworker方法
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
while (task != null || (task = getTask()) != null) { // 获取worker里面的firstTask属性的值，如果为空，则调用getTask方法（线程能保活的关键）
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(); //调用firstTask里面的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);
}
}
runWorker流程大致为：
- 获取worker里面的firstTask属性的值，如果为空，则调用getTask方法,看队列内是否有线程。（线程能保活的关键,如果线程没有超时，则一直处于自旋状态，不跳出循环）
2.调用firstTask里面的run方法
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// Are workers subject to culling?
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();// 线程允不允许超时，如果允许，则从队列中取出线程，如果没有 则等待一段时间（keepAliveTime）
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
