• 线程池ThreadPoolExecutor

    ThreadPoolExecutor是JDK1.5版本推出的一个线程池。是ExecutorService接口的实现之一，也是阿里推荐使用的一种线程池。线程池解决了两个不同的问题：

    1. 在执行大量异步任务的时候，减少创建线程带来的性能开销，提高系统性能
    2. 它提供了一种限制和管理资源的方法，包括在正在执行任务的线程集合。

    每一个ThreadPoolExecutor也维护了一些基本的统计数据，例如已完成的任务数量。为了在广泛的上下文中应用，该类提供了许多的可调参数和可扩展性的钩子。

    简介

    核心线程数和非核心线程数

    ThreadPoolExecutor会根据corePoolSize和maximumPoolSize参数自动调整线程池的大小。当方法execute(Runnable)提交一个新任务的时候，如果运行的线程数量小于corePoolSize，即使存在空闲的线程也会创建一个新的线程来处理请求，如果运行的线程数量大于corePoolSize但小于maximumPoolSize时，并且队列已经满了，则会创建一个新的线程执行任务。corePoolSize和maximumPoolSize的数量也可以通过setCorePoolSize和setMaximumPoolSize动态的修改。
    默认情况下ThreadPoolExecutor核心线程在任务到达的时候创建和启动的。你可以使用prestartCoreThread、prestartAllCoreThreads来动态覆盖来预先启动线程，或者使用非空队列的时候，可能需要虚线启动线程。

    创建线程

    使用ThreadFactory创建新的线程，如果不指定，默认使用Executors.defaultThreadFactory，他创建的线程具有相同的TheadGroup、相同的优先级、相同的非守护状态。通过不同的ThreadFactory，你可以方便的更改线程的名字、线程的优先级、线程的守护状态。

    非核心线程空闲后终止策略

    当前线程池中超过了corePoolSize数量的线程，多余的空闲线程在keepAliveTime时间到达后将被终止。这提供了一种减少资源消耗的方法。如果稍后任务量更加大的时候，线程池又会创建新的线程执行任务。默认情况下只有线程达到了corePoolSize数量时，才会应用keepAliveTime策略。但是方法allowCoreThreadTimeOut(boolean)也可以将这个超时时间策略应用在核心线程上，前提是keepAliveTime不是0。

    队列

    任何的BlockingQueue队列都可以用来传输和提交任务。

    1. 如果线程池中运行线程数量小于corePoolSize，那么线程池更倾向于创建一个新的线程执行任务，而不是将线程放入等待队列中，
    2. 如果线程池中的任务大于了corePoolSize，那么线程池更加倾向于将任务放入队列，而不是创建新的线程，如果此时队列满了，线程池会创建一个新的线程，如果在此时线程数已经达到了maximumPoolSize，这种情况下任务将被拒绝。
    直接交换队列

    直接交换队列的一个很好的选择是SynchronousQueue。它本身不存储任何任务，只是将任务传递给线程，如果此时没有任何线程来执行这个任务，并且下一个任务已经到达，那么任务就排队就会失败，导致线程池创建一个新的非核心线程。直接传递需要无限的maxmunPoolsize,以避免决绝新提交的任务。然任务的到达速度高于线程处理任务的速度，那么就会导致线程无限增长。

    无界队列

    无界队列LinkedBlockingQueue表示没有设置具体容量的队列，这将导致在corePoolSize都处于繁忙时，任务都会放入队列中，因而不会创建超过corePoolSize的线程，因此maxmunPoolSize没有任何效果。如果任务平均到达速度高于线程处理任务的速度，将可能造成队列的任务过多导致OOM。

    有界队列

    有界队列，例如ArrayBlockingQueue可以使用maxmumPoolSize防止系统的资源被耗尽，种情况下更加难以调优和控制。线程池大小和队列大小可以相互权衡，大队列小线程池可以减少CPU的开销，但系统的吞吐量就会变低。大线程池小队列可以增加CPU的开销，可能会遇到不可预测的调度开销，也会降低吞吐量。

    决绝策略

    当线程池已经关闭或者使用有界队列的前提下，线程的数量已经达到了maxmunPoolSize并且有界队列已经饱和，那么通过execute(Runnable)方法提交的新任务将被拒绝。这种情况下execute方法会调用RejectedExecutionHandler#rejectedExecution(Runnable,ThreadPoolExecutor)方法。提供了四个预定义的拒绝粗略：

    1. AbortPolicy(默认)：抛出RejectedExecutionException运行时异常
    2. CallerRunPolicy：抛给调用execute(Runnable)的线程去执行，也就是交给启动该任务的线程去执行。
    3. DiscardPolicy：丢弃策略，直接将任务丢弃。
    4. DiacardOldestPolicy：丢弃队列头部任务，也就是丢弃最老的任务。

    Hook函数

    ThreadPoolExecutor提供可覆写的preExecute(Thread,Runbale)、afterExecute(Thread,Runnable)方法，在任务执行之前和执行之后调用。这些可以用来控制执行环境，例如可以重新初始化ThreadLocals、收集统计数据或者增加日志。此外，还可以覆写的terminated方法，可以在所有Executor全部终止后执行所需要的特殊处理逻辑。
    如果Hook函数或者回调方法失败，那么内部的工作线程将会失败并突然终止。

    队列维护

    getQueue()方法允许访问工作队列，以进行监视和调试。强烈反对将此方法用于其他目的。提供了两个remove(Runnable)和purge方法，可用于在大量任务被取消执行的时候进行回收存储。

    构造函数

    /**
 *
 * 根据给定的初始化参数、默认的线程工厂、默认的拒绝策略创建一个ThreadPoolExecutor线程池对象
 * 使用Executors工厂创建线程池可能比使用这个方法创建线程池更加方便
 *
 * @param corePoolSize 即使闲成处于空闲状态，也要保留在线程池中的线程数
 * @param maximumPoolSize 线程池允许存在的最大闲成数量
 * @param keepAliveTime 当线线程池中的线程数大于核心线程数的时候，多余出来空闲线程在终止前等待任务的最大时间
 * @param unit 等待时间的单位
 * @param workQueue 用于在执行任务之前保存任务的队列，该队列只保存由 execute()方法提交的Runnable任务
 */
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), defaultHandler);
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19

    成员变量

    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    线程池控制状态ctl是一个原子整数，包含了两个概念字段：

    1. workerCount：有效的线程数数量
    2. runState有效线程的运行状态，包括是否运行和关闭

    runState提供了主要生命周期控制，接受以下值：

    1. RUNNING：接收新任务和处理队列中的任务
    2. SHUTDOWN：不接收新任务，但是处理队列中的任务
    3. STOP：不接受收新任务，也不处理队列中的任务，并中断正在执行的任务
    4. TIDYING：所有任务都已经终止，workCount已变为0，转换为到TIDYING状态的线程将会运行terminated()钩子方法
    5. TERMINATED： 方法terminated()已经执行完

    成员方法

    execute(Runnbale)

    public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    //获取正在运行的线程数量
    int c = ctl.get();
    if (workerCountOf(c) < corePoolSize) {
        //调用addWorker方法
        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);
    }
	//调入addWorker失败,调用拒绝策略
    else if (!addWorker(command, false))
        reject(command);
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23

    在将来某个时间执行给定的任务，可以是新创建的线程来执行，也可以是线程池中已存在的线程来执行。如果线程池已关闭或者已经达到其最大容量，则会执行RejectedExecutionHandler的拒绝策略。
    该方法时一个核心方法，是任务提交的入口。该方法中最核心的方法就是addWorker方法。
    该方法的主要逻辑：

    1. 如果运行的线程是否小于corePoolSize,则会尝试用给定的Runnable对象作为第一个任务启动第一个新线程. 调用addWorker方法时会自动检查runState和workerCount,从而通过返回false来防止在不应该添加线程的情况下添加线程.
    2. 如果一个任务可以进入队列,还需要再次检查是否可以添加一个线程
    3. 如果一个任务进入队列失败,则尝试创建一个新的线程,如果创建线程也失败了,则会执行RejectedExecutionHandler拒绝策略.

    addWorker(Runnable,boolean)

    /**
 * @param 线程对象
 *
 * @param 如果为true,则创建核心线程,如果为false,则创建非核心线程
 * @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.
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;

        for (;;) {
            int wc = workerCountOf(c);
            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
        }
    }

    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;
            mainLock.lock();
            try {
                // Recheck while holding lock.
                // Back out on ThreadFactory failure or if
                // shut down before lock acquired.
                int rs = runStateOf(ctl.get());

                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    //添加到worker集合中,也就时放入线程池中
                    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)
            //worker启动未成功,则回滚,包括worker计数-1,从worker集合中移除当前worker
            addWorkerFailed(w);
    }
    return workerStarted;
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75

    该方法的主要作用就是根据当前线程池的状态和容量用给定的线程添加到worker中.如果添加成功,则调整worker的计数,并且可能会启动一个新的Worker,将Runnable作为其第一个任务执行.如果线程池已停止或者符合关闭条件,则直接返回fasle,如果线程工厂返回null,或者线程创建失败,或者由于异常导致创建失败,则会回滚.
    runWorker(Worker)

    final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        //while循环一直去获取任务：getTask()方法是一个核心方法
        //getTask()方法也是实现了核心线程复用的关键代码
        while (task != null || (task = getTask()) != null) {
            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();
            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;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        //根据任务处理成功与否，处理worker退出逻辑
        processWorkerExit(w, completedAbruptly);
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49

    该方法主要时循环从队列中获取任务并执行它们,同时处理一些其他的问题.

    1. 我们可能启动的是第一个任务,此时我们不需要从池里获取任务,如果不是第一个任务,我就需要通过getTask()方法从池中获取任务,如果返回为null，则执行worker退出逻辑。
    2. 在运行任务之前，首先获取锁，防止任务在执行过程中发生其他池中断，然后确保除非池停止，否则不会为线程设置中断。
    3. 每个任务调用之前都会执行beforeExecute方法，该方法如果抛出异常，则线程会终止。
    4. 假设beforeExecute方法正常完成，我们会捕获任务执行中抛出的异常，并且将异常传递给afterExecute方法。任何异常也会导致任务的终止。
    5. 任务运行完成以后，我们调用afterExecute方法，该方法抛出异常也会导致线程的终止。

    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);

        // allowCoreThreadTimeOut=true：核心线程也会被回收
        //当前运行的线程已经超过了核心线程数量
        //**核心线程复用关键代码一**
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }

        try {
            //**核心线程复用关键代码二**
            Runnable r = timed ?
                //非核心线程像队列中获取任务的时候超时返回null，则会当前worker会被回收
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                //核心线程在任务队列中没有任务的时候会阻塞在这里
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42

    该方法主要作用就是根据当前线程池的配置，对任务进行阻塞或者超时等待**(该方法最核心的功能就是核心线程复用)**。如果worker必须退出，则返回null。worker退出条件如下:

    1. 当前线程池的线程数量超过了maximumPoolSize
    2. 线程池已停止
    3. 线程池已关闭，并且队列为空
    4. worker从队列中获取任务超时，超时后的共工作线程将会被终止，如果队列非空，则该worker线程不是线程池中最后一个线程。
  • 相关阅读:
    淘宝商品API使用示例：如何通过调用外部API来获取淘宝商品价格销量主图详情数据
    相信中国杂交水稻技术 国稻种芯：中菲农业创繁荣发展时代
    武汉新时标文化传媒有限公司抖音小店如何提高出单量？
    ElasticSearch学习
    C语言-atoi函数的使用和实现
    【Android抓包】- Stetho
    day007--MySQL中的数值函数
    AcWing第 68 场周赛题解
    测试设计场景题
    Cobalt Strike（三）DNS bacon 的使用与原理
  • 原文地址：https://blog.csdn.net/Suubyy/article/details/127826758