资讯专栏INFORMATION COLUMN

ThreadPoolExecutor线程池源码分析

stormzhang / 575人阅读

摘要:线程池技术旨在解决两个不同的问题在处理大量异步任务时可以提高性能,因为减少了线程的销毁,新建,切换等消耗性能的操作。线程池还有能力统一管理,调度,监控,调优线程等,还提供了一下基本的统计,比如已完成的任务数量。线程数量,线程池的状态。

了解ThreadPoolExecutor

先看一下线程池类的类图关系:

Executor接口

Executor作者描述的是Executor提供了一种解耦方式将任务的提交和任务以何种技术执行分离;
Executor接口只有一个方法:

void execute(Runnable command);

execute方法接收一个Runnable对象,方法的描述是在未来的某个时间执行command。不管是在一个新的线程中执行,还是在线程池中执行,甚至在调用者线程中立即执行。

ExecutorService接口

ExecutorService继承了Executor接口,ExecutorService可以被关闭,关闭以后不再接收新的任务。ExecutorService提供了两个不同的方法关闭ExecutorService。shutdown方法会等待之前还未执行的任务执行完毕再关闭,而shutdownNow则不会再启动新的任务,还会中断正在执行的任务。一旦关闭后,ExecutorService就不会有正在执行的任务,也不会有等待被执行的任务,更不会有新的任务被提交。ExecutorService关闭后应该处理好一些资源的回收。

ThreadPoolExecutor

线程池技术旨在解决两个不同的问题:

在处理大量异步任务时可以提高性能,因为减少了线程的销毁,新建,切换等消耗性能的操作。

线程池还有能力统一管理,调度,监控,调优线程等,还提供了一下基本的统计,比如已完成的任务数量。

重要的状态和状态判断的方法
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//高3位和低29位分别表示状态和线程数
private static final int COUNT_BITS = Integer.SIZE - 3;
//1左移29位减一得到低29位都是1,即线程的最大数量,大概5亿多
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

// runState is stored in the high-order bits
private static final int RUNNING    = -1 << COUNT_BITS;//111
private static final int SHUTDOWN   =  0 << COUNT_BITS;//000
private static final int STOP       =  1 << COUNT_BITS;//001
private static final int TIDYING    =  2 << COUNT_BITS;//010
private static final int TERMINATED =  3 << COUNT_BITS;//011

// Packing and unpacking ctl
//获得状态
private static int runStateOf(int c)     { return c & ~CAPACITY; }
//获得线程数量
private static int workerCountOf(int c)  { return c & CAPACITY; }
//通过状态和线程数量组装ctl
private static int ctlOf(int rs, int wc) { return rs | wc; }

/*
 * Bit field accessors that don"t require unpacking ctl.
 * These depend on the bit layout and on workerCount being never negative.
 */
//c状态是否小于s状态
private static boolean runStateLessThan(int c, int s) {
    return c < s;
}
//c状态是否大于等于s状态
private static boolean runStateAtLeast(int c, int s) {
    return c >= s;
}
//线程池是否是运行状态
private static boolean isRunning(int c) {
    return c < SHUTDOWN;
}

整个类最重要的一个状态标志ctl是一个AtomicInteger,它包含了两个字段的含义。workerCount线程数量,runState线程池的状态。
这一个字段是如何包含两个字段的含义的呢,Doug Lea大牛使用了一个int的32位bits的高三位保存了状态值,低29位保存了线程数量。

其中五个状态:
RUNNING:接收新的任务,处理队列中的任务;
SHUTDOWN:不接收新的任务,但处理队列中的任务;
STOP:不接收新的任务,不处理队列中的任务,中断正在执行的任务;
TIDYING:所有任务都终止,线程数为0, 线程过度到TIDYING时会调用terminated钩子方法;
TERMINATED:terminated执行完毕;

状态之间的转换:
RUNNING -> SHUTDOWN:调用shutdown方法;
(RUNNING or SHUTDOWN) -> STOP:调用shutdownNow方法;
SHUTDOWN -> TIDYING:当线程池和任务队列都为空;
STOP -> TIDYING:当线程池为空;
TIDYING -> TERMINATED:当terminated方法执行完毕;

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);
    }

    // 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继承了AQS,实现了Runnable接口;在构造函数中,初始化了它的第一次仍无,使用threadFactory创建一个新的线程;
Worker继承AQS,目的是想使用独占锁来表示线程是否正在执行任务,Worker的线程获取了独占锁就说明它在执行任务,不能被中断。从tryAcquire方法可以看出,它实现的是不可重入锁,因为是否获得锁在这里表示一个状态,如果可以重入的话,独占锁就失去了只表示一个状态的含义。在构造函数初始化时,Worker将state设置为-1,因为在tryAcquire中CAS操作compareAndSetState(0, 1),表示state在-1时不能被中断。在runWorker中将state设置为0.

ThreadPooleExecutor构造方法
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue 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:核心线程数量,即使线程是空闲的也保持在线程池中,除非allowCoreThreadTimeOut参数被设置;

maximumPoolSize:最大线程数量;

keepAliveTime:当线程数量超过核心线程数量时,超出的空闲线程等待新任务的最大时长;

unit:时间单位;

workQueue:存放将要被执行的任务的队列;

threadFactory:创建线程的线程工厂;

handler:当任务队列满且没有空闲的线程时处理任务的handler,线程池提供了四种策略:

AbortPolicy:直接抛出异常,默认;

CallerRunsPolicy:使用调用者的线程执行;

DiscardOldestPolicy:抛弃队列最前的任务,执行当前任务;

DiscardPolicy:直接丢弃任务;

这些参数对整个线程池运行非常重要;

execute方法
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();
    }
    //线程数大于核心线程数
    //如果线程池running状态且添加任务到队列成功
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        //如果线程池不是运行状态,队列移除任务,使用拒绝策略处理任务
        if (! isRunning(recheck) && remove(command))
            reject(command);
        //如果这时线程数为0,添加任务
        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.
        //如果运行状态大于等于SHUTDOWN,不再接受新的任务,返回false
        //如果运行状态等于SHUTDOWN且firstTask不为空,继续执行下去,如果firstTask为空,queue为空,返回false,否则继续执行;只要SHUTDOWN状态下还有任务在,就需要往下执行,可能需要新建worker执行
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;

        for (;;) {
            //获得线程数量
            int wc = workerCountOf(c);
            //如果线程数量大于容量或者当core为true时wc大于等于核心线程数,当core为falsewc大于等于最大线程数量时,返回false
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            //CAS线程数加一,成功则中断循环
            if (compareAndIncrementWorkerCount(c))
                break retry;
            //如果CAS失败,重新获取ctl,线程池运行状态没变的话继续loop
            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 {
        //新建一个worker
        w = new Worker(firstTask);
        //能得到worker的thread
        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());

                //如果rs是RUNNING或者SHUTDOWN且firstTask为null
                //因为SHUTDOWN时还需要执行queue中的任务
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    //往线程池中添加worker
                    workers.add(w);
                    int s = workers.size();
                    //记录线程池出现的最大线程数量
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {
                //启动worker
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

worker的run方法调用的是runWorker;

runWorker方法
final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    //保存worker的第一个任务
    Runnable task = w.firstTask;
    //清空worker的第一个任务
    w.firstTask = null;
    //这里将worker的state设置为0,允许中断
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        //如果task为空,则从队列中获取任务
        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
            //如果当前状态大于等于STOP要保持当前线程中断
            //如果当前线程小于STOP即RUNNING或者SHUTDOWN,调用Thread.interrupted()清空中断标志,如果这时调用了shutdownNow状态为STOP,还是要保持中断状态
            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;
                //worker的完成任务数量加一,此时是线程安全的
                w.completedTasks++;
                //释放锁
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        //线程退出
        processWorkerExit(w, completedAbruptly);
    }
}

每个task在调用runWorker后会一直循环执行任务,直到queue中没有任务了,循环结束,worker生命周期结束。

getTask

上面runWorker时调用了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.
        //如果rs大于等于SHUTDOWN,当RS大于等于STOP说明线程池已经不处理队列中的任务了,当rs为SHUTDOWN时,如果队列是空的,返回null
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            //线程数减一
            decrementWorkerCount();
            return null;
        }

        int wc = workerCountOf(c);

        // Are workers subject to culling?
        //是否超时控制,allowCoreThreadTimeOut默认false,代表不允许核心线程超时,对于超出核心线程的线程需要控制超时
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        //当线程数大于最大线程数,或者需要超时控制且上次获取任务超时
        //且线程数大于1或者队列为空,尝试将线程数减一并返回null
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            //失败重试
            continue;
        }

        try {
            //当需要超时控制时,在keepAliveTime时间内没有获取到任务的话返回null,否则调用take获取任务,此时线程时阻塞的
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

getTask方法在线程数量大于核心线程数时会判断在获取task时进行超时判断(poll),超时返回null这时getTask返回null,那当前worker的loop结束即run方法结束,线程生命周期结束。而核心线程则会调用take方法,当没有任务时会阻塞。

processWorkerExit

runTask方法最后会调用processWorkerExit方法进行一些cleanup工作。

private void processWorkerExit(Worker w, boolean completedAbruptly) {
    //completedAbruptly为true时代表发生了异常,线程数减一
    if (completedAbruptly) // If abrupt, then workerCount wasn"t adjusted
        decrementWorkerCount();

    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        //统计完成任务数
        completedTaskCount += w.completedTasks;
        //线程池移除当前worker
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }
    // 根据线程池状态进行判断是否结束线程池
    tryTerminate();

    int c = ctl.get();
    //当线程池状态为RUNNING或者SHUTDOWN时
    //如果发生异常,重新加入一个worker replacement
    if (runStateLessThan(c, STOP)) {
        if (!completedAbruptly) {
            //当allowCoreThreadTimeOut为true,最少要一个worker
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            //当线程数大于等于最少需要的线程数,则不需要add新的worker
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        addWorker(null, false);
    }
}

tryTerminate方法

上面我们跳过了tryTerminate方法,该方法判断是否要结束线程池,这里看一下

final void tryTerminate() {
    for (;;) {
        int c = ctl.get();
        //当线程池状态时RUNNING或者已经TIDYING或者已经TERMINATED或者SHUTDOWN且还有任务没有被执行,直接返回
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
        // 如果线程数不为0,则中断一个空闲的工作线程
        if (workerCountOf(c) != 0) { // Eligible to terminate
            //workQueue.take()时如果queue一直为空的话,线程会一直阻塞
            interruptIdleWorkers(ONLY_ONE);
            return;
        }

        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            //如果状态设置成功为TIDYING,调用勾子方法terminated,该方法留给了子类实现
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                    terminated();
                } finally {
                    //设置状态为TERMINATED
                    ctl.set(ctlOf(TERMINATED, 0));
                    termination.signalAll();
                }
                return;
            }
        } finally {
            mainLock.unlock();
        }
        // else retry on failed CAS
    }
}
interruptIdleWorkers

上面说为了当队列一直为空的时候,核心线程会一直阻塞,所以调用了interruptIdleWorkers,我们看一下执行了什么:

private void interruptIdleWorkers(boolean onlyOne) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (Worker w : workers) {
            Thread t = w.thread;
            if (!t.isInterrupted() && w.tryLock()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                } finally {
                    w.unlock();
                }
            }
            if (onlyOne)
                break;
        }
    } finally {
        mainLock.unlock();
    }
}

遍历线程池中所有的线程,若线程没有被中断tryLock成功,就中断该线程,LockSupport.park()能响应中断信号,阻塞的线程被中断唤醒。

文章版权归作者所有,未经允许请勿转载,若此文章存在违规行为,您可以联系管理员删除。

转载请注明本文地址:https://www.ucloud.cn/yun/70949.html

相关文章

  • 使用 Executors,ThreadPoolExecutor,创建线程源码分析理解

    摘要:源码分析创建可缓冲的线程池。源码分析使用创建线程池源码分析的构造函数构造函数参数核心线程数大小,当线程数,会创建线程执行最大线程数,当线程数的时候,会把放入中保持存活时间,当线程数大于的空闲线程能保持的最大时间。 之前创建线程的时候都是用的 newCachedThreadPoo,newFixedThreadPool,newScheduledThreadPool,newSingleThr...

    Chiclaim 评论0 收藏0
  • 后端ing

    摘要:当活动线程核心线程非核心线程达到这个数值后,后续任务将会根据来进行拒绝策略处理。线程池工作原则当线程池中线程数量小于则创建线程,并处理请求。当线程池中的数量等于最大线程数时默默丢弃不能执行的新加任务,不报任何异常。 spring-cache使用记录 spring-cache的使用记录,坑点记录以及采用的解决方案 深入分析 java 线程池的实现原理 在这篇文章中,作者有条不紊的将 ja...

    roadtogeek 评论0 收藏0
  • 线程源码分析

    摘要:线程池的作用线程池能有效的处理多个线程的并发问题,避免大量的线程因为互相强占系统资源导致阻塞现象,能够有效的降低频繁创建和销毁线程对性能所带来的开销。固定的线程数由系统资源设置。线程池的排队策略与有关。线程池的状态值分别是。 线程池的作用 线程池能有效的处理多个线程的并发问题,避免大量的线程因为互相强占系统资源导致阻塞现象,能够有效的降低频繁创建和销毁线程对性能所带来的开销。 线程池的...

    enda 评论0 收藏0
  • 一看就懂的Java线程分析详解

    摘要:任务性质不同的任务可以用不同规模的线程池分开处理。线程池在运行过程中已完成的任务数量。如等于线程池的最大大小,则表示线程池曾经满了。线程池的线程数量。获取活动的线程数。通过扩展线程池进行监控。框架包括线程池,,,,,,等。 Java线程池 [toc] 什么是线程池 线程池就是有N个子线程共同在运行的线程组合。 举个容易理解的例子:有个线程组合(即线程池,咱可以比喻为一个公司),里面有3...

    Yangder 评论0 收藏0
  • Java ThreadPoolExecutor 线程源码分析

    摘要:线程池常见实现线程池一般包含三个主要部分调度器决定由哪个线程来执行任务执行任务所能够的最大耗时等线程队列存放并管理着一系列线程这些线程都处于阻塞状态或休眠状态任务队列存放着用户提交的需要被执行的任务一般任务的执行的即先提交的任务先被执行调度 线程池常见实现 线程池一般包含三个主要部分: 调度器: 决定由哪个线程来执行任务, 执行任务所能够的最大耗时等 线程队列: 存放并管理着一系列线...

    greatwhole 评论0 收藏0
  • Java调度线程ScheduledThreadPoolExecutor源码分析

    摘要:当面试官问线程池时,你应该知道些什么一执行流程与不同,向中提交任务的时候,任务被包装成对象加入延迟队列并启动一个线程。当我们创建出一个调度线程池以后,就可以开始提交任务了。 最近新接手的项目里大量使用了ScheduledThreadPoolExecutor类去执行一些定时任务,之前一直没有机会研究这个类的源码,这次趁着机会好好研读一下。 原文地址:http://www.jianshu....

    kohoh_ 评论0 收藏0

发表评论

0条评论

stormzhang

|高级讲师

TA的文章

阅读更多
最新活动
阅读需要支付1元查看
<