Java并发之线程池使用及源码剖析

线程池

ThreadPoolExecutor(int corePoolSize,
    int maximumPoolSize,
    long keepAliveTime,
    TimeUnit unit,
    BlockingQueue<Runnable> workQueue, 
    ThreadFactory threadFactory,    RejectedExecutionHandler handler)

• corePoolSize 线程池核心线程数最大值,核心线程默认情况下即使在没有工作时也会保留
• maximumPoolSize 线程池最大线程数大小
• keepAliveTime 非核心线程空闲的存活时间大小
• unit 线程空闲存活时间单位
• workQueue 用于存放任务的阻塞队列,核心线程满时新任务先放入阻塞队列，队列满时再创建非核心线程，如果超过最大线程数，则抛出异常
• threadFactory线程工厂，主要用来创建线程，比如可以指定线程的名字
• handler 当队列和最大线程池都满了之后的饱和策略，主要有四种类型

使用

通过execute(Runnable command)方法来向线程池提交任务

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Main {

    public static void main(String[] args){
        ExecutorService threadPool = Executors.newFixedThreadPool(5);
        for(int i=0;i<5;i++)
            threadPool.execute(()->{
                System.out.println(Thread.currentThread().getName());
            });
        threadPool.shutdown();
    }
}

常见线程池

newCachedThreadPool

public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>(),
                                      threadFactory);
    }

• corePoolSize => 0，核心线程池的数量为0
• maximumPoolSize => Integer.MAX_VALUE，线程池最大数量为Integer.MAX_VALUE，可以认为可以无限创建线程
• keepAliveTime => 60L
• unit => 秒
• workQueue => SynchronousQueue
• 因为Integer.MAX_VALUE非常大，可以认为是可以无限创建线程的，在资源有限的情况下容易引起OOM异常。由于空闲 60 秒的线程会被终止，长时间保持空闲的 CachedThreadPool 不会占用任何资源。
• 用于并发执行大量短期的小任务

newFixedThreadPool

public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

FixedThreadPool是固定核心线程的线程池，固定核心线程数由用户传入

• corePoolSize => nThreads
• maximumPoolSize => nThreads，线程池最大数量为nThreads，即最多只可以创建nThreads个线程
• keepAliveTime => 0L
• unit => 毫秒
• workQueue => LinkedBlockingQueue
• 使用的是LinkedBlockingQueue，在资源有限的时候容易引起OOM异常
• FixedThreadPool 适用于处理CPU密集型的任务，确保CPU在长期被工作线程使用的情况下，尽可能的少的分配线程，即适用执行长期的任务。

newSingleThreadExecutor

public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

• 核心线程数为1
• 最大线程数也为1
• 阻塞队列是LinkedBlockingQueue
• keepAliveTime为0
• 适用于串行执行任务的场景，一个任务一个任务地执行。

newScheduledThreadPool

public ScheduledThreadPoolExecutor(int corePoolSize,
                                       ThreadFactory threadFactory) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue(), threadFactory);
    }

• 最大线程数为Integer.MAX_VALUE
• 阻塞队列是DelayedWorkQueue
• keepAliveTime为0
• scheduleAtFixedRate() ：按某种速率周期执行
• scheduleWithFixedDelay()：在某个延迟后执行
• 周期性执行任务的场景，需要限制线程数量的场景

原理

线程池的状态

线程池有5种状态

• RUNNING，接收新任务，处理缓存的任务
• SHUTDOWN，不接受新任务，但处理缓存的任务
• STOP，不接受新任务，不处理缓存的任务，中断正在执行的任务
• TIDYING，所有线程都已终止，有效线程为0，触发_terminated()_方法
• TERMINATED，_terminated()_方法结束

线程池的状态通过一个AtomicInteger来实现，使用3位来代表状态，剩下的位用于统计线程数。线程使用Worker来包装。

源码

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.
         *
         * 1. 如果线程数少于corePoolSize，则尝试创建一个新的线程，用传入的command对象作为此线程
         * 要首先执行的任务。 addWorker原子性的检查线程池的状态和worker数量，通过返回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.
         *
         * 2. 如果一个任务被成功的加入队列，仍旧需要二次检查确认是否需要新建一个线程（因为从上一次检
         * 查以来可能有线程结束）或者在进入此方法后线程池shutdown。所以我们重新检查状态，如果线程池
         * 已关闭则撤回入队操作，或者在没有线程时新建一个线程
         *
         * 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.
         *
         * 3. 如果不能将任务加入队列，那么尝试新建一个线程。如果失败了则拒绝任务。
         */
        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            //工作线程少于核心线程时，添加一个worker来执行任务，true代表使用核心线程
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
    	
        if (isRunning(c) && workQueue.offer(command)) {
            //到这里，任务进入了队列
            int recheck = ctl.get();
            //重新检查，如果线程池不running了，移除并拒绝
            if (! isRunning(recheck) && remove(command))
                reject(command);
            //如果线程池仍在运行，且在运行的线程数为0，那么开启新的线程
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
    	//添加任务失败，拒绝
        else if (!addWorker(command, false))
            reject(command);
    }

再看addWorker方法

/**
     * Checks if a new worker can be added with respect to current
     * pool state and the given bound (either core or maximum). If so,
     * the worker count is adjusted accordingly, and, if possible, a
     * new worker is created and started, running firstTask as its
     * first task. This method returns false if the pool is stopped or
     * eligible to shut down. It also returns false if the thread
     * factory fails to create a thread when asked.  If the thread
     * creation fails, either due to the thread factory returning
     * null, or due to an exception (typically OutOfMemoryError in
     * Thread.start()), we roll back cleanly.
     * 
     * 根据线程池的状态以及给的上限（核心线程数或最大线程数）检查是否可以创建新的worker
     * 如果可以，那么调整worker的数量，如果新的worker被创建和启动，那么运行它的firstTask
     * 如果线程池已经stopped或者可以shutdown，此方法返回false。如果threadfactory创建
     * 线程失败，同样返回false。如果threadfactory创建线程失败，要么是由于它返回了null，
     * 要么是出现异常，我们回滚并进行清理
     *
     * @param firstTask the task the new thread should run first (or
     * null if none). Workers are created with an initial first task
     * (in method execute()) to bypass queuing when there are fewer
     * than corePoolSize threads (in which case we always start one),
     * or when the queue is full (in which case we must bypass queue).
     * Initially idle threads are usually created via
     * prestartCoreThread or to replace other dying workers.
     *
     * @param firstTask 是新创建的线程应该首先执行的任务，如果没有的话为null。worker在创建时
     * excute方法中）会被附带上first task，以在线程数少于核心线程数（此时我们永远创建新线程）
     * 或队列满时（此时我们必须跳过排队）跳过排队。通常空闲线程是由已创建的核心线程创建的，或者用
     * 于替换已死线程
     *	
     * @param core if true use corePoolSize as bound, else
     * maximumPoolSize. (A boolean indicator is used here rather than a
     * value to ensure reads of fresh values after checking other pool
     * state).
     * 
     * @param core 为true使用核心线程数做上限，否则使用最大线程数。
     *
     * @return true if successful
     * @return 如果成功返回true if successful
     */
    private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);

            /* Check if queue empty only if necessary.
            * 这里有点复杂，把!分配进括号里展开，逻辑变成以下三个
            * 1.rs >= SHUTDOWN && rs != SHUTDOWN,也就是STOP,TIDYING,TERMINATED状态
            * 2.rs >= SHUTDOWN && firstTask != null
            * 3.rs >= SHUTDOWN && workQueue.isEmpty()
            * 以上三个条件满足则不创建新的worker，因为
            * 1.大于等于STOP状态，不再接受新任务，也不处理已缓存任务，自然不能添加worker
            * 2.到这里必然是SHUTDOWN状态，否则在第1步就已经为true了。此时如果firstTask != null
            * 	说明是创建线程用于执行新任务，当然不允许。注意如果firstTask为null是可以创建的，此时
           	* 	worker会从队列中取任务执行，符合SHUTDOWN状态的定义。
            * 3.SHUTDOWN且队列已经空了，那么肯定不允许新建worker。注意如果这里非空，那也是允许创建的
            * 	因为SHUTDOWN状态还要运行缓存的任务。
            */
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;

            for (;;) {
                int wc = workerCountOf(c);
                //如果线程数超过了CAPACITY或者给定的上限，那么返回false
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                //cas修改线程数，成功跳出外层循环
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                c = ctl.get();  // Re-read ctl 重新检查状态
                if (runStateOf(c) != rs)//如果线程池的状态变化了，那么重新外层循环
                    continue retry;
                // 那么cas失败是由于线程数变化，重复内层循环继续cas即可
                // 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)) {
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        //把worker加入集合中
                        workers.add(w);
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        //worker成功加入集合
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {
                    t.start();//启动worker中的线程
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

addWorker之后，调用worker中Thread的start方法，start方法又调用worker的run，worker中的run方法又调用runWorker方法，runWorker中最终调用task的run

final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
            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
                // 如果线程池正在stopping，确保中断线程，如果没有，确保不中断线程
                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();//终于调用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);
        }
    }