线程池原理总结

女爷i 2023-06-18 02:57 7阅读 0赞

**【引言】**

关于线程池，印象中好像自己有写过相关的总结博客。翻了翻之前的博客，确实，在去年十一月写过一篇[《线程池使用总结》][Link 1]。

时隔一年，我已经离开了那家让我成长很多的公司，在那里，写了很多的代码，学了很多的技术。而现在的我，每天在公司都要面试招人，让我又有了加深理论学习的机会。所以，本篇博客内容总结线程池的原理。

#### 线程池处理流程 ####

当向线程池提交一个任务之后，线程池是如何处理这个任务的呢？线程池的处理流程如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwMzQyMjM_size_16_color_FFFFFF_t_70]

1.  线程池判断核心线程池里的线程是否都在执行任务。如果不是，则创建一个新的工作线程来执行任务。如果核心线程池里的线程都在执行任务，则进入下一个流程。
2.  线程池判断工作队列是否已经满。如果工作队列没有满，则将新提交的任务存储在这个工作队列里。如果工作队列满了，则进入下个流程。
3.  线程池判断线程池里的线程是否都处于工作状态。如果没有，则创建一个新的线程来执行此任务。如果已经满了，则交给饱和策略来处理任务。

在ThreadPoolExecutor类中，最核心的任务提交方法是execute()方法，执行示意图如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwMzQyMjM_size_16_color_FFFFFF_t_70 1]

1.  如果当前运行的线程少于corePoolSize，则创建新线程来执行任务（这一过程，需要获取全局锁）
2.  如果运行的线程等于或多于corePoolSize，则将任务假如BlockingQueue。
3.  如果无法加入BlockQueue（任务队列已满），则创建新的线程来处理任务（这一过程，需要获取全局锁）
4.  如果创建新线程将使当前运行的线程超过maxmumPoolSize，任务将被拒绝，并调用RejectedExecutionHandler.rejectedExecution()方法。

#### 线程池核心代码 ####

**1. execute方法**

public void execute(Runnable command) {
    //判断提交的任务command是否为null，若是null，则抛出空指针异常；
        if (command == null)
            throw new NullPointerException();
    //如果线程数小于基本线程数，则创建线程并执行当前任务
        if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
    //如果线程数大于基本线程数或线程创建失败，则将当前任务放到工作队列中
            if (runState == RUNNING && workQueue.offer(command)) {
                if (runState != RUNNING || poolSize == 0)
                    ensureQueuedTaskHandled(command);
            }
    //如果线程池不处于运行中或任务无法放入队列，并且当前线程数量小于最大允许的线程数量，则创建一个线程执行任务
            else if (!addIfUnderMaximumPoolSize(command))
                reject(command); // is shutdown or saturated
        }
    }

**2. addIfUnderCorePoolSize方法**

//当低于核心池大小时执行的方法
    private boolean addIfUnderCorePoolSize(Runnable firstTask) {
        Thread t = null;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (poolSize < corePoolSize && runState == RUNNING)
                t = addThread(firstTask);      //创建线程去执行firstTask任务   
            } finally {
            mainLock.unlock();
        }
    //判断t是否为空，为空则表明创建线程失败（即poolSize>=corePoolSize或者runState不等于RUNNING），否则调用t.start()方法启动线程。
        if (t == null)
            return false;
        t.start();
        return true;
    }

**3. addThread方法**

//参数为提交的任务，返回值为Thread类型
    private Thread addThread(Runnable firstTask) {
    //创建一个Worker对象
        Worker w = new Worker(firstTask);
        Thread t = threadFactory.newThread(w);  //创建一个线程，执行任务   
        if (t != null) {
            w.thread = t;            //将创建的线程的引用赋值为w的成员变量       
            workers.add(w);
            int nt = ++poolSize;     //当前线程数加1       
            if (nt > largestPoolSize)
                largestPoolSize = nt;
        }
        return t;
    }

**4. Worker对象**

private final class Worker implements Runnable {
        private final ReentrantLock runLock = new ReentrantLock();
        private Runnable firstTask;
        volatile long completedTasks;
        Thread thread;
        Worker(Runnable firstTask) {
            this.firstTask = firstTask;
        }
        boolean isActive() {
            return runLock.isLocked();
        }
        void interruptIfIdle() {
            final ReentrantLock runLock = this.runLock;
            if (runLock.tryLock()) {
                try {
            if (thread != Thread.currentThread())
            thread.interrupt();
                } finally {
                    runLock.unlock();
                }
            }
        }
        void interruptNow() {
            thread.interrupt();
        }
     
        private void runTask(Runnable task) {
            final ReentrantLock runLock = this.runLock;
            runLock.lock();
            try {
                if (runState < STOP &&
                    Thread.interrupted() &&
                    runState >= STOP)
                boolean ran = false;
                beforeExecute(thread, task);   //beforeExecute方法是ThreadPoolExecutor类的一个方法，没有具体实现，用户可以根据
                //自己需要重载这个方法和后面的afterExecute方法来进行一些统计信息，比如某个任务的执行时间等           
                try {
                    task.run();
                    ran = true;
                    afterExecute(task, null);
                    ++completedTasks;
                } catch (RuntimeException ex) {
                    if (!ran)
                        afterExecute(task, ex);
                    throw ex;
                }
            } finally {
                runLock.unlock();
            }
        }
     
    //首先执行的是通过构造器传进来的任务firstTask，在调用runTask()执行完firstTask之后，在while循环里面不断通过getTask()去缓存队列中取新的任务来执行
        public void run() {
            try {
                Runnable task = firstTask;
                firstTask = null;
                while (task != null || (task = getTask()) != null) {
                    runTask(task);
                    task = null;
                }
            } finally {
                workerDone(this);   //当任务队列中没有任务时，进行清理工作       
            }
        }
    }

**5. ThreadPoolExecutor类中的getTask方法：**

Runnable getTask() {
        for (;;) {
            try {
                int state = runState;
    	//判断当前线程池状态，如果runState大于SHUTDOWN（即为STOP或者TERMINATED），则直接返回null
                if (state > SHUTDOWN)
                    return null;
                Runnable r;
    	//如果线程状态为SHUTDOWN，则从缓存队列中取任务
                if (state == SHUTDOWN)  // Help drain queue
                    r = workQueue.poll();
                else if (poolSize > corePoolSize || allowCoreThreadTimeOut) //如果线程数大于核心池大小或者允许为核心池线程设置空闲时间，
                    //则通过poll取任务，若等待一定的时间取不到任务，则返回null
                    r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
                else
                    r = workQueue.take();
                if (r != null)
                    return r;
                if (workerCanExit()) {    //如果没取到任务，即r为null，则判断当前的worker是否可以退出
                    if (runState >= SHUTDOWN) // Wake up others
                        interruptIdleWorkers();   //中断处于空闲状态的worker
                    return null;
                }
                // Else retry
            } catch (InterruptedException ie) {
                // On interruption, re-check runState
            }
        }
    }

**6. workerCanExit方法**

//如果runState大于等于STOP，或者任务缓存队列为空了，或者允许为核心池线程设置空闲存活时间并且线程池中的线程数目大于1，允许worker退出
    private boolean workerCanExit() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        boolean canExit;
        try {
            canExit = runState >= STOP ||
                workQueue.isEmpty() ||
                (allowCoreThreadTimeOut &&
                 poolSize > Math.max(1, corePoolSize));
        } finally {
            mainLock.unlock();
        }
        return canExit;
    }

**7. interruptldleWorkers方法**

//如果允许worker退出，则调用该方法中断处于空闲状态的worker
    void interruptIdleWorkers() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            for (Worker w : workers)  //实际上调用的是worker的interruptIfIdle()方法
                w.interruptIfIdle();
        } finally {
            mainLock.unlock();
        }
    }
    
    void interruptIfIdle() {
        final ReentrantLock runLock = this.runLock;
        if (runLock.tryLock()) {    //调用tryLock()来获取锁的，因为如果当前worker正在执行任务，锁已经被获取了，是无法获取到锁的
     //如果成功获取了锁，说明当前worker处于空闲状态
       try {
        if (thread != Thread.currentThread())  
        thread.interrupt();
            } finally {
                runLock.unlock();
            }
        }
    }

**8. addIfUnderMaximumPoolSize方法**

//线程池中的线程数达到了核心池大小并且往任务队列中添加任务失败的情况下执行
    private boolean addIfUnderMaximumPoolSize(Runnable firstTask) {
        Thread t = null;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (poolSize < maximumPoolSize && runState == RUNNING)
                t = addThread(firstTask);
        } finally {
            mainLock.unlock();
        }
        if (t == null)
            return false;
        t.start();
        return true;
    }

在上面多次提到的任务队列，即workQueue，用来存放等待执行的任务，类型为BlockQueue，在[《线程池使用总结》][Link 1]文章中有写到，这里就不再列举了，下面总结下饱和策略。

#### 线程池饱和策略： ####

当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumPoolSize，如果还有任务到来就会采取任务拒绝策略，通常有以下四种策略：

1.  ThreadPoolExecutor.AbortPolicy：**丢弃任务并抛出RejectedExecutionException异常。**
2.  ThreadPoolExecutor.DiscardPolicy：**丢弃任务，但不抛出异常**
3.  ThreadPoolExecutor.DiscardOldestPolicy：**丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）**
4.  ThreadPoolExecutor.CallerRunsPolicy：**由调用线程处理该任务**

**【总结】**

关于线程池的总结，使用+原理，这下全面了。

时隔一年，又写到了线程池，博客记录成长，也记录生活。

今天的朋友圈被帝都二零一九年的初雪刷屏了，这场雪，相比去年，是早了些，二零二零的脚步也越来越近了。

[Link 1]: https://blog.csdn.net/hu_zhiting/article/details/84147620
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwMzQyMjM_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/2019112919492712.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwMzQyMjM=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwMzQyMjM_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20191129194936924.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwMzQyMjM=,size_16,color_FFFFFF,t_70