前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >ThreadPoolExecutor源码分析

ThreadPoolExecutor源码分析

原创
作者头像
冰枫
发布2018-04-16 15:32:16
9996
发布2018-04-16 15:32:16
举报
文章被收录于专栏:冰枫冰枫

ThreadPoolExecutor继承AbstractExecutorService,层级实现了ExecutorService,ExecutorService继承了Executor

1 . Executor Executor提供了一种解耦任务提交与任务如何运行的 execute(Runnable command)方法,例如创建线程并启动是这样的:

代码语言:javascript
复制
new Thread(()->{}).start();

而使用execute(Runnable command) 方法后:

代码语言:javascript
复制
Executor executor = anExecutor; 
executor.execute(new Runnable(()->{}));
class MyExecutor implements Executor{
    public void execute(Runnable command){
        new Thread(command).start();    
    }
}

2 . ExecutorService

ExecutorService提供了管理终结,并且生成追踪一个或多个异步任务的Future的方法。

代码语言:javascript
复制
package java.util.concurrent;
import java.util.List;
import java.util.Collection;
public interface ExecutorService extends Executor {
    void shutdown();
    List<Runnable> shutdownNow();
    boolean isShutdown();
    boolean isTerminated();
    boolean awaitTermination(long timeout, TimeUnit unit)
        throws InterruptedException;
    <T> Future<T> submit(Callable<T> task);
    <T> Future<T> submit(Runnable task, T result);
    Future<?> submit(Runnable task);
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException;
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                  long timeout, TimeUnit unit)
        throws InterruptedException;
    <T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException;
    <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                    long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

Future提供了取消任务,以及获取任务的执行状态等方法。例如:

代码语言:javascript
复制
ExecutorService service = anExecutorService;
Future future = service.submit(new Runnable(()->{}));
future.cancel();
future.isDone();

3 . AbstractExecutorService

AbstractExectorService则是提供了ExectorService的默认实现,这里不详细描述了

4 . ThreadPoolExecutor

字段分析:

ctl 的高位存储线程的运行状态,低位存储工作线程的数量,可见线程刚创建时线程状态为运行状态,工作线程数为0

COUNT_BITS用于决定线程状态移动的位数

CAPACITY用于计算线程运行状态与工作线程数量

代码语言:javascript
复制
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;
代码语言:javascript
复制
private final BlockingQueue<Runnable> workQueue;//阻塞队列
    private final ReentrantLock mainLock = new ReentrantLock(); //可重入锁
    private final HashSet<Worker> workers = new HashSet<Worker>(); //用来存放工作线程的Set集合
    private final Condition termination = mainLock.newCondition();
    private int largestPoolSize; // 用来记录线程池最大时候的大小(线程池监控)
    private long completedTaskCount; // 任务完成的数量(线程池监控)
    private volatile ThreadFactory threadFactory; // 顾名思义线程创建工厂,线程池在创建线程时会调用它的newThread(Runnable r)方法
    private volatile RejectedExecutionHandler handler; // 当线程池达到最大容量时的处理方法 默认AbortPolicy
    private volatile long keepAliveTime; //超出corePoolSize的线程会在空闲keepAliveTime后被销毁
    private volatile boolean allowCoreThreadTimeOut; //当为true时,核心线程也会被销毁
    private volatile int corePoolSize;// 核心线程池大小
    private volatile int maximumPoolSize;// 线程池最大容量
    private static final RejectedExecutionHandler defaultHandler =
        new AbortPolicy();  

核心方法分析:

先从最最最常用的execute(Runnable command)方法讲起,但是在讲之前呢,还要先了解一下addWorker(Runnable firstTask,Boolean core)方法:

代码语言:javascript
复制
private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);// 计算线程池运行状态
            // 如果线程池不为RUNNING状且者线程池为SHUTDOWN状态firstTask为空,或工作队列为空时返回false
            //意思就是,当线程处于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();  // cas失败,如果运行状态改变,则自循环
                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); //Worker构造方法会调用ThreadFactory来构造Thread
            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();
                        //J将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 {
                //如果线程不能启动,则将worker从集合中移除
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }
代码语言:javascript
复制
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))
                //并用构造构造线程池时的RejectedExecutionHandlerL来处理 也就是说当线程状态不为RUNNING时,将会拒绝新的任务
                reject(command);
            else if (workerCountOf(recheck) == 0)
            //如果当前没有工作线程,那么创建一个工作线程来执行工作队列中的人物
                addWorker(null, false);
        }
        else if (!addWorker(command, false))
            reject(command);
    }

调用shutDown()方法,不会立即终结线程池,而是不再接受新的任务,如果任务队列中还有任务,会等待任务执行完毕。

代码语言:javascript
复制
public void shutdown() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            // 检查权限
            checkShutdownAccess();
            // 自旋锁 cas来改变线程的状态为SHUTDOWN
            advanceRunState(SHUTDOWN);
            // 终止空闲的工作线程 
            interruptIdleWorkers();
            onShutdown(); // hook for ScheduledThreadPoolExecutor
        } finally {
            mainLock.unlock();
        }
        // 尝试终结线程池
        tryTerminate();
    }

而shutDownNow()方法则为

代码语言:javascript
复制
public List<Runnable> shutdownNow() {
        List<Runnable> tasks;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            // 检查权限
            checkShutdownAccess();
            // 自旋锁 cas将线程状态置为STOP状态
            advanceRunState(STOP);
            // 每个工作线程都会调用interrupt()方法
            interruptWorkers();
            // 将工作队列中任务全部移除,不再执行
            tasks = drainQueue();
        } finally {
            mainLock.unlock();
        }
        // 尝试终结线程
        tryTerminate();
        // 返回工作队列的List集合
        return tasks;
    }

hutDown()与shutDownNow的最大区别就是shutDown()只会为空闲的worker调用interrupt()方法,而shutDownNow()则会为每一个worker调用interrupt()方法,shutDown()会将工作队列中的现有任务执行完毕,再终结,而shutDownNow()会将工作队列中的任务都移除。

如有错误,欢迎指正

原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。

如有侵权,请联系 cloudcommunity@tencent.com 删除。

原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。

如有侵权,请联系 cloudcommunity@tencent.com 删除。

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档