概述
ThreadPoolExecutor是ExecutorService的间接实现。ThreadPoolExecutor是继承AbstractExecutorService。而AbstractExecutorService实现了ExecutorService接口。
public class ThreadPoolExecutor extends AbstractExecutorService
public abstract class AbstractExecutorService implements ExecutorService
构造函数
public ThreadPoolExecutor(int corePoolSize, //核心线程容量
int maximumPoolSize, //线程池所允许的最大容量
long keepAliveTime,//存活时间
TimeUnit unit,//时间单位
BlockingQueue<Runnable> workQueue//任务队列
ThreadFactory threadFactory) { //线程工厂
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}
在构造函数中，里面有一个线程工厂ThreadFactory。我们这里先看下线程工厂
ThreadFactory介绍
public interface ThreadFactory {
/**
* Constructs a new {@code Thread}. Implementations may also initialize
* priority, name, daemon status, {@code ThreadGroup}, etc.
*
* @param r a runnable to be executed by new thread instance
* @return constructed thread, or {@code null} if the request to
* create a thread is rejected
*/
Thread newThread(Runnable r);
}
其默认实现类是DefaultThreadFactory类，此类是在Executors类的内部类.
/**
* The default thread factory.
*/
private 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-";
}
public Thread newThread(Runnable r) {
Thread t = new Thread(group, r,
namePrefix + threadNumber.getAndIncrement(),
0);
if (t.isDaemon())
t.setDaemon(false);//所有线程都默认设置为非Daemon线程
if (t.getPriority() != Thread.NORM_PRIORITY)
t.setPriority(Thread.NORM_PRIORITY); //所有线程的优先级相同
return t;
}
}
从上面代码中可以得到以下几点信息：
- 在这个线程工厂中，同一个线程池的所有线程同属于一个线程组，也就是创建线程池的哪个线程组，同时线程池的名称都是"pool-- thread --"
其中poolNum是线程池的数量序号，threadNum是此线程池中的线程数量序号 - 另外对于线程池中的所有线程默认都转换为非后台线程，这样主线程退出时不会直接退出JVM，而是等待线程池结束。
- 还有一点就是默认将线程池中的所有线程都调为同一个级别，这样在操作系统角度来看所有系统都是公平的，不会导致竞争堆积。
executor方法
当我们使用ThreadPoolExecutor的构造方法构造出一个线程池后，我们将调用executor方法来执行任务。下面我们就来分析下ThreadPoolExecutor中executor的内部实现.
/**
* Executes the given task sometime in the future. The task
* may execute in a new thread or in an existing pooled thread.
*
* If the task cannot be submitted for execution, either because this
* executor has been shutdown or because its capacity has been reached,
* the task is handled by the current {@code RejectedExecutionHandler}.
**/
我们从注释中可以得到如下信息：
- 任务将在未来的某个时间可能被执行，意味着可能不被执行，为什么呢？
- 任务可能在一个新线程中也可能在线程池中已存在的线程中执行
- 如果任务不能被提交执行，可能时由于线程池已经shutdown或者时任务队列中已经满了。这种情况下任务将由此时的任务拒绝策略决定怎么处理 这就是可能不被执行的答案。
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)//如果工作线程数为0 则新建一个线程
addWorker(null, false);
}
else if (!addWorker(command, false)) //如果我们添加队列失败，尝试重新开启新线程
reject(command);
}
private static boolean isRunning(int c) {
return c < SHUTDOWN;
}
上面的方法中调用了addWorker来添加新的线程来处理新的任务。下面我们来看下这个方法的内部实现:
addWorker方法
根据当前线程的状态和给的边界条件来检测是否需要一个新的线程添加。
如果需要，则添加到线程队列中并调整工作线程数并启动线程执行第一个任务。
如果该方法检测到线程池处于STOP状态或者将要停止，则返回false。
如果线程工厂创建线程失败，则返回false。
private boolean addWorker(Runnable firstTask, boolean core) {
//各种检测来判断是否需要建立一个新的Worker对象
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;//退出循环 即需要建立一个新的Worker对象
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
//开始产生一个Worker对象，并将其添加到线程队列中去
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方法
addWorkerFailed(w); //从线程队列中将其移除。
}
return workerStarted;
}
private void addWorkerFailed(Worker w) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (w != null)
workers.remove(w);
decrementWorkerCount();
tryTerminate();
} finally {
mainLock.unlock();
}
}
最后我们就来看下线程类Worker
线程类Worker
Worker类主要维持线程执行任务时的中断控制状态，以及其他一些小的记录
Worker类继承了AQS同步器，用于在执行任务前后来获取和释放锁。这可防止中断，在等待一个任务时，旨在唤醒工作线程而不是中断正在运行的任务。
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;
this.thread = getThreadFactory().newThread(this);
}
/** Delegates main run loop to outer runWorker. */
public void run() {
runWorker(this);//调用runWorker方法来执行这个任务
}
// Lock methods
//
// The value 0 represents the unlocked state.
// The value 1 represents the locked state.
protected boolean isHeldExclusively() {
return getState() != 0;
}
protected boolean tryAcquire(int unused) {
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
protected boolean tryRelease(int unused) {
setExclusiveOwnerThread(null);
setState(0);
return true;
}
public void lock() { acquire(1); }
public boolean tryLock() { return tryAcquire(1); }
public void unlock() { release(1); }
public boolean isLocked() { return isHeldExclusively(); }
void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
}
当Worker类中的线程启动之后，则会调用Worker类中的run方法，而run方法调回用了runWorker方法，因此我们看下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) {
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);
}
}
上面方法抛出各种检查异常，简单来说就是：线程池正常工作的时候直接调用了我们自己提交的run方法。
由此可见，线程池的基本原理其实也很简单，无非预先启动一些线程，线程进入死循环状态，每次从任务队列中获取一个任务进行执行，直到线程池被关闭。如果某个线程因为执行某个任务发生异常而终止，那么重新创建一个新的线程而已。如此反复。
总结
线程池的工作原理：当我们写程序中使用submit往线程池中提交新任务的时候，线程池首先会判断是否需要生成新的线程来执行这个任务。如果需要，则新建一个，然后启动此线程，接着就是一个一个的任务被启动的线程执行了。
网友评论