Java线程池运行原理，线程池源码解读【Java线程池学习二】

一、前奏

有了上一篇博文的学习，相信你对于线程池的使用这块已经不在存在什么问题了，日常开发和面试也都足够了。线程池最优使用策略【Java线程池学习一】

但随着时间的推移在闲下来的时候我突然想，当任务进入了队列之后是怎么取出来的呢？然后列举了几个问题

添加的一个任务是怎么运行的？
任务丢到了队列，怎么取出来呢？
过了时间怎么销毁线程？
怎么拒绝的？
线程池，这个池是什么？线程怎么放进去？

毫无疑问想要解决上面的问题，那只有研究源码，下面我们就来看下 ThreadPoolExecutor 的源码，此次目的就是解决上面的问题，先对线程池的核心工作原理进行理解，后面我们再来对线程池来一个全面的解读。

二、开始

public static void main(String[] args) throws InterruptedException {

    ThreadPoolExecutor executor = new ThreadPoolExecutor(1,2, 1, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1));

    executor.execute(()->{
        try {
            Thread.sleep(1000 * 6);
            System.out.println("task-1 " + Thread.currentThread().getName()+ " " + Thread.currentThread().hashCode());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });

    executor.execute(()->{
        System.out.println("task-2 " + Thread.currentThread().getName()+ " " + Thread.currentThread().hashCode());
    });

    executor.execute(()->{
        System.out.println("task-3 " + Thread.currentThread().getName()+ " " + Thread.currentThread().hashCode());
    });

    Thread.sleep(1000 * 10);

    executor.execute(()->{
        System.out.println("task-4 " + Thread.currentThread().getName()+ " " + Thread.currentThread().hashCode());
    });
}

上面的代码案例很简单，现在我们用上一篇博文的知识来解读一下这段代码：

task-1 进入线程池，这时候线程池里面还没有线程，所以会启一个线程去执行【thread-1】
task-2 进入线程池，虽然线程池里面有一个【thread-1】，但是它被上一个任务阻塞着，所以 task-2进入队列
task-3 进入线程池，这时候队列满了，【thread-1】还在阻塞中，所以会开启一个新的线程去执行【thread-2】
task-3 执行完毕后，队列中还有一个 task-2，所以【thread-2】会去执行 task-2
执行完 task-2 之后，我们暂停添加任务，【thread-2】会在等待 1s 之后就销毁了
这时候 task-1执行完毕，过了一会我们又添加了一个 task-4 到线程池，【thread-1】是长期存活的它会去执行 task-4

打印结果如下：

task-3 pool-1-thread-2 448689266
task-2 pool-1-thread-2 448689266
task-1 pool-1-thread-1 1721383753
task-4 pool-1-thread-1 1721383753

三、分析

注：ThreadPoolExecutor 的源码里面用到了与或非左移右移运算，理解起来比较困难，但又不是业务的主流程，所以我在后面进行简化。

3-1、execute(Runnable command)

上面的案例代码我们只是调用了线程池的 execute 方法，所以我们直接来看看这个方法。

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
        
   /**
    * workerCountOf(c) 就是返回的当前线程池存在的线程数
    * 
    * 如果小于核心线程数就进行 addWorker ，把任务添加进去，添加成功就结束（注这个addWorker并不是加入队列）
    */
    int c = ctl.get();
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    
   /**
    * 当前线程池没有在跑，并成功把当前任务放入队列中
    *
    * 插入成功后会检查一下当前线程池的运行情况，如果isRunning返回false说明线程池已经超负荷运行了 就删除任务 并拒绝任务
    *
    * 如果上一步没有问题，就再判断一下当然线程池是否有在运行的线程，没有就开启一个，下面我们会详细说明【addWorker】
    */
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            reject(command);
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    
   /**
    * 如果插入队列失败了（以超过核心线程数来工作），会尝试直接运行该任务，如果还是失败，就拒绝
    */
    else if (!addWorker(command, false))
        reject(command);
}

offer 方法是在队列允许的范围内把数据插入进去，如果容量不足就返回false
线程池的实时性很高，所以可以看到它的源码里面做了很多补偿的机制，比如任务插入到队列中去后还会再判断线程池的状态是否正常。（如果在插入成功的一瞬间线程池已经终止了就会出问题）

3-2、 addWorker(Runnable firstTask, boolean core)

firstTask 就是当前任务，当然它可以为null，如果是null就是让当前线程去执行队列中其它任务（后面会看到）
core 判断线程的数量是在核心线程数量还是最大线程数量的范围（true 就是核心线程数量范围内）

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        // 检查特殊情况下队列是否为空
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;

        for (;;) {
            // wc 为当前线程池的线程数
            int wc = workerCountOf(c);
            // 如果线程数大于最大容量 或 大于规定容量就返回 false
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            // 使用CAS去让当前线程数+1 成功则【整个退出】    
            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
        }
    }

    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        // Worker 是private final修饰的内部类，实现了 Runnable接口，里面有【当前任务和当前线程】下面详解
        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());
               /**
                * 判断当前任务是否符合加入到任务中的条件
                *
                * rs 是一个很大的负数 可以理解规定能最大创建的线程数【536870911】 所以不用考虑，不会创建那么多
                * SHUTDOWN 是个常数 0
                */
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    // 判断当前线程是否可以被 开启/start
                    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;
}

关于 t.isAlive() 可以参看 https://www.javatpoint.com/java-thread-isalive-method

3-3、Worker

现在我们来看一下Worker这个内部类，上面我们也说了Worker 实现了Runnable，并且在最后的一步中调用了 start方法。

它继承了AbstractQueuedSynchronizer 实现了 Runnable，但代码很简单，定义了三个参数，代码比较简单直接看。

thread 执行当前任务的线程
firstTask 当前任务
completedTasks 该线程执行成功的任务数（会统计整个线程池执行成功的任务数）

private final class Worker
    extends AbstractQueuedSynchronizer
    implements Runnable {
 
    private static final long serialVersionUID = 6138294804551838833L;

    final Thread thread;

    Runnable firstTask;

    volatile long completedTasks;

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

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

我们单独来看一下它的构造方法，任务是由外面传递过来的，线程是每次创建一个新的，上面我们在创建的时候没有指定ThreadFactory，则用的默认的 Executors.DefaultThreadFactory

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

3-4、runWorker(Worker w)

3-2 里面我们调用了 worker 的 start方法，后续肯定就会调用 run 方法了，也就是 runWorker 方法

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

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
            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 里面没有任何实现是空方法，可以留给后续扩展。
我们以为当前线程只是去执行当前的任务/task，但其实不是 task != null || (task = getTask()) != null 它先去执行自己的任务，如果自己的任务执行完了，就会执行 getTask 从队列中拿任务直到队列中没有任务为止。
我们在上面也看到调用 addWorker方法的时候任务传递的是 null，其实就是让它去执行队列中的任务。

3-5、getTask()

该方法如同它自己的名【获取任务】，它的作用就是从队列中获取任务。

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

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

        // 特殊情况下校验队列是不是为空
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }

        int wc = workerCountOf(c);

       /**
        * allowCoreThreadTimeOut 是否允许核心线程超时 默认 false
        * wc > corePoolSize  当前活跃线程是否大于核心线程数
        * 
        * 其实就是相当于只有超过核心线程数的线程才设置超时时间
        */
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        
        /**
         * 如果当前线程数 > 最大线程数 就退出（销毁当前线程）
         * 或 当前线程是设置了超时的，并且已经超时了 就退出（销毁当前线程）
         */
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }

        try {
           /**
            * 从队列中获取数据
            * 
            * poll 等待固定时间
            * take 一直阻塞，直到获取到了任务数据
            */
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

take方法一直阻塞，直到获取到了任务数据

四、问题剖析

4-1、添加的一个任务是怎么运行的？

黑色的线是正常添加一个任务的流程
红色的线是线程组里面的线程一直阻塞获取队列中的数据

在这里插入图片描述

4-2、任务丢到队列，怎么取出来呢？

下面是截取 getTask() 里面的一段代码

 Runnable r = timed ?
    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
    workQueue.take();

简单理解就是非核心线程调用 poll 方法，核心线程调用 take 方法，take方法是阻塞的，获取不到数据会一直等着。

4-3、过了时间怎么销毁线程？

咱们把 getTask 方法复制过来，删掉多余的部分

private Runnable getTask() {
    boolean timedOut = false; // Did the last poll() time out?
    for (;;) {
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }

        try {
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

我们知道超过核心线程的部分会调用 poll 方法，如果超过时间没有获取到对应的任务： timedOut = true

如果 timedOut = true，那上面的if判断就会进去，当前死循环就结束了，线程也就正常结束。

4-4、怎么拒绝的？

拒绝这个就很简单了，在excute方法的讲解里面已经说了。

4-5、线程池，这个池是什么？线程怎么放进去？

再来回顾一下是怎么创建线程的，在进入 addWorker 方法的时候会进行一系列的判断，当判断通过后就会创建一个 w = new Worker(firstTask);

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

找到这个 Worker 的构造方法再来看一下，前面我们也说了默认的线程工厂是Executors.DefaultThreadFactory，来看看它的 newThread 方法

DefaultThreadFactory

static class DefaultThreadFactory implements ThreadFactory {
    private static final AtomicInteger poolNumber = new AtomicInteger(1);
    private final ThreadGroup group;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;

    DefaultThreadFactory() {
        SecurityManager s = System.getSecurityManager();
        group = (s != null) ? s.getThreadGroup() :
                              Thread.currentThread().getThreadGroup();
        namePrefix = "pool-" +
                      poolNumber.getAndIncrement() +
                     "-thread-";
    }

    public Thread newThread(Runnable r) {
        Thread t = new Thread(group, r,
                              namePrefix + threadNumber.getAndIncrement(),
                              0);
        if (t.isDaemon())
            t.setDaemon(false);
        if (t.getPriority() != Thread.NORM_PRIORITY)
            t.setPriority(Thread.NORM_PRIORITY);
        return t;
    }
}

可以看到其实每次都是 new Thread，但是它的 ThreadGroup 是同一个。

线程启动的时候是调用的 Thread.start() 方法，隐藏其它方法，里面有一个把当前线程添加到线程组里面。

public synchronized void start() {
    // ...
    
    group.add(this);

   // ...
}

ThreadGroup.add
int nthreads;
Thread threads[];

void add(Thread t) {
    synchronized (this) {
        if (destroyed) {
            throw new IllegalThreadStateException();
        }
        if (threads == null) {
            threads = new Thread[4];
        } else if (nthreads == threads.length) {
            threads = Arrays.copyOf(threads, nthreads * 2);
        }
        threads[nthreads] = t;
        nthreads++;
        nUnstartedThreads--;
    }
}