Java 线程池执行原理分析

澳门新葡亰平台官网 25

上一篇已经对线程池的创建进行了分析,了解线程池既有预设的模板,也提供多种参数支撑灵活的定制。

前言:
线程是稀缺资源,如果被无限制的创建,不仅会消耗系统资源,还会降低系统的稳定性,合理的使用线程池对线程进行统一分配、调优和监控,有以下好处:
1、降低资源消耗;
2、提高响应速度;
3、提高线程的可管理性。

本文将会围绕线程池的生命周期,分析线程池执行任务的过程。

线程池用一个32位的int来同时保存runState和workerCount,其中高3位是runState,其余29位是workerCount。代码中会反复使用runStateOf和workerCountOf来获取runState和workerCount。

线程池状态

首先认识两个贯穿线程池代码的参数:

  • runState:线程池运行状态
  • workerCount:工作线程的数量

线程池用一个32位的int来同时保存runState和workerCount,其中高3位是runState,其余29位是workerCount。代码中会反复使用runStateOf和workerCountOf来获取runState和workerCount。

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;

// 线程池状态
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;

// 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; }
  • RUNNING:可接收新任务,可执行等待队列里的任务
  • SHUTDOWN:不可接收新任务,可执行等待队列里的任务
  • STOP:不可接收新任务,不可执行等待队列里的任务,并且尝试终止所有在运行任务
  • TIDYING:所有任务已经终止,执行terminated()
  • TERMINATED:terminated()执行完成

线程池状态默认从RUNNING开始流转,到状态TERMINATED结束,中间不需要经过每一种状态,但不能让状态回退。下面是状态变化可能的路径和变化条件:

澳门新葡亰平台官网 1

图1 线程池状态变化路径

澳门新葡亰平台官网 2

Worker的创建

线程池是由Worker类负责执行任务,Worker继承了AbstractQueuedSynchronizer,引出了Java并发框架的核心AQS。

AbstractQueuedSynchronizer,简称AQS,是Java并发包里一系列同步工具的基础实现,原理是根据状态位来控制线程的入队阻塞、出队唤醒来处理同步。

AQS不会在这里展开讨论,只需要知道Worker包装了Thread,由它去执行任务。

调用execute将会根据线程池的情况创建Worker,可以归纳出下图四种情况:

澳门新葡亰平台官网 3

图2 worker在线程池里的四种可能

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();

    int c = ctl.get();
    //1
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    //2
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            //3
            reject(command);
        else if (workerCountOf(recheck) == 0)
            //4
            addWorker(null, false);
    }
    //5
    else if (!addWorker(command, false))
        //6
        reject(command);
}

标记1对应第一种情况,要留意addWorker传入了core,core=true为corePoolSize,core=false为maximumPoolSize,新增时需要检查workerCount是否超过允许的最大值。

标记2对应第二种情况,检查线程池是否在运行,并且将任务加入等待队列。标记3再检查一次线程池状态,如果线程池忽然处于非运行状态,那就将等待队列刚加的任务删掉,再交给RejectedExecutionHandler处理。标记4发现没有worker,就先补充一个空任务的worker。

标记5对应第三种情况,等待队列不能再添加任务了,调用addWorker添加一个去处理。

标记6对应第四种情况,addWorker的core传入false,返回调用失败,代表workerCount已经超出maximumPoolSize,那就交给RejectedExecutionHandler处理。

private boolean addWorker(Runnable firstTask, boolean core) {
        //1
        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
            }
        }
        //2
        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;
    }

标记1的第一段代码,目的很简单,是为workerCount加一。至于为什么代码写了这么长,是因为线程池的状态在不断变化,并发环境下需要保证变量的同步性。外循环判断线程池状态、任务非空和队列非空,内循环使用CAS机制保证workerCount正确地递增。不了解CAS可以看认识非阻塞的同步机制CAS,后续增减workerCount都会使用CAS。

标记2的第二段代码,就比较简单。创建一个新Worker对象,将Worker添加进workers里(Set集合)。成功添加后,启动worker里的线程。在finally里判断线程是否启动成功,不成功直接调用addWorkerFailed。

private void addWorkerFailed(Worker w) {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (w != null)
                workers.remove(w);
            decrementWorkerCount();
            tryTerminate();
        } finally {
            mainLock.unlock();
        }
    }

addWorkerFailed将减少已经递增的workerCount,并且调用tryTerminate结束线程池。

2184951-16c59af5d9d66527.png

Worker的执行

Worker(Runnable firstTask) {
    setState(-1); // inhibit interrupts until runWorker
    this.firstTask = firstTask;
    this.thread = getThreadFactory().newThread(this);
}

public void run() {
    runWorker(this);
}

Worker在构造函数里采用ThreadFactory创建Thread,在run方法里调用了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 {
       //1
        while (task != null || (task = getTask()) != null) {
            w.lock();
           //2
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();
            try {
               //3
                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;
                 //4
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;       //5
    } finally {
        //6
        processWorkerExit(w, completedAbruptly);
    }
}

标记1进入循环,从getTask获取要执行的任务,直到返回null。这里达到了线程复用的效果,让线程处理多个任务。

标记2是一个比较复杂的判断,保证了线程池在STOP状态下线程是中断的,非STOP状态下线程没有被中断。如果你不了解Java的中断机制,看如何正确结束Java线程这篇。

标记3调用了run方法,真正执行了任务。执行前后提供了beforeExecute和afterExecute两个方法,由子类实现。

标记4里的completedTasks统计worker执行了多少任务,最后累加进completedTaskCount变量,可以调用相应方法返回一些统计信息。

标记5的变量completedAbruptly表示worker是否异常终止,执行到这里代表执行正常,后续的方法需要这个变量。

标记6调用processWorkerExit结束,后面会分析。

接着来看worker从等待队列获取任务的getTask方法:

private Runnable getTask() {
    boolean timedOut = false; // Did the last poll() time out?

    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        //1
        // Check if queue empty only if necessary.
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }

        int wc = workerCountOf(c);
        //2
        // 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;
        }
       //3
        try {
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

标记1检查线程池的状态,这里就体现出SHUTDOWN和STOP的区别。如果线程池是SHUTDOWN状态,还会先处理完等待队列的任务;如果是STOP状态,就不再处理等待队列里的任务了。

澳门新葡亰平台官网,标记2先看allowCoreThreadTimeOut这个变量,false时worker空闲,也不会结束;true时,如果worker空闲超过keepAliveTime,就会结束。接着是一个很复杂的判断,好难转成文字描述,自己看吧。注意一下wc>maximumPoolSize,出现这种可能是在运行中调用setMaximumPoolSize,还有wc>1,在等待队列非空时,至少保留一个worker。

标记3是从等待队列取任务的逻辑,根据timed分为等待keepAliveTime或者阻塞直到有任务。

最后来看结束worker需要执行的操作:

private void processWorkerExit(Worker w, boolean completedAbruptly) {
   //1
    if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
        decrementWorkerCount();

  //2
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        completedTaskCount += w.completedTasks;
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }

   //3
    tryTerminate();

    int c = ctl.get();
    //4
    if (runStateLessThan(c, STOP)) {
        if (!completedAbruptly) {
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        addWorker(null, false);
    }
}

正常情况下,在getTask里就会将workerCount减一。标记1处用变量completedAbruptly判断worker是否异常退出,如果是,需要补充对workerCount的减一。

标记2将worker处理任务的数量累加到总数,并且在集合workers中去除。

标记3尝试终止线程池,后续会研究。

标记4处理线程池还是RUNNING或SHUTDOWN状态时,如果worker是异常结束,那么会直接addWorker。如果allowCoreThreadTimeOut=true,并且等待队列有任务,至少保留一个worker;如果allowCoreThreadTimeOut=false,workerCount不少于corePoolSize。

总结一下worker:线程池启动后,worker在池内创建,包装了提交的Runnable任务并执行,执行完就等待下一个任务,不再需要时就结束。

1、Executors.newFixedThreadPool(10)初始化一个包含10个线程的线程池executor;
2、通过executor.execute方法提交20个任务,每个任务打印当前的线程名;
3、负责执行任务的线程的生命周期都由Executor框架进行管理;

线程池的关闭

线程池的关闭不是一关了事,worker在池里处于不同状态,必须安排好worker的”后事”,才能真正释放线程池。ThreadPoolExecutor提供两种方法关闭线程池:

  • shutdown:不能再提交任务,已经提交的任务可继续运行;
  • shutdownNow:不能再提交任务,已经提交但未执行的任务不能运行,在运行的任务可继续运行,但会被中断,返回已经提交但未执行的任务。

public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();   //1 安全策略机制
        advanceRunState(SHUTDOWN);   //2
        interruptIdleWorkers();   //3
        onShutdown(); //4 空方法,子类实现
    } finally {
        mainLock.unlock();
    }
    tryTerminate();   //5
}

shutdown将线程池切换到SHUTDOWN状态,并调用interruptIdleWorkers请求中断所有空闲的worker,最后调用tryTerminate尝试结束线程池。

public List<Runnable> shutdownNow() {
    List<Runnable> tasks;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        advanceRunState(STOP);
        interruptWorkers();
        tasks = drainQueue();  //1
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
    return tasks;
}

shutdownNow和shutdown类似,将线程池切换为STOP状态,中断目标是所有worker。drainQueue会将等待队列里未执行的任务返回。

interruptIdleWorkers和interruptWorkers实现原理都是遍历workers集合,中断条件符合的worker。

上面的代码多次出现调用tryTerminate,这是一个尝试将线程池切换到TERMINATED状态的方法。

final void tryTerminate() {
    for (;;) {
        int c = ctl.get();
        //1
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
        //2
        if (workerCountOf(c) != 0) { // Eligible to terminate
            interruptIdleWorkers(ONLY_ONE);
            return;
        }
       //3
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                    terminated();
                } finally {
                    ctl.set(ctlOf(TERMINATED, 0));
                    termination.signalAll();
                }
                return;
            }
        } finally {
            mainLock.unlock();
        }
        // else retry on failed CAS
    }
}

标记1检查线程池状态,下面几种情况,后续操作都没有必要,直接return。

  • RUNNING(还在运行,不能停)
  • TIDYING或TERMINATED(已经没有在运行的worker)
  • SHUTDOWN并且等待队列非空(执行完才能停)

标记2在worker非空的情况下又调用了interruptIdleWorkers,你可能疑惑在shutdown时已经调用过了,为什么又调用,而且每次只中断一个空闲worker?你需要知道,shutdown时worker可能在执行中,执行完阻塞在队列的take,不知道要结束,所有要补充调用interruptIdleWorkers。每次只中断一个是因为processWorkerExit时,还会执行tryTerminate,自动中断下一个空闲的worker。

标记3是最终的状态切换。线程池会先进入TIDYING状态,再进入TERMINATED状态,中间提供了terminated这个空方法供子类实现。

调用关闭线程池方法后,需要等待线程池切换到TERMINATED状态。awaitTermination检查限定时间内线程池是否进入TERMINATED状态,代码如下:

public boolean awaitTermination(long timeout, TimeUnit unit)
    throws InterruptedException {
    long nanos = unit.toNanos(timeout);
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (;;) {
            if (runStateAtLeast(ctl.get(), TERMINATED))
                return true;
            if (nanos <= 0)
                return false;
            nanos = termination.awaitNanos(nanos);
        }
    } finally {
        mainLock.unlock();
    }
}

ThreadPoolExecutor
Executors是java线程池的工厂类,通过它可以快速初始化一个符合业务需求的线程池,如Executors.newFixedThreadPool方法可以生成一个拥有固定线程数的线程池。

后言

以上过了一遍线程池主要的逻辑,总体来看线程池的设计是很清晰的。如有错误或不足,欢迎指出,也欢迎留言交流。今次介绍了线程池运行的生命周期,下篇会研究更细粒度地控制任务的生命周期,也就是submit和Future。

澳门新葡亰平台官网 4

2184951-35585a83d35ee516.png

其本质是通过不同的参数初始化一个ThreadPoolExecutor对象,具体参数描述如下:
corePoolSize:
线程池中的核心线程数,当提交一个任务时,线程池创建一个新线程执行任务,直到当前线程数等于corePoolSize;如果当前线程数为corePoolSize,继续提交的任务被保存到阻塞队列中,等待被执行;如果执行了线程池的prestartAllCoreThreads()方法,线程池会提前创建并启动所有核心线程。

maximumPoolSize:
线程池中允许的最大线程数。如果当前阻塞队列满了,且继续提交任务,则创建新的线程执行任务,前提是当前线程数小于maximumPoolSize;

keepAliveTime:
线程空闲时的存活时间,即当线程没有任务执行时,继续存活的时间;默认情况下,该参数只在线程数大于corePoolSize时才有用;

unit:
keepAliveTime的单位;

workQueue:
用来保存等待被执行的任务的阻塞队列,且任务必须实现Runable接口,在JDK中提供了如下阻塞队列:
1、ArrayBlockingQueue:基于数组结构的有界阻塞队列,按FIFO排序任务;
2、LinkedBlockingQuene:基于链表结构的阻塞队列,按FIFO排序任务,吞吐量通常要高于ArrayBlockingQuene;
3、SynchronousQuene:一个不存储元素的阻塞队列,每个插入操作必须等到另一个线程调用移除操作,否则插入操作一直处于阻塞状态,吞吐量通常要高于LinkedBlockingQuene;
4、priorityBlockingQuene:具有优先级的无界阻塞队列;

threadFactory:
创建线程的工厂,通过自定义的线程工厂可以给每个新建的线程设置一个具有识别度的线程名。

澳门新葡亰平台官网 5

2184951-d2d8fd007c7f7a27.png

handler:
线程池的饱和策略,当阻塞队列满了,且没有空闲的工作线程,如果继续提交任务,必须采取一种策略处理该任务,线程池提供了4种策略:
1、AbortPolicy:直接抛出异常,默认策略;
2、CallerRunsPolicy:用调用者所在的线程来执行任务;
3、DiscardOldestPolicy:丢弃阻塞队列中靠最前的任务,并执行当前任务;
4、DiscardPolicy:直接丢弃任务;
当然也可以根据应用场景实现RejectedExecutionHandler接口,自定义饱和策略,如记录日志或持久化存储不能处理的任务。

Exectors:
Exectors工厂类提供了线程池的初始化接口,主要有如下几种:
newFixedThreadPool

澳门新葡亰平台官网 6

2184951-35585a83d35ee516.png

初始化一个指定线程数的线程池,其中corePoolSize ==
maximumPoolSize,使用LinkedBlockingQuene作为阻塞队列,不过当线程池没有可执行任务时,也不会释放线程。

newCachedThreadPool

澳门新葡亰平台官网 7

2184951-9b76630ac48f318c.png

1、初始化一个可以缓存线程的线程池,默认缓存60s,线程池的线程数可达到Integer.MAX_VALUE,即2147483647,内部使用SynchronousQueue作为阻塞队列;
2、和newFixedThreadPool创建的线程池不同,newCachedThreadPool在没有任务执行时,当线程的空闲时间超过keepAliveTime,会自动释放线程资源,当提交新任务时,如果没有空闲线程,则创建新线程执行任务,会导致一定的系统开销;

所以,使用该线程池时,一定要注意控制并发的任务数,否则创建大量的线程可能导致严重的性能问题

newSingleThreadExecutor

澳门新葡亰平台官网 8

2184951-deded05302aaf255.png

初始化的线程池中只有一个线程,如果该线程异常结束,会重新创建一个新的线程继续执行任务,唯一的线程可以保证所提交任务的顺序执行,内部使用LinkedBlockingQueue作为阻塞队列。

newScheduledThreadPool

澳门新葡亰平台官网 9

2184951-14823f17e4e4a09e.png

初始化的线程池可以在指定的时间内周期性的执行所提交的任务,在实际的业务场景中可以使用该线程池定期的同步数据。

实现原理
除了newScheduledThreadPool的内部实现特殊一点之外,其它几个线程池都是基于ThreadPoolExecutor类实现的。

线程池内部状态

澳门新葡亰平台官网 10

2184951-5a620e0f56cbb008.png

其中AtomicInteger变量ctl的功能非常强大:利用低29位表示线程池中线程数,通过高3位表示线程池的运行状态:
1、RUNNING:-1 <<
COUNT_BITS,即高3位为111,该状态的线程池会接收新任务,并处理阻塞队列中的任务;
2、SHUTDOWN: 0 <<
COUNT_BITS,即高3位为000,该状态的线程池不会接收新任务,但会处理阻塞队列中的任务;
3、STOP : 1 <<
COUNT_BITS,即高3位为001,该状态的线程不会接收新任务,也不会处理阻塞队列中的任务,而且会中断正在运行的任务;
4、TIDYING : 2 << COUNT_BITS,即高3位为010;
5、TERMINATED: 3 << COUNT_BITS,即高3位为011;

线程池状态默认从RUNNING开始流转,到状态TERMINATED结束,中间不需要经过每一种状态,但不能让状态回退。下面是状态变化可能的路径和变化条件:

澳门新葡亰平台官网 11

3294095-6a507005d933355b.png

任务提交
线程池框架提供了两种方式提交任务,根据不同的业务需求选择不同的方式。

Executor.execute()

澳门新葡亰平台官网 12

2184951-834971c24d085d31.png

通过Executor.execute()方法提交的任务,必须实现Runnable接口,该方式提交的任务不能获取返回值,因此无法判断任务是否执行成功。

ExecutorService.submit()

澳门新葡亰平台官网 13

2184951-ea9fe289ca3de89f.png

通过ExecutorService.submit()方法提交的任务,可以获取任务执行完的返回值。

任务执行
当向线程池中提交一个任务,线程池会如何处理该任务?

execute实现:
调用execute将会根据线程池的情况创建Worker,可以归纳出下图四种情况:

澳门新葡亰平台官网 14

3294095-c3b458db36a11d5e.png

标记1对应第一种情况,要留意addWorker传入了core,core=true为corePoolSize,core=false为maximumPoolSize,新增时需要检查workerCount是否超过允许的最大值。
标记2对应第二种情况,检查线程池是否在运行,并且将任务加入等待队列。标记3再检查一次线程池状态,如果线程池忽然处于非运行状态,那就将等待队列刚加的任务删掉,再交给RejectedExecutionHandler处理。标记4发现没有worker,就先补充一个空任务的worker。
标记5对应第三种情况,等待队列不能再添加任务了,调用addWorker添加一个去处理。
标记6对应第四种情况,addWorker的core传入false,返回调用失败,代表workerCount已经超出maximumPoolSize,那就交给RejectedExecutionHandler处理。

addWorker实现
从方法execute的实现可以看出:addWorker主要负责创建新的线程并执行任务,代码实现如下:

private boolean addWorker(Runnable firstTask, boolean core) {
        //1
        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
            }
        }
        //2
        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;
    }

在addWoker的标记1部分:
1、判断线程池的状态,如果线程池的状态值大于或等SHUTDOWN,则不处理提交的任务,直接返回;
2、通过参数core判断当前需要创建的线程是否为核心线程,如果core为true,且当前线程数小于corePoolSize,则跳出循环,开始创建新的线程,具体实现在标记2部分。

在标记2部分中,线程池的工作线程通过Woker类实现,在ReentrantLock锁的保证下,把Woker实例插入到HashSet后,并启动Woker中的线程,其中Worker类设计如下:
1、继承了AQS类,可以方便的实现工作线程的中止操作;
2、实现了Runnable接口,可以将自身作为一个任务在工作线程中执行;
3、当前提交的任务firstTask作为参数传入Worker的构造方法;

Worker的执行

Worker(Runnable firstTask) {
    setState(-1); // inhibit interrupts until runWorker
    this.firstTask = firstTask;
    this.thread = getThreadFactory().newThread(this);
}

public void run() {
    runWorker(this);
}

从Woker类的构造方法实现可以发现:线程工厂在创建线程thread时,将Woker实例本身this作为参数传入,当执行start方法启动线程thread时,本质是执行了Worker的runWorker方法。

runWorker实现:

澳门新葡亰平台官网 15

2184951-1e8ed00138c189ea.png

runWorker方法是线程池的核心:
1、线程启动之后,通过unlock方法释放锁,设置AQS的state为0,表示运行中断;
2、获取第一个任务firstTask,执行任务的run方法,不过在执行任务之前,会进行加锁操作,任务执行完会释放锁;
3、在执行任务的前后,可以根据业务场景自定义beforeExecute和afterExecute方法;
4、firstTask执行完成之后,通过getTask方法从阻塞队列中获取等待的任务,如果队列中没有任务,getTask方法会被阻塞并挂起,不会占用cpu资源;

getTask实现:

澳门新葡亰平台官网 16

2184951-a63a6646c456f715.png

整个getTask操作在自旋下完成:
1、workQueue.take:如果阻塞队列为空,当前线程会被挂起等待;当队列中有任务加入时,线程被唤醒,take方法返回任务,并执行;
2、workQueue.poll:如果在keepAliveTime时间内,阻塞队列还是没有任务,则返回null;

所以,线程池中实现的线程可以一直执行由用户提交的任务。

Future和Callable实现
通过ExecutorService.submit()方法提交的任务,可以获取任务执行完的返回值。

澳门新葡亰平台官网 17

2184951-cdacf86769a61288.png

在实际业务场景中,Future和Callable基本是成对出现的,Callable负责产生结果,Future负责获取结果。
1、Callable接口类似于Runnable,只是Runnable没有返回值。
2、Callable任务除了返回正常结果之外,如果发生异常,该异常也会被返回,即Future可以拿到异步执行任务各种结果;
3、Future.get方法会导致主线程阻塞,直到Callable任务执行完成;

submit实现:

澳门新葡亰平台官网 18

2184951-15281a586352e509.png

通过submit方法提交的Callable任务会被封装成了一个FutureTask对象。

FutureTask:

澳门新葡亰平台官网 19

2184951-d109631c519443cc.png

1、FutureTask在不同阶段拥有不同的状态state,初始化为NEW;
2、FutureTask类实现了Runnable接口,这样就可以通过Executor.execute方法提交FutureTask到线程池中等待被执行,最终执行的是FutureTask的run方法;

FutureTask.get实现:

澳门新葡亰平台官网 20

2184951-62fdfdd3f3a14e22.png

内部通过awaitDone方法对主线程进行阻塞,具体实现如下:

澳门新葡亰平台官网 21

2184951-aa841730137097ac.png

1、如果主线程被中断,则抛出中断异常;
2、判断FutureTask当前的state,如果大于COMPLETING,说明任务已经执行完成,则直接返回;
3、如果当前state等于COMPLETING,说明任务已经执行完,这时主线程只需通过yield方法让出cpu资源,等待state变成NORMAL;
4、通过WaitNode类封装当前线程,并通过UNSAFE添加到waiters链表;
5、最终通过LockSupport的park或parkNanos挂起线程;

FutureTask.run实现:

澳门新葡亰平台官网 22

2184951-29642db6d9c6589e.png

FutureTask.run方法是在线程池中被执行的,而非主线程
1、通过执行Callable任务的call方法;
2、如果call执行成功,则通过set方法保存结果;
3、如果call执行有异常,则通过setException保存异常;

set:

澳门新葡亰平台官网 23

2184951-19a5160726da146d.png

setException:

澳门新葡亰平台官网 24

2184951-34a8bba028ba58be.png

set和setException方法中,都会通过UnSAFE修改FutureTask的状态,并执行finishCompletion方法通知主线程任务已经执行完成;

finishCompletion:

澳门新葡亰平台官网 25

2184951-c380e58cdcb02e31.png

1、执行FutureTask类的get方法时,会把主线程封装成WaitNode节点并保存在waiters链表中;
2、FutureTask任务执行完成后,通过UNSAFE设置waiters的值,并通过LockSupport类unpark方法唤醒主线程;

最后来看结束worker需要执行的操作:

private void processWorkerExit(Worker w, boolean completedAbruptly) {
   //1
    if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
        decrementWorkerCount();

  //2
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        completedTaskCount += w.completedTasks;
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }

   //3
    tryTerminate();

    int c = ctl.get();
    //4
    if (runStateLessThan(c, STOP)) {
        if (!completedAbruptly) {
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        addWorker(null, false);
    }
}

正常情况下,在getTask里就会将workerCount减一。标记1处用变量completedAbruptly判断worker是否异常退出,如果是,需要补充对workerCount的减一。
标记2将worker处理任务的数量累加到总数,并且在集合workers中去除。
标记3尝试终止线程池,后续会研究。
标记4处理线程池还是RUNNING或SHUTDOWN状态时,如果worker是异常结束,那么会直接addWorker。如果allowCoreThreadTimeOut=true,并且等待队列有任务,至少保留一个worker;如果allowCoreThreadTimeOut=false,workerCount不少于corePoolSize。

线程池的关闭
线程池的关闭不是一关了事,worker在池里处于不同状态,必须安排好worker的”后事”,才能真正释放线程池。ThreadPoolExecutor提供两种方法关闭线程池:

shutdown:不能再提交任务,已经提交的任务可继续运行;
shutdownNow:不能再提交任务,已经提交但未执行的任务不能运行,在运行的任务可继续运行,但会被中断,返回已经提交但未执行的任务。

public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();   //1 安全策略机制
        advanceRunState(SHUTDOWN);   //2
        interruptIdleWorkers();   //3
        onShutdown(); //4 空方法,子类实现
    } finally {
        mainLock.unlock();
    }
    tryTerminate();   //5
}

shutdown将线程池切换到SHUTDOWN状态,并调用interruptIdleWorkers请求中断所有空闲的worker,最后调用tryTerminate尝试结束线程池。

public List<Runnable> shutdownNow() {
    List<Runnable> tasks;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        advanceRunState(STOP);
        interruptWorkers();
        tasks = drainQueue();  //1
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
    return tasks;
}

shutdownNow和shutdown类似,将线程池切换为STOP状态,中断目标是所有worker。drainQueue会将等待队列里未执行的任务返回。

interruptIdleWorkers和interruptWorkers实现原理都是遍历workers集合,中断条件符合的worker。

上面的代码多次出现调用tryTerminate,这是一个尝试将线程池切换到TERMINATED状态的方法。

final void tryTerminate() {
    for (;;) {
        int c = ctl.get();
        //1
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
        //2
        if (workerCountOf(c) != 0) { // Eligible to terminate
            interruptIdleWorkers(ONLY_ONE);
            return;
        }
       //3
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                    terminated();
                } finally {
                    ctl.set(ctlOf(TERMINATED, 0));
                    termination.signalAll();
                }
                return;
            }
        } finally {
            mainLock.unlock();
        }
        // else retry on failed CAS
    }
}

标记1检查线程池状态,下面几种情况,后续操作都没有必要,直接return。

RUNNING(还在运行,不能停)
TIDYING或TERMINATED(已经没有在运行的worker)
SHUTDOWN并且等待队列非空(执行完才能停)
标记2在worker非空的情况下又调用了interruptIdleWorkers,你可能疑惑在shutdown时已经调用过了,为什么又调用,而且每次只中断一个空闲worker?你需要知道,shutdown时worker可能在执行中,执行完阻塞在队列的take,不知道要结束,所有要补充调用interruptIdleWorkers。每次只中断一个是因为processWorkerExit时,还会执行tryTerminate,自动中断下一个空闲的worker。

标记3是最终的状态切换。线程池会先进入TIDYING状态,再进入TERMINATED状态,中间提供了terminated这个空方法供子类实现。

调用关闭线程池方法后,需要等待线程池切换到TERMINATED状态。awaitTermination检查限定时间内线程池是否进入TERMINATED状态,代码如下:

public boolean awaitTermination(long timeout, TimeUnit unit)
    throws InterruptedException {
    long nanos = unit.toNanos(timeout);
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (;;) {
            if (runStateAtLeast(ctl.get(), TERMINATED))
                return true;
            if (nanos <= 0)
                return false;
            nanos = termination.awaitNanos(nanos);
        }
    } finally {
        mainLock.unlock();
    }
}

参考文章:
http://www.jianshu.com/p/87bff5cc8d8c
http://www.jianshu.com/p/f62a3f452869

You can leave a response, or trackback from your own site.

Leave a Reply

网站地图xml地图