Java Review - 并发编程_ThreadPoolExecutor原理&源码剖析(上)

简介: Java Review - 并发编程_ThreadPoolExecutor原理&源码剖析(上)

195d03d17afc4a928bc581f313b01dfe.png

线程池主要解决两个问题


一是当执行大量异步任务时线程池能够提供较好的性能。在不使用线程池时,每当需要执行异步任务时直接new一个线程来运行,而线程的创建和销毁是需要开销的。线程池里面的线程是可复用的,不需要每次执行异步任务时都重新创建和销毁线程。


二是线程池提供了一种资源限制和管理的手段,比如可以限制线程的个数,动态新增线程等。每个ThreadPoolExecutor也保留了一些基本的统计数据,比如当前线程池完成的任务数目等。


另外,线程池也提供了许多可调参数和可扩展性接口,以满足不同情境的需要,程序员可以使用更方便的Executors的工厂方法,比如newCachedThreadPool(线程池线程个数最多可达Integer.MAX_VALUE,线程自动回收)、newFixedThreadPool(固定大小的线程池)和newSingleThreadExecutor(单个线程)等来创建线程池,当然用户还可以自定义。


类关系图


image.png

在上图中,Executors其实是个工具类,里面提供了好多静态方法,这些方法根据用户选择返回不同的线程池实例。


ctl 含义 ---- 记录线程池状态和线程池中线程个数

ThreadPoolExecutor继承了AbstractExecutorService,成员变量ctl是一个Integer的原子变量,用来记录线程池状态和线程池中线程个数,类似于ReentrantReadWriteLock使用一个变量来保存两种信息。

  private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

这里假设Integer类型是32位二进制表示,则其中高3位用来表示线程池状态,后面29位用来记录线程池线程个数。

/用来标记线程池状态(高3位),线程个数(低29位)
//默认是RUNNING状态,线程个数为0
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//线程个数掩码位数
private static final int COUNT_BITS = Integer.SIZE - 3;
//线程最大个数(低29位)00011111111111111111111111111111
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;


线程池状态:

//(高3位):11100000000000000000000000000000
private static final int RUNNING    = -1 << COUNT_BITS;
//(高3位):00000000000000000000000000000000
private static final int SHUTDOWN   =  0 << COUNT_BITS;
//(高3位):00100000000000000000000000000000
private static final int STOP       =  1 << COUNT_BITS;
//(高3位):01000000000000000000000000000000
private static final int TIDYING    =  2 << COUNT_BITS;
//(高3位):01100000000000000000000000000000
private static final int TERMINATED =  3 << COUNT_BITS;
// 获取高三位 运行状态
private static int runStateOf(int c)     { return c & ~CAPACITY; }
//获取低29位 线程个数
private static int workerCountOf(int c)  { return c & CAPACITY; }
//计算ctl新值,线程状态 与 线程个数
private static int ctlOf(int rs, int wc) { return rs | wc; }


线程池状态 及转换


RUNNING:接受新任务并且处理阻塞队列里的任务。


SHUTDOWN:拒绝新任务但是处理阻塞队列里的任务。


STOP:拒绝新任务并且抛弃阻塞队列里的任务,同时会中断正在处理的任务。


TIDYING:所有任务都执行完(包含阻塞队列里面的任务)后当前线程池活动线程数为0,将要调用terminated方法。


TERMINATED:终止状态。terminated方法调用完成以后的状态。


线程池状态转换列举如下。


RUNNING -> SHUTDOWN :显式调用shutdown()方法,或者隐式调用了finalize()、方法里面的shutdown()方法。


RUNNING 或 SHUTDOWN)-> STOP :显式调用 shutdownNow()方法时。


SHUTDOWN -> TIDYING :当线程池和任务队列都为空时。


STOP -> TIDYING :当线程池为空时。


TIDYING -> TERMINATED: 当 terminated() hook 方法执行完成时


线程池参数


corePoolSize:线程池核心线程个数。


workQueue:用于保存等待执行的任务的阻塞队列, 比如基于数组的有界ArrayBlockingQueue、基于链表的无界LinkedBlockingQueue、最多只有一个元素的同步队列SynchronousQueue及优先级队列PriorityBlockingQueue等。


maximunPoolSize:线程池最大线程数量。


ThreadFactory:创建线程的工厂。


RejectedExecutionHandler:饱和策略,当队列满并且线程个数达到maximunPoolSize后采取的策略。


比如

AbortPolicy(抛出异常)、

CallerRunsPolicy(使用调用者所在线程来运行任务)、

DiscardOldestPolicy(调用poll丢弃一个任务,执行当前任务)

DiscardPolicy(默默丢弃,不抛出异常)


keeyAliveTime:存活时间。如果当前线程池中的线程数量比核心线程数量多,并且是闲置状态,则这些闲置的线程能存活的最大时间。


TimeUnit:存活时间的时间单位


线程池类型


  • newFixedThreadPool :创建一个核心线程个数和最大线程个数都为nThreads的线程池,并且阻塞队列长度为Integer.MAX_VALUE。keeyAliveTime=0说明只要线程个数比核心线程个数多并且当前空闲则回收


    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }
  // 使用自定义线程创建工厂
    public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>(),
                                      threadFactory);
    }

  • newSingleThreadExecutor: 创建一个核心线程个数和最大线程个数都为1的线程池,并且阻塞队列长度为Integer.MAX_VALUE。keeyAliveTime=0说明只要线程个数比核心线程个数多并且当前空闲则回收。
    public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }
    public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>(),
                                    threadFactory));
    }


newCachedThreadPool :创建一个按需创建线程的线程池,初始线程个数为0,最多线程个数为Integer.MAX_VALUE,并且阻塞队列为同步队列。keeyAliveTime=60说明只要当前线程在60s内空闲则回收。这个类型的特殊之处在于,加入同步队列的任务会被马上执行,同步队列里面最多只有一个任务。

  public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }
    public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>(),
                                      threadFactory);
    }

mainLock & termination

    /**
     * Lock held on access to workers set and related bookkeeping.
     * While we could use a concurrent set of some sort, it turns out
     * to be generally preferable to use a lock. Among the reasons is
     * that this serializes interruptIdleWorkers, which avoids
     * unnecessary interrupt storms, especially during shutdown.
     * Otherwise exiting threads would concurrently interrupt those
     * that have not yet interrupted. It also simplifies some of the
     * associated statistics bookkeeping of largestPoolSize etc. We
     * also hold mainLock on shutdown and shutdownNow, for the sake of
     * ensuring workers set is stable while separately checking
     * permission to interrupt and actually interrupting.
     */
    private final ReentrantLock mainLock = new ReentrantLock();
   /**
     * Wait condition to support awaitTermination
     */
    private final Condition termination = mainLock.newCondition();


mainLock是独占锁,用来控制新增Worker线程操作的原子性。

termination是该锁对应的条件队列,在线程调用awaitTermination时用来存放阻塞的线程。


Worker


Worker继承AQS和Runnable接口,是具体承载任务的对象。Worker继承了AQS,自己实现了简单不可重入独占锁,其中state=0表示锁未被获取状态,state=1表示锁已经被获取的状态,state=-1是创建Worker时默认的状态,创建时状态设置为-1是为了避免该线程在运行runWorker()方法前被中断。其中变量firstTask记录该工作线程执行的第一个任务,thread是具体执行任务的线程。


DefaultThreadFactory是线程工厂,newThread方法是对线程的一个修饰。其中poolNumber是个静态的原子变量,用来统计线程工厂的个数,threadNumber用来记录每个线程工厂创建了多少线程,这两个值也作为线程池和线程的名称的一部分。


源码分析


public void execute(Runnable command)


execute方法的作用是提交任务command到线程池进行执行。用户线程提交任务到线程池的模型图如下图所示。

653f89d726ed4b1fbc5dcee1bf0ade74.png


从该图可以看出,ThreadPoolExecutor的实现实际是一个生产消费模型,当用户添加任务到线程池时相当于生产者生产元素,workers线程工作集中的线程直接执行任务或者从任务队列里面获取任务时则相当于消费者消费元素。

用户线程提交任务的execute方法的具体代码如下

public void execute(Runnable command) {
    // 1  任务为null ,抛出 npe异常
        if (command == null)
            throw new NullPointerException();
        /*
         * Proceed in 3 steps:
         *
         * 1. If fewer than corePoolSize threads are running, try to
         * start a new thread with the given command as its first
         * task.  The call to addWorker atomically checks runState and
         * workerCount, and so prevents false alarms that would add
         * threads when it shouldn't, by returning false.
         *
         * 2. If a task can be successfully queued, then we still need
         * to double-check whether we should have added a thread
         * (because existing ones died since last checking) or that
         * the pool shut down since entry into this method. So we
         * recheck state and if necessary roll back the enqueuing if
         * stopped, or start a new thread if there are none.
         *
         * 3. If we cannot queue task, then we try to add a new
         * thread.  If it fails, we know we are shut down or saturated
         * and so reject the task.
         */
        // 2 获取当前线程池的状态 + 线程个数变量的组合值 
        int c = ctl.get();
    // 3 当前线程池中的个数是否小于corePoolSize ,小于的话则开启新的线程 
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        // 4 如果线程池处于running状态,则添加任务到阻塞队列
        if (isRunning(c) && workQueue.offer(command)) {
          // 4.1 二次检查
            int recheck = ctl.get();
            // 4.2 如果当前线程池的状态不是running 则从队列中移除任务,并执行拒绝策略
            if (! isRunning(recheck) && remove(command))
                reject(command);
            // 4.3 否则当线程数数量为空,则添加一个线程    
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        } 
        // 5 如果队列满,则新增线程,新增线程失败,触发拒绝策略 
        else if (!addWorker(command, false))
            reject(command);
    }


代码(3)判断如果当前线程池中线程个数小于corePoolSize,会向workers里面新增一个核心线程(core线程)执行该任务。


如果当前线程池中线程个数大于等于corePoolSize则执行代码(4)。如果当前线程池处于RUNNING状态则添加当前任务到任务队列。这里需要判断线程池状态是因为有可能线程池已经处于非RUNNING状态,而在非RUNNING状态下是要抛弃新任务的。


如果向任务队列添加任务成功,则代码(4.2)对线程池状态进行二次校验,这是因为添加任务到任务队列后,执行代码(4.2)前有可能线程池的状态已经变化了。这里进行二次校验,如果当前线程池状态不是RUNNING了则把任务从任务队列移除,移除后执行拒绝策略;如果二次校验通过,则执行代码(4.3)重新判断当前线程池里面是否还有线程,如果没有则新增一个线程。


如果代码(4)添加任务失败,则说明任务队列已满,那么执行代码(5)尝试新开启线程(如上的thread3和thread4)来执行该任务,如果当前线程池中线程个数>maximumPoolSize则执行拒绝策略。


新增线程addWorkder源码分析


/**
     * Checks if a new worker can be added with respect to current
     * pool state and the given bound (either core or maximum). If so,
     * the worker count is adjusted accordingly, and, if possible, a
     * new worker is created and started, running firstTask as its
     * first task. This method returns false if the pool is stopped or
     * eligible to shut down. It also returns false if the thread
     * factory fails to create a thread when asked.  If the thread
     * creation fails, either due to the thread factory returning
     * null, or due to an exception (typically OutOfMemoryError in
     * Thread.start()), we roll back cleanly.
     *
     * @param firstTask the task the new thread should run first (or
     * null if none). Workers are created with an initial first task
     * (in method execute()) to bypass queuing when there are fewer
     * than corePoolSize threads (in which case we always start one),
     * or when the queue is full (in which case we must bypass queue).
     * Initially idle threads are usually created via
     * prestartCoreThread or to replace other dying workers.
     *
     * @param core if true use corePoolSize as bound, else
     * maximumPoolSize. (A boolean indicator is used here rather than a
     * value to ensure reads of fresh values after checking other pool
     * state).
     * @return true if successful
     */
    private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
            // Check if queue empty only if necessary. 
            // 6 检查队列是否只在必要的时候为空
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;
      // 7 循环CAS增加线程个数 
            for (;;) {
                int wc = workerCountOf(c);
                // 7.1 如果线程个数超过限制 则返回false
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                // 7.2 cas增加线程个数,同时只能有1个线程成功  
                if (compareAndIncrementWorkerCount(c))
                    break retry;
               // 7.3 cas失败了,则看线程池状态是否变化了,变化则跳到外层循环重试重新获取线程池状态,否者内层循环重新cas。
                c = ctl.get();  // Re-read ctl
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }
        }
    // 8 到这里,说明CAS成功了
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
          // 8.1 创建Worker
           w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {
              // 8.2 加独占锁,为了workers同步,因为可能多个线程调用了线程池的execute方法。
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    //8.3 重新检查线程池的状态,避免在获取锁前调用了shutdown接口
                    int rs = runStateOf(ctl.get());
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        // 8.4 添加任务
                        workers.add(w);
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                // 8.5 添加成功,则启动线程
                if (workerAdded) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }


主要分两个部分:

  • 第一部分双重循环的目的是通过CAS操作增加线程数;
  • 第二部分主要是把并发安全的任务添加到workers里面,并且启动任务执行。


相关文章
|
1天前
|
Java 调度 开发者
Java 并发编程的探索与实践
【5月更文挑战第3天】在当今多核处理器普及的时代,并发编程已经成为提高程序性能的重要手段。本文将深入探讨 Java 并发编程的基本概念、原理及其在实际项目中的应用,帮助读者更好地理解和掌握 Java 并发编程技巧。
|
2天前
|
并行计算 安全 Java
Java 并发编程的探索之旅
【5月更文挑战第2天】 在多线程的世界里,程序的行为变得错综复杂。本文将带您走进 Java 并发编程的核心概念,通过深入分析并发工具的使用和原理,帮助您构建高效、安全且响应迅速的应用程序。我们将探讨线程的基本知识,同步机制,以及高级并发工具如 Executors、Futures 和 Streams。通过理论与实践相结合的方式,为开发者提供一份清晰、实用的并发编程指南。
7 2
|
3天前
|
存储 安全 Java
深入理解Java并发编程:线程安全与性能优化
【5月更文挑战第1天】本文将深入探讨Java并发编程的核心概念,包括线程安全和性能优化。我们将详细分析线程安全问题的根源,以及如何通过合理的设计和编码实践来避免常见的并发问题。同时,我们还将探讨如何在保证线程安全的前提下,提高程序的并发性能,包括使用高效的同步机制、减少锁的竞争以及利用现代硬件的并行能力等技术手段。
|
3天前
|
安全 Java 程序员
Java并发编程:理解并应用ReentrantLock
【4月更文挑战第30天】 在多线程的世界中,高效且安全地管理共享资源是至关重要的。本文深入探讨了Java中的一种强大同步工具——ReentrantLock。我们将从其设计原理出发,通过实例演示其在解决并发问题中的实际应用,以及如何比传统的synchronized关键字提供更灵活的锁定机制。文章还将讨论在使用ReentrantLock时可能遇到的一些挑战和最佳实践,帮助开发者避免常见陷阱,提高程序性能和稳定性。
|
Java
Java并发编程笔记之FutureTask源码分析
FutureTask可用于异步获取执行结果或取消执行任务的场景。通过传入Runnable或者Callable的任务给FutureTask,直接调用其run方法或者放入线程池执行,之后可以在外部通过FutureTask的get方法异步获取执行结果,因此,FutureTask非常适合用于耗时的计算,主线程可以在完成自己的任务后,再去获取结果。
4259 0
|
Java 调度 API
Java并发编程笔记之Timer源码分析
timer在JDK里面,是很早的一个API了。具有延时的,并具有周期性的任务,在newScheduledThreadPool出来之前我们一般会用Timer和TimerTask来做,但是Timer存在一些缺陷,为什么这么说呢?   Timer只创建唯一的线程来执行所有Timer任务。
2967 0
|
Java
Java并发编程笔记之Semaphore信号量源码分析
JUC 中 Semaphore 的使用与原理分析,Semaphore 也是 Java 中的一个同步器,与 CountDownLatch 和 CycleBarrier 不同在于它内部的计数器是递增的,那么,Semaphore 的内部实现是怎样的呢?   Semaphore 信号量也是Java 中一个同步容器,与CountDownLatch 和 CyclicBarrier 不同之处在于它内部的计数器是递增的。
4250 0
|
Java
Java并发编程笔记之CyclicBarrier源码分析
JUC 中 回环屏障 CyclicBarrier 的使用与分析,它也可以实现像 CountDownLatch 一样让一组线程全部到达一个状态后再全部同时执行,但是 CyclicBarrier 可以被复用。
2197 0
|
Java
Java并发编程笔记之 CountDownLatch闭锁的源码分析
JUC 中倒数计数器 CountDownLatch 的使用与原理分析,当需要等待多个线程执行完毕后在做一件事情时候 CountDownLatch 是比调用线程的 join 方法更好的选择,CountDownLatch 与 线程的 join 方法区别是什么? 日常开发中经常会遇到需要在主线程中开启多线程去并行执行任务,并且主线程需要等待所有子线程执行完毕后再进行汇总的场景,它的内部提供了一个计数器,在构造闭锁时必须指定计数器的初始值,且计数器的初始值必须大于0。
6355 0
|
Java
Java并发编程笔记之ArrayBlockingQueue源码分析
JDK 中基于数组的阻塞队列 ArrayBlockingQueue 原理剖析,ArrayBlockingQueue 内部如何基于一把独占锁以及对应的两个条件变量实现出入队操作的线程安全? 首先我们先大概的浏览一下ArrayBlockingQueue 的内部构造,如下类图: 如类图所示,可以看到ArrayBlockingQueue 内部有个数组items 用来存放队列元素,putIndex变量标示入队元素的下标,takeIndex是出队的下标,count是用来统计队列元素个数, 从定义可以知道,这些属性并没有使用valatile修饰,这是因为访问这些变量的使用都是在锁块内被用。
4314 0