江苏龙网

  1. 首页 > 人文 > >

车驰夜幕|ThreadPoolExecutor线程池实现原理+源码解析( 四 )

Worker类实现了Runnable接口 ， 所以本身就是一个可执行的任务 ， 并且在构造方法中将自己传递给ThreadFactory创建的线程去执行Worker类继承了AbstractQueuedSynchronizer类 ， 所以它本身也是一把锁 ， 执行任务的时候锁住自己 ， 任务执行完成后解锁 。 了解了Worker类 ， 再来看核心方法 。
executeexecute方法用于在将来的某个时间执行指定的任务 ， execute方法源码比较复杂 ， 应该先理清楚整体逻辑 ， 在逐步深入细节 。
public void execute(Runnable command) {if (command == null)// 提交空任务 ， 直接抛异常throw new NullPointerException();int c = ctl.get();if (workerCountOf(c) < corePoolSize) {// worker数量小于核心线程数 ， 创建核心线程执行任务（第二个参数为true ， 表示创建核心线程）// addWorker方法会检查线程池的状态if (addWorker(command, true))return;c = ctl.get();}if (isRunning(c)if (! isRunning(recheck)else if (workerCountOf(recheck) == 0)// 因为构造方法中corePoolSize可能为0或者核心线程也都被回收了 ， 所以此处需要判断addWorker(null, false);}else if (!addWorker(command, false))// 线程池不是RUNNING状态 ， 或者任务加入缓冲队列失败 ， 创建非核心线程执行任务（第二个参数为false）// 任务执行失败 ， 需要执行拒绝策略reject(command);}整体逻辑就是前文所示的流程图 。 相信有了流程图的对比 ， execute方法的理解就容易多了 。
addWorkeraddWorker方法用于往线程池添加新的worker 。 其实现如下：
private boolean addWorker(Runnable firstTask, boolean core) {retry: // 这种写法叫做label语法 ， 一般用于多重性循环中跳转到指定位置for (;;) {// 外层自旋int c = ctl.get();int rs = runStateOf(c);// Check if queue empty only if necessary.// 线程池状态 >= SHUTDOWNif (rs >= SHUTDOWNfor (;;) {// 内层自旋int wc = workerCountOf(c);if (wc >= CAPACITY ||wc >= (core ? corePoolSize : maximumPoolSize))// 工作中的线程数大于线程池的容量 ， 或者已经大于等于核心线程数 ， 或者大于等于最大线程数// core为true ， 表示要创建核心线程 ， false表示要创建非核心线程// 为什么大于等核心线程数的时候要返回false ， 因为要添加到缓冲队列 ， 或者创建非核心线程来执行 ， 不能创建核心线程了return false;if (compareAndIncrementWorkerCount(c))// 以CAS的方式尝试把线程数加1// 注意这里只是把线程池中的线程数加1 ， 并没有在线程池中真正的创建线程// 成功后跳出内层自旋break retry;// CAS失败 ， 再次获取ctl ， 检查线程池状态c = ctl.get();// Re-read ctlif (runStateOf(c) != rs)// 线程池状态被改变了 ， 从外层自旋开始再次执行之前的逻辑continue retry;// else CAS failed due to workerCount change; retry inner loop}}// 可以看到两层自旋 + CAS ， 仅仅是为了把线程池中的线程数加1 ， 还没有新建线程boolean workerStarted = false;boolean workerAdded = false;Worker w = null;try {// 把task包装成Workerw = 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// 添加到wordersworkers.add(w);int s = workers.size();if (s > largestPoolSize)// 维护largestPoolSize变量largestPoolSize = s;workerAdded = true;}} finally {// 解锁mainLock.unlock();}if (workerAdded) {// 添加成功t.start();workerStarted = true;}}} finally {if (! workerStarted)// 执行worker的线程启动失败addWorkerFailed(w);}return workerStarted;}


推荐阅读


上一篇：份子钱|同学结婚我转了800她没收，给我发了一张截图，我把她删除了好友

下一篇：前沿追踪|面见各国领导人最多的互联网大佬马云，为何马化腾就不行？