本文共 17082 字，大约阅读时间需要 56 分钟。

1.线程池简介

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

public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,                                      60L, TimeUnit.SECONDS,                                      new SynchronousQueue
   
    (),                                      threadFactory);    }    public static ExecutorService newSingleThreadExecutor() {        return new FinalizableDelegatedExecutorService            (new ThreadPoolExecutor(1, 1,                                    0L, TimeUnit.MILLISECONDS,                                    new LinkedBlockingQueue
    
     ()));    }    public static ExecutorService newFixedThreadPool(int nThreads) {        return new ThreadPoolExecutor(nThreads, nThreads,                                      0L, TimeUnit.MILLISECONDS,                                      new LinkedBlockingQueue
     
      ());    }
     
    
   

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

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参数，该参数会在线程池启动之后立马创建核心池数量的线程）。随着任务的不断增加，现有线程无法满足要求，就会不断的创建新线程，直到线程数达到corePoolSize的值，后续新来的任务会放入阻塞队列；
• maximumPoolSize: 最大池大小，当任务太多，阻塞队列满了之后，如果线程数量还没有超过该参数的值，就会继续创建新线程，直到线程数达到该参数规定的值，后续再来的任务会使用拒绝策略进行处理；
• keepAliveTime: 如果线程数超过corePoolSize的值，那么多余的线程在空闲keepAliveTime时间后会被销毁；
• unit: keepAliveTime参数的单位；
• workQueue: 阻塞队列；
• threadFactory: 线程工厂，创建线程时需要使用到该工厂；
• handler: 拒绝策略。

2.核心字段

ThreadPoolExecutor的核心字段如下：

//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;

3.线程池状态

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

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

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

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()

public List
   
     shutdownNow() {        List
    
      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();        }    }
    
   

4.执行任务

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

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()方法的实现：

//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
   
    = 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类型的实例，其源码如下（省略了无关代码）：

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循环中不断获取任务去执行，该方法源码如下：

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()从阻塞队列中取任务，其实现如下：

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()方法。

5.线程池是如何关闭的

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

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

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

不论是调用哪个方法来关闭线程池，都可能会遇到线程陷入类似于synchronized导致的阻塞状态，处在这种状态的线程无法响应中断，因此这些线程以及其他还没有结束的线程的退出要由getTask()方法来决定。getTask()方法的自旋代码会首先检查线程池的状态，如下：

if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {        decrementWorkerCount();        return null;    }

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

转载地址：http://tlkkz.baihongyu.com/