ThreadPoolExecutor源码走读

约定不等于承诺〃 2024-02-18 18:36 48阅读 0赞

工作中经常用到线程池，最常见的写法是Executors.newFixedThreadPool,线程池具体是如何处理我们提交的任务的呢？这里再走读源码记录一遍。

### 线程池这块的整体结构： ###

![20180311221826445][]

从整体上看类图，具体线程池具体实现其实就2个类ThreadPoolExecutor和ScheduledThreadPoolExecutor。

经常使用到的Executors类中可以看到它提供静态方法创建各种ThreadPoolExecutor以及ScheduledThreadPoolExecutor。  
例如，创建固定线程池：

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

创建单个线程的线程池：

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

创建具有延时执行功能的线程池：

public static ScheduledExecutorService newScheduledThreadPool(
                int corePoolSize, ThreadFactory threadFactory) {
            return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
    }

其实无非就是包装了创建线程池对象的方法,返回对应的实例，其实是工厂模式。  
具体threadFactory、LinkedBlockingQueue、ThreadPoolExecutor、ScheduledThreadPoolExecutor类干了什么后面再继续看。

### 1. ThreadPoolExecutor ###

先看ThreadPoolExecutor类，它继承了AbstractExecutorService，我们知道一般抽象类其实是统一实现了一部分通用的接口功能，避免各子类重复实现，从这一点分析，也可以看出来上面的整体类图是有遗漏的，不太可能只有一个ThreadPoolExecutor继承它，应该还有其它子类，继续分析它的子类，会发现在Executors在创建线程池时，有静态内部类继承了AbstractExecutorService：

static class DelegatedExecutorService extends AbstractExecutorService {
            private final ExecutorService e;
            DelegatedExecutorService(ExecutorService executor) { e = executor; }
            public void execute(Runnable command) { e.execute(command); }
            public void shutdown() { e.shutdown(); }
            public List<Runnable> shutdownNow() { return e.shutdownNow(); }
            public boolean isShutdown() { return e.isShutdown(); }
            public boolean isTerminated() { return e.isTerminated(); }
            public boolean awaitTermination(long timeout, TimeUnit unit)
                throws InterruptedException {
                return e.awaitTermination(timeout, unit);
            }
            public Future<?> submit(Runnable task) {
                return e.submit(task);
            }
            public <T> Future<T> submit(Callable<T> task) {
                return e.submit(task);
            }
            ...

这个方法包装了AbstractExecutorService，没有设置corePoolSize、keepAliveTime等方法，这个包装类用来创建单线程池  
  
(newSingleThreadExecutor/newSingleThreadScheduledExecutor)等不可设置这些属性的实例。接下来继续看ThreadPoolExecutor源码属性  
  
  
，它有几个状态值：

static final int RUNNING = 0;  
static final int SHUTDOWN = 1;  
static final int STOP = 2;  
static final int TERMINATED = 3;

线程池状态之间的切换：

![20180311221826800][]

其它属性：

private final BlockingQueue<Runnable> workQueue;
    private final HashSet<Worker> workers = new HashSet<Worker>();
    private final ReentrantLock mainLock = new ReentrantLock();
    private final Condition termination = mainLock.newCondition();
    private volatile long  keepAliveTime;
    private volatile boolean allowCoreThreadTimeOut;
    private volatile int   corePoolSize;
    private volatile int   maximumPoolSize;
    private volatile int   poolSize;
    ...

workQueue用于存储可执行的任务，是一个阻塞队列类型BlockingQueue。  
workers保存了所有线程池中工作的线程。  
mainLock锁用来控制poolSize、corePoolSize、maximumPoolSize、runState、workers集合。由此可见多线程环境下，对共享变量的所有访问都需要同步化。  
keepAliveTime如果池中当前有多于 corePoolSize 的线程，则这些多出的线程在空闲时间超过 keepAliveTime 时将会终止。  
allowCoreThreadTimeOut 超时策略应用于核心线程。  
corePoolSize核心线程数  
maximumPoolSize允许的最大线程数。  
poolSize池中的当前线程数。

#### 其内部方法有很多，那就先从我们最常用的方法开始看起，串成一条执行链路。 ####

newFixedThreadPool.execute(new ThreadForpools(index));

根据代码的执行逻辑，大概提交一个任务的执行流程如下：

![20180311221826592][]

看源码，你会发现都使用到了充入锁ReentrantLock：

private boolean addIfUnderCorePoolSize(Runnable firstTask) {
       Thread t = null;
       final ReentrantLock mainLock = this.mainLock;
       mainLock.lock();
       try {
           if (poolSize < corePoolSize && runState == RUNNING)
               t = addThread(firstTask);
       } finally {
           mainLock.unlock();
       }
       return t != null;
    }

mainLock 是全局属性，一个线程池实例共用一个锁。

#### 当任务添加到worders后，又是怎么执行的呢？ ####

worker实际上是线程池自己定义的一个内部类 java.util.concurrent.ThreadPoolExecutor.Worker：

![20180311221826578][]

当提交一个任务到workers中后，调用了线程的start方法，实际上执行了,Worker类的run方法：

public void run() {
                try {
                    hasRun = true;
                    Runnable task = firstTask;
                    firstTask = null;
                    while (task != null || (task = getTask()) != null) {
                        runTask(task);
                        task = null;
                    }
                } finally {
                    workerDone(this);
                }
     }

启动一个任务后，该线程会不停的调用getTask方法获取任务调用runTask执行:

Runnable getTask() {
            for (;;) {
                try {
                    int state = runState;
                    if (state > SHUTDOWN)
                        return null;
                    Runnable r;
                    if (state == SHUTDOWN)  // Help drain queue
                        r = workQueue.poll();
                    else if (poolSize > corePoolSize || allowCoreThreadTimeOut)
                        r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
                    else
                        r = workQueue.take();
                    if (r != null)
                        return r;
                    if (workerCanExit()) {
                        if (runState >= SHUTDOWN) // Wake up others
                            interruptIdleWorkers();
                        return null;
                    }
                    // Else retry
                } catch (InterruptedException ie) {
                    // On interruption, re-check runState
                }
            }
    }

getTask会判断线程池状态：

*  1.如果是STOP或者TERMINATED，则返回task=null，调用workers.remove(w) 从workers中移除。
 *  2.如果是SHUTDOWN，则获取任务队列workQueue头部的任务(workQueue.poll())
 *  3.如果是poolSize > corePoolSize || allowCoreThreadTimeOut，则返回workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS)，这里就体现了keepAliveTime的用法，如果allowCoreThreadTimeOut设置为true,则空闲的多余corePoolSize的线程将会在存活keepAliveTime时长后终止。
 *  4.如果是其他情况，则会调用workQueue.take() 一直阻塞，直到有新的任务添加到workQueue。

当获取到task!=null时，继续调用runTask执行，如果线程池的状态是STOP|TERMINATED,或者当前任务状态是hasRun但是被中断了，则中断当前任务线程，否则调用run方法，这里需要注意了，start和run方法的区别，直接调用run方法是不会创建新线程的，只是同步的run方法调用，当该任务线程执行完run方法后，又会堵塞到workQueue.take方法上，所以能够保证线程池的线程一直存活。

#### 再看下，线程池的状态是如何切换的 ####

runState因为是int，所以初始是0（RUNNING），有如下的方法会改变线程池的状态：

public void shutdown()  
public List shutdownNow()

1.调用shutDown方法，无返回。

*  置线程池状态为SHUTDOWN，中断所有workers空闲线程。这是判断是否线程空闲，用了个很巧妙的办法：  
    ![20180311221826542][]
 *  在执行runTask时，都会获取runLock锁，然后在判断是否空闲时，参数下获取该锁tryLock：  
    ![20180311221826784][]

如果没有获取到，说明该线程是任务执行状态，如果获取到则interrupt。

*  调用tryTerminate如果poolSize=0，workQueue也为空，则把线程池状态置为TERMINATED。

2.调用shutDownNow方法，有返回

*  置线程池状态为STOP,立即调用workers中所有执行线程的thread.interrupt方法
 *  copy workQueue中任务
 *  调用tryTerminate 如果poolSize=0，workQueue也为空，则把线程池状态置为TERMINATED。

可以看出来这二个方法的区别是：  
shutDownNow 直接调用thread.interrupt，不会等正在执行中的任务线程执行结束，返回任务队列中的任务。  
shutDown 等待执行中的线程执行结束，没有返回值。

### 2.BlockingQueue ###

队列是作为线程池暂存提交任务的地方，先来看下队列这块的整体结构：

![20180317225556222][]

从整体上看ConcurrentLinkedQueue的功能相对较少，只具备Queue接口、abstractQueue类的功能。其它6类在此之上还具备blockingQueue或/blockingDeque接口的功能。

#### 先看初始化线程池时，最常用的LinkedBlockingQueue。 ####

LinkedBlockingQueue阻塞队列大小的配置是可选的，如果我们初始化时指定一个大小，它就是有边界的，如果不指定，它就是无边界的。说是无边界，其实是采用了默认大小为Integer.MAX\_VALUE（231-1）的容量 。它的内部实现是一个链表。

上面走读线程池源码的时候说到，如果当前线程数大于corePoolSize并且线程池状态是RUNNING，就会添加到阻塞任务队列中，调用了workQueue.offer(command)。

public boolean offer(E e) {
            if (e == null) throw new NullPointerException();
            //count表示当前队列中元素个数，AtomicInteger类型确保并发条件下的更新操作是原子的。
            final AtomicInteger count = this.count;
            //如果队列满了，直接返回false
            if (count.get() == capacity)
                return false;
            int c = -1;
            final ReentrantLock putLock = this.putLock;
            //如果队列没有满，则获取该队列putLock锁
            putLock.lock();
            try {
                if (count.get() < capacity) {
                    enqueue(e);
                    c = count.getAndIncrement();
                    if (c + 1 < capacity)
                        //如果放入后，队列没有满，则唤醒notFull条件等待线程。
                        notFull.signal();
                }
            } finally {
                putLock.unlock();
            }
            if (c == 0)
                signalNotEmpty();
            return c >= 0;
        }
        ```
    
    当worker在执行时，线程池状态state == SHUTDOWN 或者 poolSize > corePoolSize 或者allowCoreThreadTimeOut有设置时， 线程池获取任务队列都会调用poll() : 
    
    ```java
    public E poll() {
            final AtomicInteger count = this.count;
            if (count.get() == 0)
                return null;
            E x = null;
            int c = -1;
            final ReentrantLock takeLock = this.takeLock;
            takeLock.lock();
            try {
                if (count.get() > 0) {
                    //获取队列首项，并从队列中移除
                    x = dequeue();
                    c = count.getAndDecrement();
                    if (c > 1)
                        //移除后，队列非空，则唤醒notEmpty条件等待线程
                        notEmpty.signal();
                }
            } finally {
                takeLock.unlock();
            }
            if (c == capacity)
                signalNotFull();
            return x;
        }
        
        //获取队列首项，并从队列中移除
        private E dequeue() {
            // assert takeLock.isHeldByCurrentThread();
            Node<E> h = head;
            Node<E> first = h.next;
            h.next = h; // help GC
            head = first;
            E x = first.item;
            first.item = null;
            return x;
        }
        ```
    可以看出调用poll方法时，如果获取到锁，则不会堵塞，队列为空则直接返回null。
    
    所以线程池为 SHUTDOWN状态时，不会堵塞任务队列的获取，能速度结束线程执行。
    
    但是如果线程池状态为RUNNING切poolSize<=corePoolSize时，会调用阻塞队列的workQueue.take()获取任务：
    
    ```java
    public E take() throws InterruptedException {
            E x;
            int c = -1;
            final AtomicInteger count = this.count;
            final ReentrantLock takeLock = this.takeLock;
            //获取锁时，可中断等待
            takeLock.lockInterruptibly();
            try {   
                    // 防止虚假唤醒，所以用while
                    while (count.get() == 0) {
                        //队列为空时，等待notEmpty条件唤醒
                        notEmpty.await();
                    }
                //否则返回队首任务，并从队列中移除。
                x = dequeue();
                c = count.getAndDecrement();
                if (c > 1)
                    notEmpty.signal();
            } finally {
                takeLock.unlock();
            }
            if (c == capacity)
                signalNotFull();
            return x;
        }

可见看出，当队列任务为空时，take方法会堵塞当前任务线程，同时等待队列任务添加，这样就能保持当前线程不会结束从而被回收。

[20180311221826445]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/bb0b0b8ac67f415ba4adb5b9cf900ae4.png
[20180311221826800]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/cc57e46637644f9d8617213cfaf9652b.png
[20180311221826592]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/697a9afd4693449da84e7df9297ceae6.png
[20180311221826578]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/ae5506d05b174a27b5b3af4fec52fa05.png
[20180311221826542]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/0684e81f781b46f0b2c4d77dc26cda47.png
[20180311221826784]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/e325e373d2db4ad8abeec026bd39933c.png
[20180317225556222]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/2293a469ded24deba3671ffb13888c4b.png