一文带你了解线程池原理

时间:2023-02-24 18:14:24

1.使用线程池的意义何在?

​ 项目开发中,为了统一管理线程,并有效精准地进行排错,我们经常要求项目人员统一使用线程池去创建线程。因为我们是在受不了有些人动不动就去创建一个线程,使用的多了以后,一旦报错就只有一个线程报错信息,还是线程的共用信息,再加上如果你将异常吃了(捕获后不做处理)的情况下,这个错误。。。。em,我实在不知道去哪里排查,不然你换个人试试吧。

2.线程池的重要参数----你真的了解吗

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
  1. corePoolSize:核心线程数。设置核心线程数的意义何在?通俗来讲核心线程数就是正式员工,需要长期坚守岗位,有任务就需要执行。
  2. maximumPoolSize:最大线程池个数。设置最大线程池数量的意义何在?其实就是一个容错机制,当你的需要执行的线程个数已经爆满并且超过的时候,提供了一个容错机制,可以保证在短期内多余的任务正常执行。相当于就是临时工,临时过来执行任务,任务结束后就可以走了。
  3. keepAliveTime:保活的时间。设置的意义何在?当线程任务无剧增的情况下,维持在正常提亮。你无需那么多临时工来执行任务,所以规定时间,临时工可以走人了,也即是除核心线程外的线程可以回收了。
  4. TimeUnit:保活的时间单位。这个就不多赘述了。
  5. BlockingQueue:阻塞队列。设置阻塞队列的意义何在?当所有核心线程都正在工作时,将其放入阻塞队列,等待后续执行。也就是这个任务进行排队,等正式工忙完了继续做。
  6. ThreadFactory:线程工厂。生产线程,由你自己去定义你想生产什么样的线程。
  7. RejectedExecutionHandler:拒绝策略。当你的最大线程与阻塞队列都满了。这个时候,你已经接收不了新的任务进行处理了。所以设置拒绝策略。相当于就是我所有的员工和临时工都在工作了,并且排队的任务都满了,应对这样的情况,你打算如何做。

除此之外还有一个重要的参数:

    /**
     * If false (default), core threads stay alive even when idle.
     * If true, core threads use keepAliveTime to time out waiting
     * for work.
     */
    private volatile boolean allowCoreThreadTimeOut;//是否允许核心线程数超时退出。

该参数有在特定的业务场景下有很大的意义。比如:你的业务只在晚上需要执行,其余时间无需执行。那么为何不把资源让出来,白天的时候,可以让其他业务占有这些资源去执行呢。

3.ThreadExecutorPool线程池重要源码解析

一文带你了解线程池原理

由该类图可知,Executor执行器定义执行方法,ExecutorService定义线程池操作的基本方法,AbstractExecutorService定义了线程池操作的方法模板。

ThreadPoolExecutor任务执行流程图

一文带你了解线程池原理

1.首先是构造方法

​ 基本的参数校验与赋值,简单代码不过多赘述。

    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.acc = System.getSecurityManager() == null ?
                null :
                AccessController.getContext();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

2.线程执行的方法

    public Future<?> submit(Runnable task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<Void> ftask = newTaskFor(task, null);//将线程对象封装成RunnableFuture
        execute(ftask);//任务执行
        return ftask;
    }
    public <T> Future<T> submit(Callable<T> task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task);//将线程对象封装成RunnableFuture
        execute(ftask);//任务执行
        return ftask;
    }
    public <T> Future<T> submit(Runnable task, T result) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task, result);//将线程对象封装成RunnableFuture
        execute(ftask);//任务执行
        return ftask;
    }
 public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        int c = ctl.get();//获取当前的线程池状态。单个参数,保存了线程池的状态以及线程数量
        if (workerCountOf(c) < corePoolSize) { //当线程数量小于核心线程数
            if (addWorker(command, true)) //直接添加任务,运行线程
                return;
            c = ctl.get();
        }
        if (isRunning(c) && workQueue.offer(command)) {//如果核心线程数已经满了,那么直接添加到阻塞队列。
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))//线程池不是running状态,执行拒绝策略。
                reject(command);
            else if (workerCountOf(recheck) == 0)//线程池线程数量不能为0,需要有一个线程对线程池的后续操作进行处理,比如关闭线程池
                addWorker(null, false);
        }
        else if (!addWorker(command, false))//当核心线程与阻塞队列都满了的时候,直接添加任务到非核心线程运行。添加失败直接执行拒绝策略
            reject(command);
    }

1.关于ctl.get()方法的解释---利用了单个变量,保存了线程池状态以及线程数量的值

 private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    private static final int COUNT_BITS = Integer.SIZE - 3;
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
    // runState is stored in the high-order bits
    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; //线程池已经关闭
    // Packing and unpacking ctl
    private static int runStateOf(int c)     { return c & ~CAPACITY; }
    private static int workerCountOf(int c)  { return c & CAPACITY; }
    private static int ctlOf(int rs, int wc) { return rs | wc; }

2.addWorker方法

 private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            int c = ctl.get();//获取ctl的快照保存在栈上
            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))//CAS修改线程池ctl变量,增加线程数
                    break retry; //添加成功直接退出
                c = ctl.get();  // 添加不成功,为了保证多线程运行的安全性,重新获取
                if (runStateOf(c) != rs)//当前线程池状态发生改变
                    continue retry; //直接重新运行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); //生成自定义的线程woker
            final Thread t = w.thread;
            if (t != null) {
                final ReentrantLock mainLock = this.mainLock;//这个代码没有意义,mainLock定义的变量为final。可以直接使用
                mainLock.lock();//添加work使用锁,保证添加任务的原子性。
                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 || //线程池处于running状态
                        (rs == SHUTDOWN && firstTask == null)) {//线程池处于showdown状态但是firstTask为空。
                        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);//移除work,减少线程数量
        }
        return workerStarted;
    }

t.start()执行线程任务

//Worker类中实际执行任务的方法 
public void run() {
            runWorker(this);
        }
final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts  //将原始的线程状态为-1修改为0,后续通过getState()>=0获取线程是否已经运行的状态,允许线程中断。-1默认为初始化,此处需要进行处理
        boolean completedAbruptly = true;
        try {
            while (task != null || (task = getTask()) != null) {//task不等于空直接运行,task等于空从workerQueue阻塞队列获取任务
                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) ||//线程池运行状态大于等于STOP
                     (Thread.interrupted() && //线程是否已经被中断了
                      runStateAtLeast(ctl.get(), STOP))) &&//鲜橙汁运行状态大于等于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++;//任务完成数加1
                    w.unlock();
                }
            }
            completedAbruptly = false;//执行过程中是否发成异常
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }

//执行任务退出操作
private void processWorkerExit(Worker w, boolean completedAbruptly) {
        if (completedAbruptly) // 如果有异常中断导致任务结束
            decrementWorkerCount();//将线程数量减1

        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            completedTaskCount += w.completedTasks;//完成的任务数量累加
            workers.remove(w);//从workers的任务集合中移除当前任务
        } finally {
            mainLock.unlock();
        }

        tryTerminate();//尝试关闭线程池

        int c = ctl.get();//获取当前线程池的最新状态
        if (runStateLessThan(c, STOP)) {//如果当前任务状态小于STOP
            if (!completedAbruptly) {//当前任务执行无异常发生
                int min = allowCoreThreadTimeOut ? 0 : corePoolSize;//根据allowCoreThreadTimeOut参数获取最小的线程数量
                if (min == 0 && ! workQueue.isEmpty())//如果核心线程允许退出,并且工作队列不为空
                    min = 1;//设置最小值为1,因为最后需要有线程去执行线程池的后续处理,所有线程都没了,后续线程池退出无线程处理
                if (workerCountOf(c) >= min)//如果工作的线程数量大于等最小值
                    return; // replacement not needed  直接返回
            }
            addWorker(null, false);//如果当前线程数已经小于最小线程数,那么需要保证最小线程数在运行,所以需要有保证线程池的正常运行,添加一个空任务。
        }
    }
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())) {//如果线程池状态大于等于SHUTDOWN并且(线程数量大于等于STOP或者工作队列为空)
                decrementWorkerCount();//将线程池中线程数量减1
                return null;
            }

            int wc = workerCountOf(c);//获取当前线程池的线程数量

            // Are workers subject to culling?
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;//判断是否运行核心线程数超时,判断是否需要超时机制

            if ((wc > maximumPoolSize || (timed && timedOut))//工作线程大于最大线程池数量或者允许超时并且有超时的情况
                && (wc > 1 || workQueue.isEmpty())) {//并且线程池线程数量大于1或者阻塞队列为空
                if (compareAndDecrementWorkerCount(c))//CAS操作将线程池数量减1
                    return null;//返回空
                continue;//CAS失败继续
            }

            try {
                Runnable r = timed ?//允许超时从队列中拿任务并等待keepAliveTime时间
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();阻塞等待
                if (r != null)//获取的任务不为空
                    return r;//直接返回
                timedOut = true;//如果为空,超时标志位为true
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }

3.addWorkerFailed方法解析

private void addWorkerFailed(Worker w) {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();//获取锁
        try {
            if (w != null)//work不是空
                workers.remove(w);//直接从workers中移除当前任务
            decrementWorkerCount();//加个ctl中的woker数量减少
            tryTerminate();//如果线程池已经是showdown状态,尝试让线程池停止。多线程协作的函数
        } finally {
            mainLock.unlock();
        }
    }

3.线程池关闭shutdown方法

    public void shutdown() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess();//检查关闭权限,可以忽略
            advanceRunState(SHUTDOWN);//线程池状态递进,由running变为shutdown
            interruptIdleWorkers();//中断所有空闲线程
            onShutdown(); // hook for ScheduledThreadPoolExecutor钩子函数,调度线程池使用
        } finally {
            mainLock.unlock();
        }
        tryTerminate();//尝试将线程池关闭。
    }

1.advanceRunState方法解析

    private void advanceRunState(int targetState) {
        for (;;) {
            int c = ctl.get();//获取当前的线程状态
            if (runStateAtLeast(c, targetState) ||//当前状态已经是大于等于shutdown直接退出
                ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))//cas操作将线程状态改为targetState。
                break;
        }
    }

2.interruptIdleWorkers方法解析

    private void interruptIdleWorkers() {
        interruptIdleWorkers(false);
    }

    private void interruptIdleWorkers(boolean onlyOne) {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();//获取锁
        try {
            for (Worker w : workers) {//遍历works中所有的工作任务
                Thread t = w.thread;
                if (!t.isInterrupted() && w.tryLock()) {//如果没有被中断过,并且可以获得锁,证明属于空闲线程
                    try {
                        t.interrupt();//将线程中断,打上中断标志位
                    } catch (SecurityException ignore) {
                    } finally {
                        w.unlock();//解锁
                    }
                }
                if (onlyOne)//只中断一个线程标识
                    break;
            }
        } finally {
            mainLock.unlock();
        }
    }

4.shutdownNow方法解析

    public List<Runnable> shutdownNow() {
        List<Runnable> tasks;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess();//权限检查
            advanceRunState(STOP);//状态递进 详细方法见上面
            interruptWorkers();//中断所有启动的work线程
            tasks = drainQueue();//将所有未执行的任务出队保存
        } finally {
            mainLock.unlock();
        }
        tryTerminate();//尝试关闭线程池
        return tasks;
    }

1.interruptWorkers方法解析

    private void interruptWorkers() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();//获取锁
        try {
            for (Worker w : workers)//遍历所有woker进行处理
                w.interruptIfStarted();
        } finally {
            mainLock.unlock();
        }
    }


        void interruptIfStarted() {
            Thread t;
            if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {//当前work的状态大于0并且线程不为空且线程未被中断
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                }
            }
        }

使用getState() >= 0表示当前线程已经启动,runWorker方法中会将其状态从-1改变。证明线程已经启动
       Worker(Runnable firstTask) {
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }

2.drainQueue方法解析

//标准的入队和出队功能不做过多注释   
private List<Runnable> drainQueue() {
        BlockingQueue<Runnable> q = workQueue;
        ArrayList<Runnable> taskList = new ArrayList<Runnable>();
        q.drainTo(taskList);
        if (!q.isEmpty()) {
            for (Runnable r : q.toArray(new Runnable[0])) {
                if (q.remove(r))
                    taskList.add(r);
            }
        }
        return taskList;
    }
	

5.tryTerminate方法解析

final void tryTerminate() {
    for (;;) {
        int c = ctl.get();//获取当前线程状态ctl
        if (isRunning(c) ||//线程池正在运行
            runStateAtLeast(c, TIDYING) ||//线程池状态大于等于TIDYING,有其他线程已经改变线程池状态为TIDYING或者TERMINATED了
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))//线程池状态等于shutdown并且工作队列不为空。
            return;//以上三种情况线程池无法关闭,需要继续处理
        if (workerCountOf(c) != 0) { // Eligible to terminate//当前工作线程数量不等于0
            interruptIdleWorkers(ONLY_ONE);//中断线程且只中断一个
            return;
        }

        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {//cas操作将线程池状态置为TIDYING
                try {
                    terminated();//线程池终止
                } finally {
                    ctl.set(ctlOf(TERMINATED, 0));//设置线程池状态为TERMINATED
                    termination.signalAll();//信号唤醒所有等待线程
                }
                return;
            }
        } finally {
            mainLock.unlock();
        }
        // else retry on failed CAS
    }
}

4.总结

​ 线程池的运用在项目中已经成为一种常态,作为一个开发人员最重要的了解其背后的设计原理以及流程,更好地运用线程池,方便提升项目程序的性能以及排查错误。在阅读对应的线程池源码时,我们只局限于单线程的思维,更多的是要去考虑当多线程并发执行时的临界条件。了解设计者的设计初衷、以及设计意图,能让你更好地在项目中运用并设计符合自己项目的线程池。以上是我个人对于线程池ThreadPoolExecutor的理解,不足之处,请多多指教。