ThreadPoolExecutor 核心方法源码解析

淩亂°似流年 2022-12-22 09:05 155阅读 0赞

# 1 ThreadPoolExecutor UML #

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

# 2 线程池的状态 #

ThreadPoolExecutor 使用 int 的高 3 位来表示线程池状态，低 29 位表示线程数量。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70_pic_center 1]

//左移动29位
    private static final int COUNT_BITS = Integer.SIZE - 3;
    
    //容量，1左移29位再减去1， 000 11111 11111111 11111111 11111111，及536870911
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
    
    /** *线程池状态 */
    
    //接受新任务，并处理队列任务， -1在Java底层是由32个1表示的，左移29位的话，值为111
    private static final int RUNNING    = -1 << COUNT_BITS;
    //不接受新任务，但会处理队列任务,0左移29位，值为000
    private static final int SHUTDOWN   =  0 << COUNT_BITS;
    //不接受新任务，不会处理队列任务,1左移29位，值为001
    private static final int STOP       =  1 << COUNT_BITS;
    //所有的任务已结束，中间状态，2左移29位，值为010
    private static final int TIDYING    =  2 << COUNT_BITS;
    //终结状态,3左移29位，值为011
    private static final int TERMINATED =  3 << COUNT_BITS;

# 3 构造方法 #

/** *1 核心线程数 *2 最大线程数 (空闲线程) *3 针对空闲线程的存活时间 如果超时了则把空闲的线程kill *4 针对3的时间单位 *5 任务存放的队列 *6 线程工厂 (创建线程,并且给线程起名字) *7 拒绝策略 */
    public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,
                                  RejectedExecutionHandler handler)

**工作流程:**

线程池中刚开始没有线程，当一个任务提交给线程池后，线程池会创建一个新线程来执行任务  
当线程数达到 核心线程数上限，这时再加入任务，新加的任务会被加入队列当中去  
前提是有界队列，任务超过了队列大小时，会创建 maximumPoolSize - corePoolSize 数目的线程数目作为空闲线程来执行任务  
如果线程到达 maximumPoolSize 仍然有新任务这时会执行拒绝策略

@Slf4j
    public class TestThreadExecutorPool { 
    
    
        public static void main(String[] args)  { 
    
    
            AtomicInteger atomicInteger = new AtomicInteger();
    
            /** * * 拒绝策略： * AbortPolicy 默认拒绝策略，当最大线程和阻塞队列满的时候不在接收新的任务 * DiscardPolicy 放弃策略，当最大线程和阻塞队列满的时丢弃新来的任务 * CallerRunsPolicy run策略，当最大线程和阻塞队列满的时调用线程直接执行run方法 * DiscardOldestPolicy 放弃阻塞队列策略，当最大线程和阻塞队列满的时丢弃阻塞队列的策略 * */
            ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 2, 3,
                    TimeUnit.SECONDS, new ArrayBlockingQueue<>(1), r -> new Thread(r, "rosh" + atomicInteger.incrementAndGet()),
                    new ThreadPoolExecutor.AbortPolicy());
    
            for (int i = 1; i <= 4; i++) { 
                threadPoolExecutor.execute(() -> { 
                    try { 
                        TimeUnit.SECONDS.sleep(3);
                        log.info(Thread.currentThread().getName() + " 执行完毕");
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                });
            }
    
    
        }
    
    
    }

![在这里插入图片描述][20201118225934372.png_pic_center]

# 4 Execute 源码解析 #

## 4.1 主流程 ##

public void execute(Runnable command) { 
            if (command == null)
                throw new NullPointerException();
        
            /* * 获取ctl值,ctl作用：高3位表示线程池状态，低29位表示线程池中线程数量。 */
            int c = ctl.get();
           
            //如果工作线程数目小于核心线程数
            if (workerCountOf(c) < corePoolSize) { 
                //添加工作线程
                if (addWorker(command, true))
                    return; 
                //添加失败，需要重新获取状态
                c = ctl.get();
            }
            //如果线程的状态是Running，那么尝试给workQueue添加任务
            if (isRunning(c) && workQueue.offer(command)) { 
                //添加的过程可能会阻塞，所以重新获取ctl状态值
                int recheck = ctl.get();
                //如果这时候线程池状态已经不是可运行状态了，则进行回滚的操作，并执行拒绝策略
                if (! isRunning(recheck) && remove(command))
                    reject(command);
                //如果还是运行状态但是线程池中线程数目为0，那么创建一个空闲线程
                else if (workerCountOf(recheck) == 0)
                    addWorker(null, false);
            }
        	//如果工作队列满了，就往最大线程池数中创建线程执行任务,如果添加失败，那么执行拒绝策略
            else if (!addWorker(command, false))
                reject(command);
        }

## 4.2 addWorker (添加线程) ##

private boolean addWorker(Runnable firstTask, boolean core) { 
            retry:
            for (;;) { 
                //获取ctl的值
                int c = ctl.get();
                //获取线程状态
                int rs = runStateOf(c);
    
                /** *判断线程池状态是否是可以运行的，SHUTDOWN = 0, RUNNING < 0，所以如果rs >= SHUTDOWN就代表了不是可运行状态。 *如果线程状态是不可执行的, 那么满足这3个状态(rs == SHUTDOWN && firstTask == null && * !workQueue.isEmpty())，表示线程池在SHUTDOWN状态可以执行workQueue剩余的任务 */
                if (rs >= SHUTDOWN &&
                    ! (rs == SHUTDOWN &&
                       firstTask == null &&
                       ! workQueue.isEmpty()))
                    return false;
                  
                //自旋
                for (;;) { 
                    //获取线程池数量
                    int wc = workerCountOf(c);
                    //如果是核心线程，判断当前的线程数已经达到最大值或大于corePoolSize，如果大于，线程池满了，直接返回
                    //如果不是核心线程，判断当前线程数量是否大于最大线程
                    if (wc >= CAPACITY ||
                        wc >= (core ? corePoolSize : maximumPoolSize))
                        return false;
                    //cas添加线程池状态，添加成功，打断。
                    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;
            boolean workerAdded = false;
            Worker w = null;
            try { 
                //创建Worker
                w = new Worker(firstTask);
                //获取线程
                final Thread t = w.thread;
                //线程不为空
                if (t != null) { 
                    //获取线程池的主锁
                    final ReentrantLock mainLock = this.mainLock;
                    //加锁
                    mainLock.lock();
                    try { 
                        //获取线程池状态
                        int rs = runStateOf(ctl.get());
    					//如果状态是RUNNING或者是SHUTDOWN但执行的任务为空时
                        if (rs < SHUTDOWN ||
                            (rs == SHUTDOWN && firstTask == null)) { 
                            //如果线程已经存在，则异常
                            if (t.isAlive())  
                                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;
        }

## 4.3 runWorker (线程运行) ##

final void runWorker(Worker w) { 
            //获取当前的执行线程
            Thread wt = Thread.currentThread();
            //获取worker的任务后，清除。
            Runnable task = w.firstTask;
            w.firstTask = null;
            //清除构造时状态
            w.unlock(); // allow interrupts
            //标识是否突然被打断
            boolean completedAbruptly = true;
            try { 
                //执行当前的任务 || 调用getTask() 获取任务
                while (task != null || (task = getTask()) != null) { 
                    //加锁
                    w.lock();
                    //如果当前线程池的状态>=STOP，打断当前线程
                    if ((runStateAtLeast(ctl.get(), STOP) ||
                         (Thread.interrupted() &&
                          runStateAtLeast(ctl.get(), STOP))) &&
                        !wt.isInterrupted())
                        wt.interrupt();
                    try { 
                        //空方法
                        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(task, thrown);
                        }
                    } finally { 
                        //执行完清除任务
                        task = null;
                        //完成的任务数+1
                        w.completedTasks++;
                        //解锁
                        w.unlock();
                    }
                }
                completedAbruptly = false;
            } finally { 
                //清理数据
                processWorkerExit(w, completedAbruptly);
            }
        }

## 4.4 getTask (获取任务) ##

private Runnable getTask() { 
        //超时标志
        boolean timedOut = false; // Did the last poll() time out?
    
        //自旋
        for (;;) { 
            //获取当前线程池的状态 || 线程数
            int c = ctl.get();
            //获取当前线程池状态
            int rs = runStateOf(c);
    
            // 如果当前线程池状态状态>=SHUTDOWN，那么清除线程池
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { 
                decrementWorkerCount();
                return null;
            }
            
            //获取线程池数量
            int wc = workerCountOf(c);
    
            //线程数大于核心线程数
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
           
            //(如果线程数已经大于最大池设定的线程数 || 线程数大于核心线程数&&空闲线程存在) 并且 (线程数量大于1 || 队列为空)
            if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) { 
                //清除一个
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }
    
            try { 
                //判断是否有空闲线程
                //在空闲线程存活时间内如果从队列中获取到任务，那么不存在空闲线程。
                //在空闲线程存活时间内，在阻塞队列中获取不到任务，r为空
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                
                //获取任务不为空，直接返回任务
                if (r != null)
                    return r;
                //r为空，存在空闲线程
                timedOut = true;
            } catch (InterruptedException retry) { 
                timedOut = false;
            }
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/37a24328d823466ba3fedd2107a72d20.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/eb9fc6123d764b88a7a1fe3987b0b976.png
[20201118225934372.png_pic_center]: /images/20221120/838584aaf6e74994b3eb8a58eb1c77b0.png