源码解析线程池的执行流程

素颜马尾好姑娘i 2021-09-28 13:50 383阅读 0赞

在阅读此文章之前，应先了解线程池的基本知识，详情可以看我上一篇博客：[初识线程池][Link 1]

# 一、执行流程图 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxMzg0MzUx_size_16_color_FFFFFF_t_70]  
任务被提交到线程池，会先判断当前线程数量是否小于corePoolSize，如果小于则创建线程来执行提交的任务，否则将任务放入workQueue队列，如果workQueue满了，则判断当前线程数量是否小于maximumPoolSize,如果小于则创建线程执行任务，否则就会调用handler拒绝策略，以表示线程池拒绝接收任务  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxMzg0MzUx_size_16_color_FFFFFF_t_70 1]

# 二、源码解析 （jdk1.7） #

下来看一下，上述的流程在源码中是如何体现的，ThreadPoolExecutor类中的内容在上一个博客中已经解释，下面直接开始上代码：

用户提交线程的方式有两种：submit（）和execute（）。

*  submit（）方法主要是继承了AbstractExecutorService类中的三个方法，从源码很清晰的可以看到这个方法是含有返回值的，而且三种方法的参数列表都不一样，Runnable和Callable两种任务类型（[两种类型的区别？][Link 2]）。而且每个submit方法内部最终都调用了execute（）方法

public Future<?> submit(Runnable task) { 
            if (task == null) throw new NullPointerException();
            RunnableFuture<Void> ftask = newTaskFor(task, null);
            execute(ftask);
            return ftask;
        }
        
        public <T> Future<T> submit(Runnable task, T result) { 
            if (task == null) throw new NullPointerException();
            RunnableFuture<T> ftask = newTaskFor(task, result);
            execute(ftask);
            return ftask;
        }
        
        public <T> Future<T> submit(Callable<T> task) { 
            if (task == null) throw new NullPointerException();
            RunnableFuture<T> ftask = newTaskFor(task);
            execute(ftask);
            return ftask;
        }

*  execute（）方法。既然上面说了，submit方法最终都调用了这个方法，那么这个方法应该就是线程池提交流程中最基础的方法。

public void execute(Runnable command) { 
           //相当于参数校验，判断当前的任务是否为空
            if (command == null)
                throw new NullPointerException();
                    
           //获得当前任务的状态 
            int c = ctl.get();
    		//如果当前工作的线程数<corePoolSize核心线程数
            if (workerCountOf(c) < corePoolSize) { 
    			//调用addWorker创建线程，执行任务
    			if (addWorker(command, true))
                    return;
    		//创建线程失败：①线程池处于running之后的状态②超出核心或者最大线程数
                c = ctl.get();  //再次获取
            }  
    		//如果当前工作的线程数>corePoolSize
    		//线程池处于Running状态，并且将任务添加到workQueue队列。
            if (isRunning(c) && workQueue.offer(command)) { 
                //重新获取一次线程池的状态
                int recheck = ctl.get();
                //如果当前状态是非运行状态，则删除当前任务
                if (! isRunning(recheck) && remove(command))
                   //执行拒绝策略，不在接受新的任务提交
    			   reject(command);
    			//如果还处于running状态，并且工作线程数为0
                else if (workerCountOf(recheck) == 0)
                    addWorker(null, false);//创建新的线程
            }
    		//如果放入workQueue失败，则创建线程执行任务，
    		//如果这时创建线程失败(当前线程数>maximumPoolSize时)，
    		//就会调用reject拒绝策略(内部调用handler)拒绝接受任务
            else if (!addWorker(command, false))
                reject(command);
        }

*  addWorker（）方法

private boolean addWorker(Runnable firstTask, boolean core) { 
            retry:  //重复尝试
    		//外层for循环检查状态
            for (;;) {   
                int c = ctl.get(); //获得当前的状态表示的整型数
                int rs = runStateOf(c);  //通过整型数获得当前线程状态
    
                // Check if queue empty only if necessary.
                //如果线程池已关闭 并满足，不创建新的worker
    			if (rs >= SHUTDOWN &&
                    ! (rs == SHUTDOWN &&
                       firstTask == null &&
                       ! workQueue.isEmpty()))
                    return false; //添加失败
    
              //内层for循环检查数量
                for (;;) { 
                    int wc = workerCountOf(c);
    			    //创建失败：wc>core时和核心数比较，否则和最大线程数比较
    			    if (wc >= CAPACITY ||
                        wc >= (core ? corePoolSize : maximumPoolSize))
                        return false;  //添加失败
                    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;  //worker是否启动
            boolean workerAdded = false;   //是否添加成功
            Worker w = null;
            try { 
                final ReentrantLock mainLock = this.mainLock; //定义一个可重入锁
                w = new Worker(firstTask);  //创建Worker对象，将firstTask传给它
                final Thread t = w.thread;//实例化一个Thread对象
                if (t != null) { 
                    mainLock.lock();//加锁
                    try { 
                        // Recheck while holding lock.
                        // Back out on ThreadFactory failure or if
                        // shut down before lock acquired.
                        int c = ctl.get();
                        int rs = runStateOf(c);
                        
                        //rs处于Running状态，（rs是shutdown状态并且firstTask为空）
                        if (rs < SHUTDOWN ||
                            (rs == SHUTDOWN && firstTask == null)) { 
                            if (t.isAlive()) // precheck that t is startable
                                throw new IllegalThreadStateException();
                           //将当前的线程加入到工作的hashset中
    						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;
        }

*  addWorkerFailed（）方法

private void addWorkerFailed(Worker w) { 
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try { 
                if (w != null)  //如果当前工作为空,说明启动线程失败
                    workers.remove(w); // 直接删掉这个工作
                decrementWorkerCount();  //workerCount-1
                tryTerminate();
            } finally { 
                mainLock.unlock();
            }
        }

[Link 1]: https://blog.csdn.net/qq_41384351/article/details/89916051
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxMzg0MzUx_size_16_color_FFFFFF_t_70]: /images/20210923/a6cd11d330014b25941ec03dc451b4a9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxMzg0MzUx_size_16_color_FFFFFF_t_70 1]: /images/20210923/3044897a13a446208c754e541f71a9d0.png
[Link 2]: https://blog.csdn.net/qq_41384351/article/details/89991156