❤️并发编程❤️线程池源码分析以及线程复用的原理

叁歲伎倆 2022-08-28 07:50 191阅读 0赞

> ***【并发编程系列】***  
> [❤️并发编程❤️创建线程的四种方式 线程通信  
> ][Link 1]  
> [❤️并发编程❤️一万字线程生命周期和状态转换知识梳理  
> ][Link 2]  
> [❤️并发编程❤️Java内存模型  
> ][Java_]  
> [❤️并发编程❤️重排序与happens-before  
> ][happens-before]  
> [❤️并发编程❤️显式锁Lock和内置锁知识整理  
> ][Lock_]  
> [❤️并发编程❤️如何正确停止线程？  
> ][Link 3]  
> [❤️并发编程❤️生产者消费者模式实现的三种方式  
> ][Link 4]  
> [❤️并发编程❤️线程安全问题分析和场景总结  
> ][Link 5]  
> [❤️并发编程❤️线程池源码分析以及线程复用的原理  
> ][Link 6]

### 文章目录 ###

*   *   *  （1）execute核心代码分析
         *  （2）addWorker
         *  （3）runWorker
         *  （4）getTask

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

### （1）execute核心代码分析 ###

java.util.concurrent.ThreadPoolExecutor\#execute

public void execute(Runnable command) { 
            if (command == null)
                throw new NullPointerException();
    		//获取线程池状态，使用int表示
            int c = ctl.get();
    		//如果线程数小于核心线程数
            if (workerCountOf(c) < corePoolSize) { 
    			//则创建线程并执行当前任务,第二个参数表示是否创建核心线程，当前为true
                if (addWorker(command, true))
    				//如果创建成功则直接返回
                    return;
    				//再次检查线程池状态
                c = ctl.get();
            }
    		//如果线程池运行中，则添加任务到工作队列中
            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);
        }

### （2）addWorker ###

java.util.concurrent.ThreadPoolExecutor\#addWorker

检查是否可以根据当前池状态和给定边界（核心或最大值）添加新的工作线程。如果是这样，将相应地调整工作线程数量，并且将创建并启动一个新的工作线程，并将firstTask作为其第一个任务运行。如果线程池停止运行、线程池将要关闭或则线程工厂创建一个线程失败，那么此方法将返回false。

如果线程创建失败，或因为线程工厂返回Null，或者因为在线程启动中发生异常，会回滚此方法执行操作。

【参数：firstTask】  
新线程首先运行的任务。当此时线程数小于核心线程数时会创建一个Workers线程。

【参数：core】  
如果该参数为true则说明是核心线程，false表示最大线程数

ThreadPoolExecutor在创建线程时，会将线程封装成工作线程worker,并放入工作线程组中，然后这个worker反复从阻塞队列中拿任务去执行  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAWldaaGFuZ1l1_size_20_color_FFFFFF_t_70_g_se_x_16 1]  
我们将这个方法拆分下，分成俩个模块，如上图所示模块A和模块B.

for (;;) { 
                int c = ctl.get();
                int rs = runStateOf(c);
    
                // 检查线程池状态，检查队列是否为空
                if (rs >= SHUTDOWN &&
                    ! (rs == SHUTDOWN &&
                       firstTask == null &&
                       ! workQueue.isEmpty()))
                    return false;
                 
                for (;;) { 
    				//获取线程数量
                    int wc = workerCountOf(c);
    			    // 1.如果core是ture,证明需要创建的线程为核心线程，则先判断当前线程是否大于核心线程
                    // 如果core是false,证明需要创建的是非核心线程，则先判断当前线程数是否大于总线程数
                    // 如果不小于，则返回false
                    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
                }
            }

模块A主要是判断线程数量是否超出阈值，超过了就返回false.

boolean workerStarted = false;
            boolean workerAdded = false;
            Worker w = null;
            try { 
    			 //创建一个worker对象
    			 //该构造函数会 getThreadFactory().newThread(this);线程工厂创建一个线程，并封装为Worker
                w = new Worker(firstTask);
    			//实例化一个Thread对象
                final Thread t = w.thread;
                if (t != null) { 
    				//线程池全局锁
                    final ReentrantLock mainLock = this.mainLock;
                    mainLock.lock();
                    try { 
                        
    					//同上，检查线程池状态
                        int rs = runStateOf(ctl.get());
                        if (rs < SHUTDOWN ||
                            (rs == SHUTDOWN && firstTask == null)) { 
                            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;

Worker类实现了Runnable接口，所以Worker也是一个线程任务。在构造方法中，创建了一个线程，线程的任务就是自己。故addWorker方法调用addWorker方法源码下半部分中的t.start，会触发Worker类的run方法被JVM调用。

### （3）runWorker ###

java.util.concurrent.ThreadPoolExecutor\#runWorker

final void runWorker(Worker w) { 
            Thread wt = Thread.currentThread();
            Runnable task = w.firstTask;
            w.firstTask = null;
    		//线程启动之后，通过unlock方法释放锁
            w.unlock();
            boolean completedAbruptly = true;
            try { 
    			//Worker执行firstTask或从workQueue中获取任务，如果getTask方法不返回null,循环不退出
                while (task != null || (task = getTask()) != null) { 
                    w.lock();
    				//检查线程池状态，倘若线程池处于中断状态，当前线程将中断
                    if ((runStateAtLeast(ctl.get(), STOP) ||
                         (Thread.interrupted() &&
                          runStateAtLeast(ctl.get(), STOP))) &&
                        !wt.isInterrupted())
                        wt.interrupt();
                    try { 
    					//执行beforeExecute 
                        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方法 
                            afterExecute(task, thrown);
                        }
                    } finally { 
                        task = null;
                        w.completedTasks++;
    					//解锁操作
                        w.unlock();
                    }
                }
                completedAbruptly = false;
            } finally { 
                processWorkerExit(w, completedAbruptly);
            }
        }

首先去执行创建这个worker时就有的任务，当执行完这个任务后，worker的生命周期并没有结束，在while循环中，worker会不断地调用getTask方法从阻塞队列中获取任务然后调用task.run()执行任务,从而达到复用线程的目的。只要getTask方法不返回null,此线程就不会退出。【注意这一块也是线程复用最核心的地方】

当然，核心线程池中创建的线程想要拿到阻塞队列中的任务，先要判断线程池的状态，如果STOP或者TERMINATED，返回null。

### （4）getTask ###

java.util.concurrent.ThreadPoolExecutor\#getTask

// Worker.getTask方法源码
    private Runnable getTask() { 
        boolean timedOut = false; // Did the last poll() time out?
        for (;;) { 
            int c = ctl.get();
            int rs = runStateOf(c);
            // Check if queue empty only if necessary.
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { 
                decrementWorkerCount();
                return null;
            }
            int wc = workerCountOf(c);
            // Are workers subject to culling?
            // 1.allowCoreThreadTimeOut变量默认是false,核心线程即使空闲也不会被销毁
            // 如果为true,核心线程在keepAliveTime内仍空闲则会被销毁。 
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
            // 2.如果运行线程数超过了最大线程数，但是缓存队列已经空了，这时递减worker数量。 
    　　　　 // 如果有设置允许线程超时或者线程数量超过了核心线程数量，
            // 并且线程在规定时间内均未poll到任务且队列为空则递减worker数量
            if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) { 
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }
            try { 
                // 3.如果timed为true(想想哪些情况下timed为true),则会调用workQueue的poll方法获取任务.
                // 超时时间是keepAliveTime。如果超过keepAliveTime时长，
                // poll返回了null，上边提到的while循序就会退出，线程也就执行完了。
                // 如果timed为false（allowCoreThreadTimeOut为falsefalse
                // 且wc > corePoolSize为false），则会调用workQueue的take方法阻塞在当前。
                // 队列中有任务加入时，线程被唤醒，take方法返回任务，并执行。
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) { 
                timedOut = false;
            }
        }
    }

核心线程的会一直卡在workQueue.take方法，被阻塞并挂起，不会占用CPU资源，直到拿到Runnable 然后返回（当然如果allowCoreThreadTimeOut设置为true,那么核心线程就会去调用poll方法，因为poll可能会返回null,所以这时候核心线程满足超时条件也会被销毁）。

非核心线程会workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) ，如果超时还没有拿到，下一次循环判断compareAndDecrementWorkerCount就会返回null,Worker对象的run()方法循环体的判断为null,任务结束，然后线程被系统回收 。

[Link 1]: https://blog.csdn.net/Octopus21/article/details/120106529
[Link 2]: https://blog.csdn.net/Octopus21/article/details/120589643
[Java_]: https://blog.csdn.net/Octopus21/article/details/120120576
[happens-before]: https://blog.csdn.net/Octopus21/article/details/120120611
[Lock_]: https://blog.csdn.net/Octopus21/article/details/120468606
[Link 3]: https://blog.csdn.net/Octopus21/article/details/120584190
[Link 4]: https://blog.csdn.net/Octopus21/article/details/120591757
[Link 5]: https://blog.csdn.net/Octopus21/article/details/120596474
[Link 6]: https://blog.csdn.net/Octopus21/article/details/120608492
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAWldaaGFuZ1l1_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/6230b1374f11455190e3be1680c15030.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAWldaaGFuZ1l1_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220828/19113103ae544a0bad36fbb8453dc392.png