ThreadPoolExecutor-线程池如何保证线程不被销毁_executors.newfixedthreadpool线程执行完自动释放吗

目录

前言

线程池使用入口

源码

结论

---

前言

1、通常情况下 我们new一个线程执行任务，任务执行完之后线程也随之销毁了

2、为了减少创建线程的开销，使线程可以复用，我们使用线程池

3、那么问题来了，线程池是如何保证池子里的线程执行完不被销毁的呢？

线程池使用入口

入口：我们使用线程池时，代码如下

Executors.newFixedThreadPool(5);

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

或者

ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5,10,60L,TimeUnit.SECONDS,new LinkedBlockingDeque<Runnable>());
threadPoolExecutor.execute(new Runnable() {
    @Override
    public void run() {
        //business code..
    }
});

源码

所以我们从execute这个方法开始探寻

public void execute(Runnable command) {
    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.
     */
    //clt保存了两种状态 当前线程数workerCountOf(c) 和 线程池的状态 runStateOf(c)
    int c = ctl.get();
    //如果当前线程数 < 核心线程数 直接增加一个线程
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    //如果线程池状态是RUNNING 且 阻塞队列可以继续添加任务
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        //如果线程池状态不是RUNNING 且 成功移除任务
        if (! isRunning(recheck) && remove(command))
            //拒绝执行当前任务
            reject(command);
        //当前线程数 == 0 
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    //执行任务失败 - 拒绝执行
    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);
 
        // 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;
            //用CAS算法 将当前线程数加1，并跳出循环 继续执行
            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是ThreadPoolExecutor的私有内部类，实现了Runnable接口
        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());
                //如果线程池状态正常 或者 是SHUTDOWN且任务是空的，继续执行
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    //线程的alive是指线程已经开始执行并且没有销毁，所以如果是alive则抛异常
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    //workers是线程池定义的存放Worker的集合 HashSet<Worker> workers
                    workers.add(w);
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            //如果前边没问题 则开始执行线程run方法
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

接下来我们需要看一下Worker类和它的run方法

Worker(Runnable firstTask) {
    setState(-1); // inhibit interrupts until runWorker
    this.firstTask = firstTask;
    this.thread = getThreadFactory().newThread(this);
}

构造方法中 创建了一个线程

public void run() {
    runWorker(this);
}

然后进入runWorker方法

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        //如果task不为空 或者 getTask不为空时 会一直循环
        //有任务时肯定不为空，没任务时task=null，所以要维持while永真，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 继续走getTask的逻辑
                task = null;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly);
    }
}

getTask 代码如下：

/**
 * Performs blocking or timed wait for a task, depending on
 * current configuration settings, or returns null if this worker
 * must exit because of any of:
 * 1. There are more than maximumPoolSize workers (due to
 *    a call to setMaximumPoolSize).
 * 2. The pool is stopped.
 * 3. The pool is shutdown and the queue is empty.
 * 4. This worker timed out waiting for a task, and timed-out
 *    workers are subject to termination (that is,
 *    {@code allowCoreThreadTimeOut || workerCount > corePoolSize})
 *    both before and after the timed wait, and if the queue is
 *    non-empty, this worker is not the last thread in the pool.
 *
 * @return task, or null if the worker must exit, in which case
 *         workerCount is decremented
 */
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.
        //如果线程池状态为STOP/TIDYING/TERMINATED 或者 线程池状态为SHUTDOWN/STOP/TIDYING/TERMINATED且阻塞队列为空
        //return null 不再维持线程
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            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())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }
 
        try {
            //重点：poll会一直阻塞直到超过keepAliveTime或者获取到任务
            //take 会一直阻塞直到获取到任务
            //在没有任务的时候 如果没有特别设置allowCoreThreadTimeOut，我们的核心线程会一直阻塞在这里
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

真相大白~

结论

总结就是：如果队列中没有任务时，小于核心数的线程会一直阻塞在获取任务的方法，直到返回任务。（判断阻塞时并没有核心线程和非核心线程的概念）

而且执行完后 会继续循环执行getTask的逻辑

附take poll源码注释：

/**
 * Retrieves and removes the head of this queue, waiting if necessary
 * until an element becomes available.
 *
 * @return the head of this queue
 * @throws InterruptedException if interrupted while waiting
 */
E take() throws InterruptedException;
 
/**
 * Retrieves and removes the head of this queue, waiting up to the
 * specified wait time if necessary for an element to become available.
 *
 * @param timeout how long to wait before giving up, in units of
 *        {@code unit}
 * @param unit a {@code TimeUnit} determining how to interpret the
 *        {@code timeout} parameter
 * @return the head of this queue, or {@code null} if the
 *         specified waiting time elapses before an element is available
 * @throws InterruptedException if interrupted while waiting
 */
E poll(long timeout, TimeUnit unit)
    throws InterruptedException;

LinkedBlockingQueue实现的take方法：

这里有一个小问题，这里的while可不可以用if代替呢？

正常来说，每次signal时，只会有一个线程被唤醒，用if好像也没有问题，但是，实际上存在虚假唤醒的问题，即队列没有元素但是线程被唤醒了，需要while保证准确性。

所以，不能

2021-02-07 新增线程池结构图