ThreadPoolExecutor 线程池执行流程及核心源码分析

喜欢ヅ旅行 2023-10-12 08:23 40阅读 0赞

####  ####

*   *  线程池常用类扩展
     *  CompletableFuture
     *  线程池
     *   *  ThreadPoolExecutor 基础认识
         *   *  核心方法处理流程
             *  总结 ThreadPoolExecutor 执行流程
         *  SingleThreadExecutor 单一线程池
         *  CacheThreadPool 无界线程池
         *  FixedThreadPool 固定线程池
         *  ScheduledPool 定时任务线程池
         *  ForkJoinPool

concurrent VS parallel

*  并发：在同一个 CPU 上轮流（交替）执行，从逻辑上说它们是同时执行，但从 CPU 内核角度上去看，它们是细粒度下的串行执行的.
 *  并行：将多个任务分配到不同的 CPU 内核上去执行的.

### 线程池常用类扩展 ###

*  Executors：线程池工厂，提供各种不同的线程池，常用的方法有：newFixedThreadPool、newSingleThreadExecutor、newCachedThreadPool 等
 *  Executor：执行者，将定义和运行分开，只有一个 execute 方法，分配一个线程去运行任务
 *  ExecutorService：继承于 Executor，实现了 execute 方法，定义了一整套线程池的生命周期，提供了一个 submit 方法，可以执行有返回值的线程
 *  Callable：接口提供了有返回值的方法 call，如果不通过线程池的 submit 来运行的话，它需要 **FutureTask** 构造函数来进行实例化，然后调用 **new Thread(futureTaskXxx).start** 方法来运行此线程
 *  Future：存储将来的执行才会产生的结果

### CompletableFuture ###

管理多个 future 结果，可以任意组合我们的任务；它可以使我们线程并行的执行，异步地去执行我们的任务

*  runAsync：异步执行无返回值的任务

public static CompletableFuture<Void> runAsync(Runnable runnable) 
 public static CompletableFuture<Void> runAsync(Runnable runnable,Executor executor) // 交由自定义线程池执行

*  supplyAsync：异步执行任务有返回值

public static CompletableFuture supplyAsync(Supplier supplier) {
 
 return asyncSupplyStage(asyncPool, supplier);
 }
 public static CompletableFuture supplyAsync(Supplier supplier,Executor executor) {
 
 return asyncSupplyStage(screenExecutor(executor), supplier);
 }

*  执行完当前任务以后可以继续执行下一个任务：thenAcceptAsync

public CompletableFuture thenApplyAsync(Function<? super T,? extends U> fn)

如果需要等待所有任务都执行完成：allof(future1,future2 ... ).join()，或者等待其中一个任务完成即可：anyof(future1,future2 ...).join()

### 线程池 ###

#### ThreadPoolExecutor 基础认识 ####

线程池分为两部分：线程集合和任务集合，线程集合里的每一个线程可以执行时都会到任务集合里面去取任务执行，以下对线程池的七大参数进行介绍：

*  corePoolSize：核心线程数，一般设置该值 CPU 核数
 *  maxinumPoolSize：最大线程数
 *  keepAliveTime：非核心线程的存活时长（最大线程数-核心线程数 剩下的线程都是非核心线程）
 *  unit：存活时长的单位
 *  workQueue：工作阻塞队列，核心线程满了以后不够用就往队列里面塞
 *  threadFactory：生产线程的工厂，一般是采用默认的工厂生产线程
 *  handler：拒绝策略，当工作阻塞队列满了以后，线程数也满了，其余的任务采用那种方式去拒绝处理「`Abort：抛异常、Discard：扔掉，不抛异常、DiscardOldest：扔掉排队时间最久的任务、CallerRuns：等待调用者处理任务`」
 *  自定义拒绝策略：实现接口 **RejectedExecutionHandler** 处理满载的任务

#### ThreadPoolExecutor 源码分析 ####

##### 核心方法处理流程 #####

**常用变量分析**

// 1.ctl：可以看做一个int类型的数字，高3位表示线程池状态，低29位表示worker数量
 private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
 // 2.COUNT_BITS：`Integer.SIZE`为32，所以`COUNT_BITS`为29
 private static final int COUNT_BITS = Integer.SIZE - 3;
 // 3.CAPACITY：线程池允许的最大线程数，1左移29位，然后减1，即为 2^29 - 1
 private static final int CAPACITY = (1 << COUNT_BITS) - 1;
 // 4.线程池有5种状态，按大小排序如下：RUNNING < SHUTDOWN < STOP < TIDYING < TERMINATED
 private static final int RUNNING = -1 << COUNT_BITS;
 private static final int SHUTDOWN = 0 << COUNT_BITS;
 private static final int STOP = 1 << COUNT_BITS;
 private static final int TIDYING = 2 << COUNT_BITS;
 private static final int TERMINATED = 3 << COUNT_BITS;
 
 // 5.runStateOf()：获取线程池状态，通过按位与操作，低29位将全部变成0
 private static int runStateOf(int c) {
 return c &~CAPACITY; }
 // 6.workerCountOf()：获取线程池worker数量，通过按位与操作，高3位将全部变成0
 private static int workerCountOf(int c) {
 return c &CAPACITY; }
 // 7.ctlOf()：根据线程池状态和线程池worker数量，生成ctl值
 private static int ctlOf(int rs, int wc) {
 return rs | wc; }
 
 /*
 * Bit field accessors that don't require unpacking ctl.
 * These depend on the bit layout and on workerCount being never negative.
 */
 // 8. `runStateLessThan()`：线程池状态小于xx
 private static boolean runStateLessThan(int c, int s) {
 
 return c < s;
 }
 // 9.runStateAtLeast()：线程池状态大于等于xx
 private static boolean runStateAtLeast(int c, int s) {
 
 return c >= s;
 }

**构造方法**

public ThreadPoolExecutor(int corePoolSize,
 int maximumPoolSize,
 long keepAliveTime,
 TimeUnit unit,
 BlockingQueue<Runnable> workQueue,
 ThreadFactory threadFactory,
 RejectedExecutionHandler handler) {
 
 // 基本类型参数校验
 if (corePoolSize < 0 ||
 maximumPoolSize <= 0 ||
 maximumPoolSize < corePoolSize ||
 keepAliveTime < 0)
 throw new IllegalArgumentException();
 // 空指针校验
 if (workQueue == null || threadFactory == null || handler == null)
 throw new NullPointerException();
 this.corePoolSize = corePoolSize;
 this.maximumPoolSize = maximumPoolSize;
 this.workQueue = workQueue;
 // 根据传入参数`unit`和`keepAliveTime`，将存活时间转换为纳秒存到变量`keepAliveTime `中
 this.keepAliveTime = unit.toNanos(keepAliveTime);
 this.threadFactory = threadFactory;
 this.handler = handler;
 }

**提交 task 过程**

public void execute(Runnable command) {
 
 if (command == null)
 throw new NullPointerException();
 int c = ctl.get();
 // worker数量比核心线程数小，直接创建worker执行任务
 if (workerCountOf(c) < corePoolSize) {
 
 if (addWorker(command, true))
 return;
 c = ctl.get();
 }
 // worker数量超过核心线程数，任务直接进入队列
 if (isRunning(c) && workQueue.offer(command)) {
 
 int recheck = ctl.get();
 // 线程池状态不是RUNNING状态，说明执行过shutdown命令，需要对新加入的任务执行reject()操作。
 // 这儿为什么需要recheck，是因为任务入队列前后，线程池的状态可能会发生变化。
 if (! isRunning(recheck) && remove(command))
 reject(command);
 // 这儿为什么需要判断0值，主要是在线程池构造方法中，核心线程数允许为0
 else if (workerCountOf(recheck) == 0)
 addWorker(null, false);
 }
 // 如果线程池不是运行状态，或者任务进入队列失败，则尝试创建worker执行任务。
 // 这儿有3点需要注意：
 // 1. 线程池不是运行状态时，addWorker内部会判断线程池状态
 // 2. addWorker第2个参数表示是否创建核心线程
 // 3. addWorker返回false，则说明任务执行失败，需要执行reject操作
 else if (!addWorker(command, false))
 reject(command);
 }

**addWorker 源码解析**

private boolean addWorker(Runnable firstTask, boolean core) {
 
 retry:
 // 外层自旋
 for (;;) {
 
 int c = ctl.get();
 int rs = runStateOf(c);
 
 // 这个条件写得比较难懂，我对其进行了调整，和下面的条件等价
 // (rs > SHUTDOWN) || 
 // (rs == SHUTDOWN && firstTask != null) || 
 // (rs == SHUTDOWN && workQueue.isEmpty())
 // 1. 线程池状态大于 SHUTDOWN，直接返回false
 // 2. 线程池状态等于 SHUTDOWN，且firstTask不为null，直接返回false
 // 3. 线程池状态等于 SHUTDOWN，且队列为空，直接返回false
 if (rs >= SHUTDOWN &&
 ! (rs == SHUTDOWN &&
 firstTask == null &&
 ! workQueue.isEmpty()))
 return false;
 
 // 内层自旋
 for (;;) {
 
 int wc = workerCountOf(c);
 // worker数量超过容量，直接返回false
 if (wc >= CAPACITY ||
 wc >= (core ? corePoolSize : maximumPoolSize))
 return false;
 // 使用CAS的方式增加worker数量。
 // 若增加成功，则直接跳出外层循环进入到第二部分
 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 {
 
 w = new Worker(firstTask);
 final Thread t = w.thread;
 if (t != null) {
 
 final ReentrantLock mainLock = this.mainLock;
 // worker的添加必须是串行的，因此需要加锁
 mainLock.lock();
 try {
 
 // 这儿需要重新检查线程池状态
 int rs = runStateOf(ctl.get());
 
 if (rs < SHUTDOWN ||
 (rs == SHUTDOWN && firstTask == null)) {
 
 // worker 已经调用过了 start() 方法，则不再创建 worker
 if (t.isAlive()) // precheck that t is startable
 throw new IllegalThreadStateException();
 // worker创建并添加到workers成功
 workers.add(w);
 // 更新`largestPoolSize`变量
 int s = workers.size();
 if (s > largestPoolSize)
 largestPoolSize = s;
 workerAdded = true;
 }
 } finally {
 
 mainLock.unlock();
 }
 // 启动worker线程
 if (workerAdded) {
 
 t.start();
 workerStarted = true;
 }
 }
 } finally {
 
 // worker线程启动失败，说明线程池状态发生了变化（关闭操作被执行），需要进行shutdown相关操作
 if (! workerStarted)
 addWorkerFailed(w);
 }
 return workerStarted;
 }

**线程池 worker 任务单元**

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
    {
        
        /**
         * This class will never be serialized, but we provide a
         * serialVersionUID to suppress a javac warning.
         */
        private static final long serialVersionUID = 6138294804551838833L;
    
        /** Thread this worker is running in.  Null if factory fails. */
        final Thread thread;
        /** Initial task to run.  Possibly null. */
        Runnable firstTask;
        /** Per-thread task counter */
        volatile long completedTasks;
    
        /**
         * Creates with given first task and thread from ThreadFactory.
         * @param firstTask the first task (null if none)
         */
        Worker(Runnable firstTask) {
        
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            // 这儿是Worker的关键所在，使用了线程工厂创建了一个线程。传入的参数为当前worker
            this.thread = getThreadFactory().newThread(this);
        }
    
        /** Delegates main run loop to outer runWorker  */
        public void run() {
        
            runWorker(this);
        }
    
        // 省略代码...
    }

**核心线程执行逻辑：runWorker**

final void runWorker(Worker w) {
        
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        // 调用unlock()是为了让外部可以中断
        w.unlock(); // allow interrupts
        // 这个变量用于判断是否进入过自旋（while循环）
        boolean completedAbruptly = true;
        try {
        
            // 这儿是自旋
            // 1. 如果firstTask不为null，则执行firstTask；
            // 2. 如果firstTask为null，则调用getTask()从队列获取任务。
            // 3. 阻塞队列的特性就是：当队列为空时，当前线程会被阻塞等待
            while (task != null || (task = getTask()) != null) {
        
                // 这儿对worker进行加锁，是为了达到下面的目的
                // 1. 降低锁范围，提升性能
                // 2. 保证每个worker执行的任务是串行的
                w.lock();
                // If pool is stopping, ensure thread is interrupted;
                // if not, ensure thread is not interrupted.  This
                // requires a recheck in second case to deal with
                // shutdownNow race while clearing interrupt
                // 如果线程池正在停止，则对当前线程进行中断操作
                if ((runStateAtLeast(ctl.get(), STOP) ||
                     (Thread.interrupted() &&
                      runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                    wt.interrupt();
                // 执行任务，且在执行前后通过 beforeExecute()、afterExecute() 来扩展其功能
                // 这两个方法在当前类里面为空实现
                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 {
        
                    // 帮助gc
                    task = null;
                    // 已完成任务数加一 
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
        
            // 自旋操作被退出，说明线程池正在结束
            processWorkerExit(w, completedAbruptly);
        }
    }

##### 总结 ThreadPoolExecutor 执行流程 #####

*  当有新任务到达时，启动核心线程（第一个参数）
 *  如果核心线程被占满了，新的任务往阻塞队列里面塞（第三个参数）
 *  如果阻塞队列也满了，则启动非核心线程（第二个参数-第一个参数的值）进行处理；当非核心线程处理完了工作以后，处于闲置状态，先让其存在一段时间 keepAliveTime（第四个参数和第五个参数的组合） 以后就进行收回
 *  如果核心线程数用完了、阻塞队列满了、非核心线程数也多使用完了，那么就使用配置好的拒绝策略（第七个参数）来处理新的任务「丢弃存活时间最长的、直接丢弃、抛出异常、交由调用者处理、自定义策略进行实现」
 *  一般线程的产生，采用的都是默认的工厂进行创建的（第六个参数）

**`阿里巴巴编码规范中推荐我们使用自定义线程池的方法来管理线程，以下几种方式都有各自的缺点，没有自定义的那么灵活`**

#### SingleThreadExecutor 单一线程池 ####

Executors.newSingleThreadExecutor()：它的核心线程数和最大线程数都是 1，无存活时长，它的阻塞队列是用 **LinkedBlockingQueue** 来构建的，最大可达长度为 **Integer.MAX\_VALUE**

#### CacheThreadPool 无界线程池 ####

Executors.newCachedThreadPool()：它的核心线程数为 0，最大线程数为 Integer.MAX\_VALUE 值，它的阻塞队列是 **SynchronousQueue** 构建的，是不存储元素的，丢一个任务就取出线程处理一个

#### FixedThreadPool 固定线程池 ####

Executors.newFixedThreadPool(nThreads)：固定线程数的池，核心线程数和最大线程数都是 **nThreads** 值，它的阻塞队列是用 **LinkedBlockingQueue** 来构建的，最大可达长度为 **Integer.MAX\_VALUE**

> Cache vs Fixed
> 
> 任务忽高忽低时，适合于用 Cached，任务比较平稳时，用 Fixed  
> 引入书中建议：如果线程池中线程数量过多，最终它们会竞争稀缺的处理器和内存资源，浪费大量的时间在上下文切换上。反之，如果线程数目过少，正如你的应用所面临情况，处理器上的一些核可能就无法充分利用到

#### ScheduledPool 定时任务线程池 ####

Executors.newScheduledThreadPool(corePoolSize)：核心线程数是传入的 corePoolSize 值，最大线程数是 Integer.MAX\_VALUE，它的阻塞队列是用 **DelayedWorkQueue** 来构建 **scheduleAtFixedRate** 方法可以用来执行某项任务，它有四个参数组成：第一个参数是表示要执行什么样的任务，第二个参数是刚开始要推迟多长时间，第三个参数是间隔多长时间执行一次，第四个就是时间的单位

public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
 long initialDelay,
 long period,
 TimeUnit unit)

#### ForkJoinPool ####

* 将一个大任务分解成多个小任务去执行，最后再将分解的小任务进行汇总，即分解汇总的任务
 * 用很少的线程可以执行很多的任务（子任务），TPE 做不到先执行子任务
 * 实现分解任务无返回值的可继承至 **RecursiveAction** 类 `public abstract class RecursiveAction extends ForkJoinTask<Void>`， 
 重写其抽象方法：protected void compute()
 * 实现分解任务有返回值要继承至 **RecursiveTask** 类 `public abstract class RecursiveTask<V> extends ForkJoinTask<V>` 
 重写其抽象方法：protected Long compute()
 * CPU 密集型

更多技术文章可以查看：[vnjohn 个人博客][vnjohn]

[vnjohn]: https://www.vnjohn.com