手把手教你看懂线程池源码!

2020-04-24 18:06:18 浏览数 (1)

线程池简介

使用线程池,一般会使用JDK提供的几种封装类型,即:newFixedThreadPool、newSingleThreadExecutor、newCachedThreadPool等,这些线程池的定义在Executors类中,来看看相关的源码:

代码语言:javascript复制
public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
              60L, TimeUnit.SECONDS,
              new SynchronousQueue<Runnable>(),
              threadFactory);
}
public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<Runnable>()));
}
public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
          0L, TimeUnit.MILLISECONDS,
          new LinkedBlockingQueue<Runnable>());
}

这些方法内部都使用了ThreadPoolExecutor的构造方法,区别只是传入的参数不同。ThreadPoolExecutor有四个重载的构造方法,最终调用的是由7个参数的构造器,其源码如下:

代码语言:javascript复制
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:核心池大小,默认情况下,线程池启动之后,并不会立即创建线程,而是要等到任务到来之后,才创建线程去执行任务(除非设置了allowCoreThreadTimeOut参数,该参数会在线程池启动之后立马创建核心池数量的线程)。随着任务的不断增加,现有线程无法满足要求,就会不断的创建新线程,直到线程数达到corePoolSize的值,后续新来的任务会放入阻塞队列;
  • maximumPoolSize: 最大池大小,当任务太多,阻塞队列满了之后,如果线程数量还没有超过该参数的值,就会继续创建新线程,直到线程数达到该参数规定的值,后续再来的任务会使用拒绝策略进行处理;
  • keepAliveTime: 如果线程数超过corePoolSize的值,那么多余的线程在空闲keepAliveTime时间后会被销毁;
  • unit: keepAliveTime参数的单位;
  • workQueue: 阻塞队列;
  • threadFactory: 线程工厂,创建线程时需要使用到该工厂;
  • handler: 拒绝策略。

核心字段

ThreadPoolExecutor的核心字段如下:

代码语言:javascript复制
//ctl低29位表示线程的数量,高3位表示线程池状态,因此当前线程池允许的最大线程数量是2^29-1
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//固定值29
private static final int COUNT_BITS = Integer.SIZE - 3;
//线程最大容量
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;
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;

线程池状态

线程池的状态有5种,状态之间的转换关系如下图:

初始情况下,线程池创建完毕后会处于RUNNING状态,可以正常的接受新任务;当调用shutdown()时,线程池变成SHUTDOWN状态,此时无法接受新任务,但是会继续执行阻塞队列中的任务;当调用shutdownNow()时,线程由RUNNING状态变成STOP状态,此时不能接受新任务,并且会中断正在执行的任务;当线程池中的线程数减少为0时,就会转成TIDYING状态;在TIDYING状态会自动调用terminated()使线程池转为TERMINATED状态。

shutdown()

shutdown()方法的逻辑分别由5个不同的方法来实现,这里将这些方法整理在一起,如下:

代码语言:javascript复制
public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        //检查security manager是否允许调用方执行此方法
        checkShutdownAccess();
        //将线程池状态更新为SHUTDOWN
        advanceRunState(SHUTDOWN);
        //中断空闲线程
        interruptIdleWorkers();
        //这是一个空实现,允许子类进行重写
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
}
private void advanceRunState(int targetState) {
    for (;;) {
        int c = ctl.get();
        //如果线程池已经处在targetState及之后的状态则直接结束循环,否则使用CAS操作将线程池状态更新为targetState
        if (runStateAtLeast(c, targetState) ||
            ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
            break;
    }
}
private void interruptIdleWorkers() {
    interruptIdleWorkers(false);
}
//onlyOne表示是否只终止一个空闲线程
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();
    }
}
final void tryTerminate() {
    //自旋
    for (;;) {
        int c = ctl.get();
        //线程池还在运行,或者已经是TIDYING或TERMINATED状态,或者已经处在`SHUTDOWN`状态但阻塞队列不为空,这几种情况不再继续执行
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
        //线程数不为0时,终止一个空闲线程
        if (workerCountOf(c) != 0) { // Eligible to terminate
            interruptIdleWorkers(ONLY_ONE);
            return;
        }
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            //将线程池设置为DIDYING状态
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                //设置成功后,执行terminated()方法
                try {
                    //这也是一个空实现,子类可以根据需要进行重写
                    terminated();
                } finally {
                    //将线程池设置为TERMINATED状态
                    ctl.set(ctlOf(TERMINATED, 0));
                    termination.signalAll();
                }
                return;
            }
        } finally {
            mainLock.unlock();
        }
        // else retry on failed CAS
    }
}

shutdownNow()

代码语言:javascript复制
public List<Runnable> shutdownNow() {
    List<Runnable> tasks;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        //检查security manager是否允许调用方执行此方法
        checkShutdownAccess();
        //将线程池状态更新为STOP
        advanceRunState(STOP);
        //与shutdown的区别是,这里会中断所有线程,而不仅仅是空闲线程
        interruptWorkers();
        //将任务从workQueue中移除,转移到一个ArrayList中,此操作后,workQueue为空,已有的任务无法继承执行
        tasks = drainQueue();
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
    return tasks;
}
//中断所有线程
private void interruptWorkers() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (Worker w : workers)
            w.interruptIfStarted();
    } finally {
        mainLock.unlock();
    }
}

执行任务

线程池通过execute()方法执行任务,其源码如下:

代码语言:javascript复制
public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    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);
        //如果活跃线程为0,则创建一个非核心线程,并将firstTask设置为null
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    //如果添加非核心线程失败,则执行拒绝策略
    else if (!addWorker(command, false))
        reject(command);
}
//获取活跃的线程数
private static int workerCountOf(int c)  { return c & CAPACITY; }
//获取线程池运行状态
private static int runStateOf(int c)     { return c & ~CAPACITY; }

下图是execute()方法的执行逻辑:

来看看addWorker()方法的实现:

代码语言:javascript复制
//core表示要创建的是否是核心线程,true表示创建核心线程,false表示创建非核心线程
private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        int c = ctl.get();
        //获取线程池状态
        int rs = runStateOf(c);
        // Check if queue empty only if necessary.
        //rs >= SHUTDOWN,表示线程池不再处于RUNNING状态
        //rs>=SHUTDOWN,说明已经调用了shutdown()或者shutdownNow()方法,在此条件满足的情况下,第二项条件等同于
        //rs!=SHUTDOWN || firstTask != null || workQueue.isEmpty(),满足这三个条件的任何一个都不会再添加新任务
        //rs!=SHUTDOWN,说明是STOP、TIDYING、TERMINATE这三种
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;
        //执行到这里说明:
        //① rs<SHUTDOWN,即线程池是运行状态
        //② rs=SHUTDOWN,farstTask=null, 并且阻塞队列不为空
        for (;;) {
            int wc = workerCountOf(c);
            //有三种情况会返回false:1)线程数达到最大值;2)当前创建核心线程,但是线程数已经达到核心池大小;
            //3)当前创建非核心线程,并且线程数达到最大池大小
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            //如果使用CAS操作成功将ctl的值加1,则跳出最外层循环
            if (compareAndIncrementWorkerCount(c))
                break retry;
            //走到这里说明无法使用CAS更新ctl的值,说明此时发生了多线程竞争,需要重新查看线程池的状态
            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);
        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)) {
                    //如果线程t的start()方法已经被执行过,则抛出异常
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    //workers是个HashSet类型,只在重入锁代码中被访问
                    workers.add(w);
                    //更新当前活跃线程的最大值
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {
                //线程创建成功,则启动线程,内部会调用Worker类的run()方法
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        //成功创建新线程时,才会设置workerStarted=true,这里处理没有创建新线程的情况
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

addWorker() 方法中用到了 Worker 类,这是 ThreadPoolExecutor 的内部类,对线程进行了包装,线程池创建或者启动的线程,实际上都是 Worker 类型的实例,其源码如下(省略了无关代码):

代码语言:javascript复制
private final class Worker extends AbstractQueuedSynchronizer implements Runnable{
    /** 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;
    //构造器
    Worker(Runnable firstTask) {
        setState(-1); 
        this.firstTask = firstTask;
        //注意,这里是将Worker实例传入线程工厂进行构造,因此在调用线程的start()方法时,内部会调用Worker类的run()方法
        this.thread = getThreadFactory().newThread(this);
    }
    /** Delegates main run loop to outer runWorker  */
    public void run() {
        runWorker(this);
    }
    // ....
}

当启动Worker线程时,会通过Thread类的start()方法调用Worker类的runWorker()方法,每一个启动的线程都会在该方法的while循环中不断获取任务去执行,该方法源码如下:

代码语言:javascript复制
final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        //如果能够成功拿到任务,则执行下面的代码块,如果getTask()方法返回null,当前线程就会执行退出逻辑
        while (task != null || (task = getTask()) != null) {
            //如果能将state字段设置为1,表示成功拿到锁,就接着向下执行,否则线程会加入等待队列,不再继续执行
            //注意这里是在成功拿到新任务之后才会加锁,结合shutdown()方法的逻辑
            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);
    }
}

beforeExecute()和afterExecute()是protected类型,并且默认是空实现,很明显是留给子类去实现钩子逻辑。上面的代码使用getTask()从阻塞队列中取任务,其实现如下:

代码语言:javascript复制
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.
        //线程池正在关闭,或者阻塞队列空了,就减少线程数,并返回null
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }
        int wc = workerCountOf(c);
        // Are workers subject to culling?
        //在设置了allowCoreThreadTimeOut参数后,超过给定的时间,会将空闲的核心线程清理掉
        //或者线程数量超过了核心池数量,会在一定时间后清理掉多余的线程
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        //1)线程数量超过最大池数量,或者超时; 2)线程数大于1,或者阻塞队列为空; 这两个条件都成立时,就将ctl值减1
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }
        try {
            //如果设置了超时状态,则使用poll方法取任务,超过keepAliveTime还没有任务到来就返回true
            //否则使用take取任务,在阻塞队列为空时会一直等待
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            //线程有可能在等待新任务的到来而阻塞,但是在等待的过程中调用shutdownNow()关闭线程时,线程会抛出中断异常,在这里被捕获
            timedOut = false;
        }
    }
}

现在来整理一下runWorker()方法的思路:每一个新创建的线程都会在runWorker()方法里通过while循环不断地从阻塞队列中获取任务,取到任务之后就执行任务的run()方法,取不到任务就会一直阻塞,或者等待一定的时间之后,空闲线程超时需要回收,就会执行processWorkerExit()方法。

线程池是如何关闭的

在介绍shutdown()方法时有一个疑问,该方法只会中断空闲线程,但是非空闲的线程不会被中断,即使该线程被阻塞,因此该方法有可能无法关闭那些一直处在等待状态的非空闲线程,这一点在使用时需要注意。在runWorker()方法中,while循环会在成功拿到任务后才会加锁,因此那些由于阻塞队列为空拿不到任务而阻塞的线程也会被shutdown()方法中断

代码语言:javascript复制
while (task != null || (task = getTask()) != null) {
    //如果能将state字段设置为1,表示成功拿到锁,就接着向下执行,否则线程会加入等待队列,不再继续执行
    //注意这里是在成功拿到新任务之后才会加锁,结合shutdown()方法的逻辑
    w.lock();
    //忽略其他代码
}

shutdownNow()会中断所有的存活线程,不论这些线程是否空闲,因此可能会导致任务在执行的过程中抛出异常,这点需要注意。

不论是调用哪个方法来关闭线程池,最终线程的退出是要根据getTask()方法来决定。当getTask()方法返回null,即当前阻塞队列已经没有任务时,线程会退出,并且在getTask()方法的自旋代码会首先检查线程池的状态,如下:

代码语言:javascript复制
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
    decrementWorkerCount();
    return null;
}

在调用shutdownNow()方法关闭线程池后,rs >= STOP逻辑成立,直接返回null,而shutdown()方法会继续执行阻塞队列中的任务,直到workQueue.isEmpty()条件为真,getTask()返回null导致线程一个个结束,不论是哪种情况,最终线程池中的线程数量都会变成0。

0 人点赞