Java并发编程：04-源码分析-线程池(ThreadPoolExecutor)如何实现Thread的复用

Love The Way You Lie 2023-06-13 14:25 37阅读 0赞

### 文章目录 ###

*   *  1. 了解一下Thread的生命周期
     *  2. 为什么创建线程会很消耗资源？
     *  3. ThreadPoolExecutor 部分源码剖析：
     *   *  3.1 编写测试代码
         *  3.2 进入execute(Runnable command)方法
         *  3.3 进入private boolean addWorker(Runnable firstTask, boolean core) 方法
         *  3.4 进入ThreadPoolExecutor.Worker类（是一个内部类）
         *  3.5 线程3.3中的步骤2的t.start()分析
         *  3.6 runWorker(Worker w)方法的分析
         *  3.7 getTask()方法的分析
     *  4. 图形流程版简要概括
     *  5. 最后的话

> 一直都听说创建线程很消耗资源，我们都知道，因此都听大佬说用线程池呀，那么线程池是如何工作的呢？又是如何实现线程的复用的呢？我花了好几个小时看了几十遍源码，终于看到一点点精髓了。。。

这里不讨论线程池的基本用法以及注意实现,主要是分析线程的复用

## 1. 了解一下Thread的生命周期 ##

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

> 注意点: 如果线程运行结束，那就会进入终止状态，进而被回收，我们就没办法控制了  
> 因此，如果想要复用一个线程（执行多个任务() -> \{\} ），那必须保证线程一直在运行和阻塞状态

## 2. 为什么创建线程会很消耗资源？ ##

Java线程的创建成本很高，因为需要进行大量的工作：

*  必须为线程堆栈分配和初始化大量内存块。
 *  需要进行系统调用，以便在主机OS中创建 / 注册本机线程。
 *  描述符需要创建、初始化并添加到JVM内部数据结构中。  
    从某种意义上说，线程绑定资源的代价也很高；例如，线程堆栈、从堆栈可以访问的任何对象、JVM线程描述符、OS本机线程描述符。

然后我们看到简单看一下Thead的启动线程的源码

public synchronized void start() { 
           	// 省略部分源码, 精简如下：
            try { 
                start0();
            } finally { 
            }
        }
    
    // native 本地方法 start0
    private native void start0();

> 发现 start() 方法, 内部调用的是start0方法 【native 修饰的本地方法】，也就是并不是用Java代码实现的，它们来源于本地库的实现，底层就不在本次深究范围了

--------------------

在了解到这些之后，那可以发挥想象力一下，如果是你，你会如何设计呢？

1.  Thread 的 start() 不能随便调用，调用意味着资源的消耗
2.  一直运行感觉也不太对，没有任务需要运行，总不能一直占用CPU资源
3.  那就使用阻塞？一想到阻塞，首先想到阻塞队列，可以设想一下，把所有的任务都往阻塞队列里塞，因此就可以完美实现线程一直保持运行和阻塞之间的任务调度了。。。

基本实现思路是：  
自己的一个Runnable.run()【这里指的是Thead】，循环在跑，跑的过程中不断检查我们是否有新加入的子Runnable对象，有就调一下我们的run()，其实就一个大run()把其它小run()\#1,run()\#2,…给串联起来了

**下面是伪代码实现**

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

## 3. ThreadPoolExecutor 部分源码剖析： ##

### 3.1 编写测试代码 ###

ExecutorService executorService = Executors.newFixedThreadPool(1);
    executorService.execute(() -> { 
       System.out.println("执行开始:" + System.currentTimeMillis());
    });

### 3.2 进入execute(Runnable command)方法 ###

**进入实现类: java.util.concurrent.ThreadPoolExecutor.execute(Runnable command) \{\}**  
源码如下:

public void execute(Runnable command) { 
            if (command == null)
                throw new NullPointerException();
            /* * execute() 有三个步骤: * 1. 如果正在运行的线程少于corePoolSize线程，尝试使用给定命令作为其第一个任务来启动一个新线程。 * 【说明: 比如说: corePoolSize = 10, 哪怕目前线程池有空闲线程,也一样会启动新的线程执行任务, * 直到 线程池数量 corePoolSize = 10 才会启用复用线程的操作】 * * 2. 如果任务可以成功排队，那么仍然需要再次检查是否应该添加线程（因为现有线程自上次检查后就死掉了）， * 或者自进入此方法以来该池已关闭。因此，重新检查状态，并在必要时回滚排队， * 如果停止，或者如果没有线程，则启动一个新线程。 * * 3. 如果我们无法将任务排队，则尝试添加一个新的线程。如果失败，我们知道我们已关闭或处于饱和状态，因此拒绝该任务。 * 【说明：创建线程池的时候可以执行拒接策略的，默认是抛异常】 */
            int c = ctl.get();
            if (workerCountOf(c) < corePoolSize) { 
                // 这里走的步骤 1 
                // --> 进入 addWorker()
                if (addWorker(command, true))
                    return;
                c = ctl.get();
            }
            if (isRunning(c) && workQueue.offer(command)) { 
                // 这里走的步骤 2, 这个方法需要仔细拜读, 在这里很有学问
                // 使用了double check 的形式校验
                // ----------------------------------------------------------
                // 这里先简单理解一下:
                // 如果当前线程数 > corePoolSize, 那就会走步骤2,尝试添加到待执行的阻塞队列 (workQueue) 中
                int recheck = ctl.get();
                if (! isRunning(recheck) && remove(command))
                    reject(command);
                else if (workerCountOf(recheck) == 0)
                    addWorker(null, false);
            }
            else if (!addWorker(command, false))
                // 这里走的步骤 3
                // 执行拒接策略
                reject(command);
        }

### 3.3 进入private boolean addWorker(Runnable firstTask, boolean core) 方法 ###

**本类中的方法 addWorker()**  
这个方法：主要有两个很重要的步骤：

1.  创建 Worker 对象 **new Worker(firstTask)**
2.  启动线程 **t.start();**

源码如下(简略, 删除局部不影响深究的代码):

private boolean addWorker(Runnable firstTask, boolean core) { 
            // ...
            
            Worker w = null;
            try { 
    	        // 步骤1: 创建 Worker, 这个 Worker 是一个重点的对象, 它本身就是一个线程载体 (implements Runnable)
    	        // Worker 下面再源码分析, 这里记住他是一个 Runnable 实现类, 内部有两个重要变量 1. thread(Thread) 2. firstTask(Runnable)
                w = new Worker(firstTask);
                final Thread t = w.thread;
                if (t != null) { 
                    // ...
                    if (workerAdded) { 
    	                // 步骤2: 首次调用 start() 方法, 需要特别注意!!!
    	                // 这里需要引起注意, 否则你都不知道 t 是一个什么东西, 执行的是哪里的是哪一个类的run()代码
                        t.start();
                        workerStarted = true;
                    }
                }
            } finally { 
            	// ... 
            }
            return workerStarted;
        }

new Worker(firstTask), 传递一个 Runnable 实例进行构建 Worker ，查看 Worker 源码

### 3.4 进入ThreadPoolExecutor.Worker类（是一个内部类） ###

**先看源码部分核心源码**

private final class Worker
            extends AbstractQueuedSynchronizer
            implements Runnable
        { 
        // ...
        /** * 构造器， firstTask 实际就是测试代码中的 () -> { } 对象 */
        Worker(Runnable firstTask) { 
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            // 下面这行代码, getThreadFactory(),如果创建线程池的时候没有指定 threadFactory 的话, 
            // 那就默认是返回的实例是 Executors.DefaultThreadFactory 的静态内部类
            // 
            // newThread 方法传递 this 进入, 然后返回一个Thread 实例, 然后保存在this.thread 变量中
            // 这里很巧妙...也就是说3.3中的步骤2 t.start() 实际上调用的就是 Worker 中的 run() 方法, 
            // run() 又调用 runWorker(this)方法
            // 【runWorker()方法是ThreadPoolExecutor中的一个final方法, runWorker(Worker w) 】
            this.thread = getThreadFactory().newThread(this);
        }
    
        /** * 这个方法太重要了,这里执行的代码先记住, 后面再说,很重要很重要很重要 * 【特别注意:由于Worker是一个implements Runnable的实现类， 因此run()就是start()实际执行的内容,后面具体说 * 我在这里被糊弄了很久, 因为源码没有标注@Override, 搞得我当时都忘记了 Worker 实际上就是一个线程载体, * 一直在思考,哪里启动的线程... * 】 */
        public void run() { 
            runWorker(this);
        }
    	// ...
    }

### 3.5 线程3.3中的步骤2的t.start()分析 ###

1.  Thread t = w.thread; 这个t是Worker中的一个类变量，
2.  t.start() 调用的方法是Worker中的run() 【因为通过Worker构造参数中newThead(this), 传递的Runnable是Worker自身】
3.  Worker的run()方法调用的是 ThreadPoolExecutor类中的 runWorker 方法

### 3.6 runWorker(Worker w)方法的分析 ###

先贴出源码(删减不必要的部分代码), 然后分析:

final void runWorker(Worker w) { 
    	/** * 这个方法的代码很巧妙,需要仔细拜读 */
        Thread wt = Thread.currentThread();
        // 首先 获取 Worker 中的 firstTask 这个变量, 赋值给局部变量 task , task就是贯穿这个线程的执行者
        Runnable task = w.firstTask;
        // 这里 w.firstTask = null; 主要是用于gc作用
        w.firstTask = null;
        try { 
    		// 这里使用 while 循环
            while (task != null || (task = getTask()) != null) { 
            	// 首先 第一次进来 task 肯定是不为null, 因此 || 后面的代码不用执行,
            	// 接着调用 task.run(); 方法, 这里才是我们实际线程需要执行的代码,(也就是我们测试代码中 => System.out.println("执行开始:" + System.currentTimeMillis());)
            	// finally 之后, 会把 task 置为null, 
            	// 再次 (第二次以后) 进入while 循环, 执行判断 "task != null" , 返回false, 因此接着执行 task = getTask() 方法【下面详细分析】
            	// 这个 getTask() 方法是一个返回存储在 workQueue 阻塞队列中的待执行对象,
            	// workQueue 是干嘛的呢? 其实回到3.2进入execute(Runnable command)方法中的步骤2, 
            	// 如果当前线程数 > corePoolSize, 那就会走3.2步骤2,尝试添加到待执行的阻塞队列 (workQueue) 中
            	// 所以这里 getTask() 就是获取了 execute() 的时候尚未执行的对象出来执行,
            	// 然后直接调用 task.run(); 不是调用start()方法, 从而实现了线程的复用, 
            	// 因为 runWorker 方法中一直都是调用的 Worker 中的 Thread 中的run(), 
            	// 然后在 run() 中获取阻塞阻塞需要执行的进行直接调用run()方法
                try { 
                    beforeExecute(wt, task);
                    Throwable thrown = null;
                    try { 
                    	// 这里是直接调用run()方法, 从而实现了线程的复用
                        task.run();
                    } catch (Throwable x) { 
                        thrown = x; throw new Error(x);
                    } finally { 
                        afterExecute(task, thrown);
                    }
                } finally { 
                    task = null;
                }
            }
        } finally { 
            processWorkerExit(w, completedAbruptly);
        }
    }

### 3.7 getTask()方法的分析 ###

java.util.concurrent.ThreadPoolExecutor.getTask()  
这个方法主要是作用有如下几点:

> 根据当前配置设置执行阻塞或定时等待任务，如果由于以下3中情况任何原因而必须退出此工作程序，则返回null
> 
> 1.  超过了maximumPoolSize（由于对setMaximumPoolSize的调用）
> 2.  池已停止
> 3.  池已关闭，队列为空。
> 4.  该工作程序等待任务超时，并且超时工作程序会终止（即 \{@code allowCoreThreadTimeOut || workerCount> corePoolSize\}）在定时等待之前和之后，以及队列是否为非空，此worker不是池中的最后一个线程。  
>     …
> 
> 以上4中情况都会返回 null, 如果返回null，那么 runWorker中的while循环中就会退出，然后这个线程就会完成所有任务，从而被销毁掉

private Runnable getTask() { 
       boolean timedOut = false; // Did the last poll() time out?
    
        for (;;) { 
            int c = ctl.get();
            int rs = runStateOf(c);
    
            // 检查 池是否已停止，workQueue是否为null
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { 
                decrementWorkerCount();
                return null;
            }
    
            int wc = workerCountOf(c);
    
            // 检查是否超时等原因， 如果添加创建线程池的时候设置超时的话， 有可能会导致为执行的线程超时，因此也会返回null
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
    
            if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) { 
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }
    
            try { 
            	// 三木运算 : 
            	// 是否设置了超时?
                Runnable r = timed ?
                	// 如果是超时, 那就使用 阻塞队列中的 poll 方法出队, 这里有可能是返回null (这个是一个阻塞方法)
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    // 如果没有超时, 直接使用 take() 方法出队, 从而实现了线程在运行状态和阻塞状态之间动态调度的主要手段 (这个是一个阻塞方法)
                    workQueue.take();
                if (r != null)
                    return r;
                // 如果 r = null, 那就说明超时了, 继续重试, 因为整体代码是 for (;;) {} 包裹着的
                timedOut = true;
            } catch (InterruptedException retry) { 
                timedOut = false;
            }
        }
    }

## 4. 图形流程版简要概括 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppbmdsb25nU291cmNl_size_16_color_FFFFFF_t_70 2]

## 5. 最后的话 ##

*  本次分析的源码使用的是jdk1.8
 *  整体的复用流程大概如上，当然源码中还有很多重要的点，但这里主要是分析如何实现Thread的复用流程，如果有时间我还是很有兴趣接着分析。
 *  能力有限，可能分析的过程中也有很多不足之处，欢迎指出更正，谢谢。

> 借阅文章  
> https://zhidao.baidu.com/question/331304486472362365.html  
> https://blog.csdn.net/hellozhxy/article/details/90038127  
> https://www.jianshu.com/p/cb0f2dc087e3

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppbmdsb25nU291cmNl_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191117190016209.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppbmdsb25nU291cmNl,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppbmdsb25nU291cmNl_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20191118223047581.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppbmdsb25nU291cmNl,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppbmdsb25nU291cmNl_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20191118093117533.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppbmdsb25nU291cmNl,size_16,color_FFFFFF,t_70