在没有看线程池的源码的时候，我脑子里经常会想，线程跑完怎么之后怎么归还给线程池。其实这种思想很明显是和对象池的概念搞混淆了。
首先讲对象池。对象池的核心思想是缓存和共享。那些被频繁使用的对象，在使用完后，不立即将它们释放，而是将它们缓存起来，以供后续的应用程序重复使用，从而减少创建对象和释放对象的次数，进而改善应用程序的性能。但是这个共享并不是同时共享。而是一个线程用完之后，交给另外一个线程用。就像数据库连接池。就像一把钥匙，多个人共用，但是一次只能一个人进入。我们把这种叫做串行线程关闭。
串行线程封闭使得任意时刻，最多仅有一个线程拥有对象的所有权。当然，这不是绝对的，只要线程T1事实不会再修改对象O，那么就相当于仅有T2拥有对象的所有权。串行线层封闭让对象变得可以共享（虽然只能串行的拥有所有权），灵活性得到大大提高；相对的，要共享对象就涉及安全发布的问题，依靠BlockingQueue等同步工具很容易实现这一点。
见前面我写的自定义数据库连接池https://www.jianshu.com/p/545f8b1194ef
对象的复用是对象的所有权的交接。
而线程池的话是生产者消费者模型.
小于core个数的时候，去创建worker线程。然后调用runworker方法中的,task.run();假设core=3 ;然后不断创建3个的消费者线程。即使先创建消费者线程1，和2 。然后这个时候消费者1执行完成了。假如这个时候再有任务。还是创建消费者线程3 。后面再有任务的话，堆积在BlockingQueue上。用户不断的submit提交任务。生产者。然后 3个消费者去并发消费提交的任务。他并没有归还给线程池。task.run();这句执行完成之后，会进入下一个循环，执行task = getTask()方法。
讲完了这两个东西的区别之后。我们来看线程池的源码。
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ThreadPoolTest {
private static final ExecutorService service = Executors.newFixedThreadPool(5);
public static void main(String[] args) {
service.submit(new Task());
}
}
class Task implements Runnable{
@Override
public void run() {
System.out.println("hello world");
}
}
注意，《阿里开发手册》 上不建议用这种方式创建线程池。ExecutorService service = Executors.newFixedThreadPool(5); 因为这样创建的LinkedBlockingQueue的长度是Integer.MaxValue。会耗尽系统资源。
Runnable任务的走向。
分析源码之前，我们先画出线程池的UML类图。
image.png
主要的方法逻辑都在ThreadPoolExecutor上。
下面我们分析runnable task怎么走。
1 提交任务
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
1.首先不能提交空任务，否者抛个异常
2.然后通过newTaskFor方法将一个Runnable类型的task包装成一个RunnableFuture类型的ftask
3.然后调用executor方法执行ftask
newTaskFor(task, null)这一句的目的是将Runnable接口适配成一个返回值为null的FutureTask类型的任务。当然这都不重要，我们关心的是task到哪里去了
2 封装任务成FutureTask
包装成一个FutureTask
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
}
2.包装成一个Callable交给成员变量callable
public FutureTask(Runnable runnable, V result) {
this.callable = Executors.callable(runnable, result);
this.state = NEW; // ensure visibility of callable
}
3.包装成一个RunnableAdapter
public static <T> Callable<T> callable(Runnable task, T result) {
if (task == null)
throw new NullPointerException();
return new RunnableAdapter<T>(task, result);
}
4.交给了成员变量task,并交由call方法来执行
static final class RunnableAdapter<T> implements Callable<T> {
final Runnable task;
final T result;
RunnableAdapter(Runnable task, T result) {
this.task = task;
this.result = result;
}
public T call() {
task.run();
return result;
}
}
task先封装成 RunnableAdapter 的task属性。然后这个RunnableAdapter是FutureTask的callable属性。
然后接下来 我们关心ftask的callable属性。
3 执行exuecute方法。
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);
}
addWorker方法
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()))
return false;
for (;;) {
int wc = workerCountOf(c);
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
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);
final Thread t = w.thread;
if (t != 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) {
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
w = new Worker(firstTask); 先封装成worker对象。然后执行t.start()。
执行worker对象 的thread属性的 run()方法。
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
这段代码把线程的执行机制，和任务的提交分离. this指代当前worker对象。也就是把当前worker对象（runnable对象）给thread执行。但是这里有个问题，传入的firstTask干嘛去了。
addWorker方法中执行的t.start() ，启动worker方法中的run()方法。
调用worker对象的runwork方法。
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) {
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);
}
}
这里前面的疑问解决了，先拿到前面传入的firstTask，作为task变量。第一次task不为空，短路或后面语句不执行。然后task.run(); 最后task置为空。while循环下次判断。while (task != null || (task = getTask()) != null) .执行task =getTask()方法。
执行下一个任务
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();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
workQueue.take() 从阻塞队列中获取，任务Runnable。作为getTask()方法返回值。继续执行上面的runWorker方法中的task.run()方法
参考资料：
从一个run方法的经历看线程池https://blog.csdn.net/u011803809/article/details/77427641
网友评论