线程池ThreadPoolExecutor

作者: ChadJ | 来源:发表于2023-09-24 11:24 被阅读0次

一、线程池状态

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; // 111
private static final int SHUTDOWN   =  0 << COUNT_BITS; // 000
private static final int STOP       =  1 << COUNT_BITS; // 001
private static final int TIDYING    =  2 << COUNT_BITS; // 010
private static final int TERMINATED =  3 << COUNT_BITS; // 011

// 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; }

RUNNING：正常运行。可接收新任务，可处理队列中任务。
SHUTDOWN：关闭。不可接收新任务，可处理队列中任务。
STOP：停止。不接收新任务，不处理队列中任务，并且会中断正在处理的任务。
TIDYING：整理。所有任务已终止workCount为0，将执行terminated()钩子方法。在ThreadPoolExecutor中可以对terminated()进行重载。
TERMINATED：彻底结束。terminated()方法执行完成。

graph LR;
RUNNING(RUNNING) --shutdown--> SHUTDOWN(SHUTDOWN)
RUNNING --shutdownNow--> STOP(STOP)
SHUTDOWN --> |阻塞队列为空,执行任务为空| TIDYING(TIDYING)
STOP -->|线程池中执行任务为空| TIDYING
TIDYING --terminated--> TERMINATED(TERMINATED)

ctl 值解析

二进制前3位表示线程池状态，后29位表示工作线程数量。

二、创建线程池

线程池创建主要通过构造函数，有7个参数。

corePoolSize：核心线程数。
maximumPoolSize：最大线程数。
keepAliveTime：线程存活时间。
unit：时间单位。
workQueue：阻塞队列。
threadFactory：线程工厂。
handler：拒绝策略，四种。
1. AbortPolicy：直接抛出异常，默认策略；
2. CallerRunsPolicy：回退给调用者所在的线程来执行任务；
3. DiscardOldestPolicy：丢弃阻塞队列中靠最前的任务，并执行当前任务；
4. DiscardPolicy：直接丢弃任务；
5. 当然我们也可以实现自己的拒绝策略，实现RejectedExecutionHandler接口即可。

三、终止线程池

线程池ThreadPoolExecutor提供了shutdown()和shutDownNow()两种方式关闭线程池。

shutdown()：按过去执行已提交任务的顺序发起一个有序的关闭，但是不接受新任务。
shutdownNow() :尝试停止所有的活动执行任务、暂停等待任务的处理，并返回等待执行的任务列表。

四、核心源码阅读

1. `execute()`提交任务

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)) {
            // double check 线程池状态
        int recheck = ctl.get();
        // 线程池不是running则回滚队列
        if (! isRunning(recheck) && remove(command))
            reject(command);
        else if (workerCountOf(recheck) == 0)
                // 没有工作线程，则创建线程
            addWorker(null, false);
    }
    // 尝试将当任务程加入非核心线程
    else if (!addWorker(command, false))
            // 拒绝当前任务
        reject(command);
}

总结如下：

首先判断工作线程少于核心线程，则尝试将当前任务添加为核心线程，添加成功就返回；
未能成功添加核心线程，看线程池状态如果是RUNNING，则尝试将任务加入等待队列；

如果加入成功，则需要对当前任务负责，进行double check是为了保证已经加入队列的任务要么等待线程来处理它，要么回退拒绝。
1. 再次检查线程池状态，如果发现线程池不是running了，则回退任务执行拒绝策略；
2. 再次检查线程池内工作线程数量，没有工作线程则创建工作线程，用于执行队列中的剩余任务；
未能成功加入等待队列，则尝试将当前任务加入非核心线程，失败则执行拒绝策略；

2. `addWorker()`添加工作线程

private boolean addWorker(Runnable firstTask, boolean core) {
  
    // 1. 尝试通过CAS增加工作线程数量
    retry:
    for (;;) {
            
        int c = ctl.get();
        // 线程池状态
        int rs = runStateOf(c);
        // 非RUNNING：SHUTDOWN,STOP,TIDYING,TERMINATED
        // SHUTDOWN状态不允许添加新任务，只能处理队列中已有任务。
        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增加线程数量
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();
            // 线程池状态改变，则去外层循环重新检测状态
            if (runStateOf(c) != rs)
                continue retry;
        }
    }

    // 2. 创建工作线程，创建失败则回滚。
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        // 新建worker线程，通过firstTask
        w = new Worker(firstTask);
        // worker中通过工厂创建的线程，当前线程
        final Thread t = w.thread;
        // 线程创建成功
        if (t != null) {
            // 获取主锁
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                // 线程池状态
                int rs = runStateOf(ctl.get());
                // running || (shutdown && firstTask == null)
                // running允许添加任务，shutdown不允许添加任务（队列中可能有未执行任务，创建没有firstTask工作线程处理）
                if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) {
                    // 当前线程已经启动，抛出异常
                    if (t.isAlive())
                        throw new IllegalThreadStateException();
                    // private final HashSet<Worker> workers = new HashSet<Worker>();
                    workers.add(w);
                    // 修改最大线程数量和线程添加成功变量
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
           // 线程添加成功，则启动线程
            if (workerAdded) {
                // 启动，执行worker的run方法
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            // 未能成功启动，则走失败回滚
            addWorkerFailed(w);
    }
    return workerStarted;
}

添加工作线程比较简单，首先是通过CAS对线程数量进行增加。然后尝试创建线程，如未能成功创建则走回滚方法。

首先要判断线程线程池状态，如果是非RUNNING状态，则不能进行添加线程。这里要注意SHUTDOWN状态的时候，不能添加新任务，但是可以对阻塞队列中已有的任务进行处理。是通过这段代码判断的(rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty()))。

所以，只有两种情况能创建工作线程，即：线程池RUNNING状态，线程池SHUTDOWN状态并且没有firstTask的时候。

后者创建出的工作线程，用于处理阻塞队列中的未执行的任务，但是此时已经不接收新任务了。与excutor方法中的double check处代码对应。

在executor()方法中新建核心线程的地方有3处：

核心线程数量不够时，创建核心线程；
把当前任务加入队列后，未能从队列中回滚。创建一个没有firstTask的工作线程；
不是RUNNING或未能加入到阻塞队列，则创建非核心线程；

3. `addWorkerFailed()`添加失败回滚

private void addWorkerFailed(Worker w) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        // removes worker from workers, if present
        if (w != null)
            workers.remove(w);
        // decrements worker count
        decrementWorkerCount();
        // rechecks for termination, in case the existence of this worker was holding up termination
        tryTerminate();
    } finally {
        mainLock.unlock();
    }
}

如果添加失败则需要回滚：

如果当前work存在，则删除。
把在尝试创建时添加的工作线程数量减下去。
尝试终止当前工作线程。

4. `Worker` 工作线程

private final class Worker extends AbstractQueuedSynchronizer implements Runnable {

    private static final long serialVersionUID = 6138294804551838833L;

    // 工厂创建出的线程，失败时候为null
    final Thread thread;
    // 将要运行的第一个任务，可能为null
    Runnable firstTask;
    // 完成任务计数器
    volatile long completedTasks;

    Worker(Runnable firstTask) {
        // 设置AQS的同步状态private volatile int state，是一个计数器，大于0代表锁已经被获取
        setState(-1);
        this.firstTask = firstTask;
        // 利用工厂创建线程
        this.thread = getThreadFactory().newThread(this);
    }

    // 任务执行
    public void run() {
        runWorker(this);
    }

    // Lock methods 省略……
    //
    // The value 0 represents the unlocked state.
    // The value 1 represents the locked state.
}

继承了AQS实现了Runnable，即能释放锁获取锁，又是一个可执行的任务。线程启动后执行run()方法，其实就是调用的外部runWorker方法。

5. `runWorker()` 任务执行方法

final void runWorker(Worker w) {
    // 当前线程，即当前worker
    Thread wt = Thread.currentThread();
    // 要执行的第一个任务，可能为null
    Runnable task = w.firstTask;
    w.firstTask = null;
    // 主要是为了把添加工作线程时的setState(-1)改为0，释放锁。
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
  
    try {
        // 一直在获取要执行的任务，getTask从队列中获取。
        while (task != null || (task = getTask()) != null) {
            // worker 获取锁，要进行任务处理了
            w.lock();
            // 确保线程池STOP时线程中断，非STOP则清除中断。
            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);
    }
}

首先根据worker获取要执行的task，把添加工作线程时的setState(-1)改为0，释放锁。
然后循环从阻塞队列中获取任务，并执行任务的run方法。
在执行任务的前后分别有两个自定义方法，可供拓展。

6. `getTask()`队列中获取任务

private Runnable getTask() {
      // 超时，用于关闭非核心线程
    boolean timedOut = false; // Did the last poll() time out?

    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        // 检测线程池状态，非RUNNING状态。SHUTDOWN并且队列中没有任务
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            // 减掉工作线程
            decrementWorkerCount();
            return null;
        }

        // 判断工作线程数量是否已经大于核心线程，大于核心线程则要关心存活时间
        int wc = workerCountOf(c);
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

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

        try {
            // 从队列中取任务，非核心线程则需要有超时时间。
            Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

主要就是在阻塞队列中取待处理的任务，当线程池中工作线程多于核心线程时，利用队列的poll获取，未达到核心数则直接take。

take方法没有任务时，则会一直阻塞挂起等待，直到有任务加入时唤起该线程，返回任务。

poll方法则是在规定时间范围内获取，如果未获取到则返回null。

7. `processWorkerExit()`工作线程退出

private void processWorkerExit(Worker w, boolean completedAbruptly) {
    // true：用户线程运行异常,需要扣减线程数
    // false：正常退出的，在getTask方法中扣减的线程数量
    if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
        decrementWorkerCount();

    // 获取锁
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        // 完成任务数统计
        completedTaskCount += w.completedTasks;
        // 集合中移除工作线程
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }
    // 有worker线程移除，可能是最后一个线程退出，需要尝试终止线程池。
    tryTerminate();

    int c = ctl.get();
    // RUNNGING 和 SHUTDOWN
    if (runStateLessThan(c, STOP)) {
        // 正常退出，计算需要维护的最小线程数
        if (!completedAbruptly) {
            // 核心线程，是否超时设置。默认是否，0说明最小线程数可为0；
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            // 确保当阻塞队列不为空时，至少有1个线程来处理剩余任务。
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            // 当前工作线程数量满足最小要求，返回。
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        // 不是正常退出，则添加一个非核心线程。用来处理剩余任务
        addWorker(null, false);
    }
}

通过completedAbruptly判断是否是正常退出。如果是正常退出的则不需要进行线程数量减法操作，因为在getTask中已经对线程数量做减法了；如果是非正常退出，则需要对线程数量做减法，然后在后续的逻辑判断中根据实际情况看是否还需要再创建新线程。无论是否是正常退出，都需要把当前工作线程从工作集合中移出。然后尝试终止线程池，因为有可能是最后一个工作线程退出。

最后，检查线程池的状态如果是RUNNGING或SHUTDOWN，则需要执行兜底补偿措施。确保能有工作线程来处理阻塞队列中剩余的任务。

8. `tryTerminate()`尝试关闭线程池

final void tryTerminate() {
    for (;;) {
        int c = ctl.get();
        // RUNNGING
        // TIDYING 或 TERMINATED
        // SHUTDOWN 且 队列不为空
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
        
        // 到这里只有2种情况：STOP，SHUTDOWN & 队列为空
        // 如果线程池中还有工作线程，则尝试中断
        if (workerCountOf(c) != 0) { // Eligible to terminate
            // STOP和线程池内还有工作线程，但是任务已经没有了。需要中断等待任务的线程。
            interruptIdleWorkers(ONLY_ONE);
            return;
        }
         
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            // CAS设置线程池为TIDYING状态
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                    // 终止线程池
                    terminated();
                } finally {
                    // 线程池状态设置为 TERMINATED
                    ctl.set(ctlOf(TERMINATED, 0));
                    termination.signalAll();
                }
                return;
            }
        } finally {
            mainLock.unlock();
        }
        // else retry on failed CAS
    }
}

当线程池涉及到要移除worker工作线程时都会调用此方法来判断是否需要结束当前线程池。只有STOP和SHUTDOWN队列为空时才会尝试执行结束，并且要保证工作线程数为0。通过CAS首先改变线程池状态到TIDYING，然后调用terminated()方法来结束线程池，此方法可以拓展。最后将线程池状态改为TERMINATED。

9. `interruptIdleWorkers()`中断空闲线程

// onlyONe 说明只终止一个线程，否则终止全部。
private void interruptIdleWorkers(boolean onlyOne) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (Worker w : workers) {
            Thread t = w.thread;
            // 尝试对worker中断加锁。
            if (!t.isInterrupted() && w.tryLock()) {
                try {
                    // 对线程中断
                    t.interrupt();
                } catch (SecurityException ignore) {
                } finally {
                    w.unlock();
                }
            }
            if (onlyOne)
                break;
        }
    } finally {
        mainLock.unlock();
    }
}

对线程进行中断，对工作线程加锁。

线程池ThreadPoolExecutor

一、线程池状态

ctl 值解析

二、创建线程池

三、终止线程池

四、核心源码阅读

1. `execute()`提交任务

2. `addWorker()`添加工作线程

3. `addWorkerFailed()`添加失败回滚

4. `Worker` 工作线程

5. `runWorker()` 任务执行方法

6. `getTask()`队列中获取任务

7. `processWorkerExit()`工作线程退出

8. `tryTerminate()`尝试关闭线程池

9. `interruptIdleWorkers()`中断空闲线程

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

线程池ThreadPoolExecutor

一、线程池状态

ctl 值解析

二、创建线程池

三、终止线程池

四、核心源码阅读

1. execute()提交任务

2. addWorker()添加工作线程

3. addWorkerFailed()添加失败回滚

4. Worker 工作线程

5. runWorker() 任务执行方法

6. getTask()队列中获取任务

7. processWorkerExit()工作线程退出

8. tryTerminate()尝试关闭线程池

9. interruptIdleWorkers()中断空闲线程

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

1. `execute()`提交任务

2. `addWorker()`添加工作线程

3. `addWorkerFailed()`添加失败回滚

4. `Worker` 工作线程

5. `runWorker()` 任务执行方法

6. `getTask()`队列中获取任务

7. `processWorkerExit()`工作线程退出

8. `tryTerminate()`尝试关闭线程池

9. `interruptIdleWorkers()`中断空闲线程