深入理解线程池实现原理
2018-09-27 本文已影响10人
曾泽浩
笔者对线程池的一些理解，如果理解有偏差，欢迎指正。
代码贴着有点多，需要细心理解和阅读，有不明白的地方可以一起讨论讨论。
首先看一个最简单的线程池的使用吧
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ExecutorsTest {
public static void main(String[] args) {
ExecutorService executorService = Executors.newFixedThreadPool(2);
for (int i = 0; i < 10 ; i++) {
final int index = i;
executorService.execute(()-> {
System.out.println(Thread.currentThread().getName() + "你好啊" + index);
});
}
executorService.shutdown();
}
}
在来看看几个重要的接口和类
Executor: An object that executes submitted {@link Runnable} tasks. 用来执行提交的Runnable任务,暂且称为执行器吧。
public interface Executor {
void execute(Runnable command);
}
ExecutorService:继承了Executor，拓展了Executor的功能，比如关闭线程池等。
public interface ExecutorService extends Executor {
//关闭线程池，Don't accept new tasks, but process queued tasks
//不能接受新的执行任务，但是能继续处理已经在队列中的任务
void shutdown();
//马上关闭线程池，
//Don't accept new tasks, don't process queued tasks, and interrupt in-progress //tasks
//不能接受新的执行任务，不能继续处理已经在队列中的任务，并且会中断正在处理的任务
List<Runnable> shutdownNow();
//等待多长时间返回线程池的状态是否停止
boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException;
//提交任务，有很多方法
<T> Future<T> submit(Callable<T> task);
Future<?> submit(Runnable task);
}
AbstractExecutorService: Provides default implementations of {@link ExecutorService}
execution methods. 这个类是对ExecutorService默认的实现。这里主要关注submit()方法
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
//主要就是把Runnable包装一下，然后扔到线程池去执行 execute(ftask). 实现这个方法的正是ThreadPoolExecutor类里的execute(Runnable command)方法
Executors: 线程池的工具类，主要用来创建一个线程池服务ExecutorService
ThreadFactory: 线程工厂类，用来创建线程
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 : 该线程池中线程总数最大值
workQueue : 该线程池中的任务队列：维护着等待执行的Runnable对象
threadFactory : 线程工厂
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.
*/
//ctl可以获得线程池的状态和正在运行的线程数量
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);
}
紧跟着，看看addWoker
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
//如果线程池已经关闭，firstTask是空，工作队列是空，直接返回false
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;
//工作线程+1，退出循环
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) {
//启动线程，然后会执行Worker的run()方法（runWorker(this);）
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
/**
* This class will never be serialized, but we provide a
* serialVersionUID to suppress a javac warning.
*/
private static final long serialVersionUID = 6138294804551838833L;
/** Thread this worker is running in. Null if factory fails. */
final Thread thread;
/** Initial task to run. Possibly null. */
Runnable firstTask;
/** Per-thread task counter */
volatile long completedTasks;
/**
* Creates with given first task and thread from ThreadFactory.
* @param firstTask the first task (null if none)
*/
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
//通过线程工厂来创建线程
//Executors.defaultThreadFactory()默认的线程工厂
//新建的线程将Worker传进去，线程启动的时候将会执行下面的run()方法
this.thread = getThreadFactory().newThread(this);
}
/** Delegates main run loop to outer runWorker */
//
public void run() {
runWorker(this);
}
...
}
默认线程工厂的实现
static class DefaultThreadFactory implements ThreadFactory {
private static final AtomicInteger poolNumber = new AtomicInteger(1);
private final ThreadGroup group;
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final String namePrefix;
DefaultThreadFactory() {
SecurityManager s = System.getSecurityManager();
group = (s != null) ? s.getThreadGroup() :
Thread.currentThread().getThreadGroup();
namePrefix = "pool-" +
poolNumber.getAndIncrement() +
"-thread-";
}
//这个将Worker传进来
public Thread newThread(Runnable r) {
Thread t = new Thread(group, r,
namePrefix + threadNumber.getAndIncrement(),
0);
if (t.isDaemon())
t.setDaemon(false);
if (t.getPriority() != Thread.NORM_PRIORITY)
t.setPriority(Thread.NORM_PRIORITY);
return t;
}
}
总结一下上面的思路先，
addWorker()主要就是新建线程并且启动线程。
启动线程的时候会执行runWorker()。
最后看看runWorker()这个方法吧
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
//firstTask是新建一个线程时携带的第一个任务，后面的任务直接从队列中拿
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
//这里有一个很重要的点，在工作任务队列为空的时候，getTask()会一直阻塞
//因为queue.take()会阻塞
//但如果线程池STOP或者SHUTDOWN，getTask()会返回空，这个时候就会结束循环
//也就是线程停止执行（销毁）
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);
}
}
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.
//1.线程池状态SHUTDOWN并且任务工作队列为空，直接返回null
//这里保证了调用shutdown()方法，任务队列的工作还会继续执行
//2.线程池状态为STOP，直接返回null
//这个印证了shutdownNow()方法不会执行任务队列的工作
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// Are workers subject to culling?
//allowCoreThreadTimeOut 核心线程也设置过期时间，默认为false
//工作线程数量大于核心线程数，返回ture
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
//工作线程大于最大线程数，返回null
//timed
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
Runnable r = timed ?
//keepAliveTime 在allowCoreThreadTimeOut设为ture 或者工作线程数量大于核心线程数才生效，poll()方法有可能返回null
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
//如果为workQueue的话，将会一直阻塞
//这也是getTask()方法会一直阻塞的原因，相当于线程池里的线程一直死循环///在处理工作任务
//如果在阻塞过程中，线程被中断将会抛出InterruptedException，此时也会结//束调用getTask()的那个while循环
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
