专栏首页冰枫ThreadPoolExecutor源码分析
原创

ThreadPoolExecutor源码分析

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

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

new Thread(()->{}).start();

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

Executor executor = anExecutor; 
executor.execute(new Runnable(()->{}));
class MyExecutor implements Executor{
    public void execute(Runnable command){
        new Thread(command).start();    
    }
}

2 . ExecutorService

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

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提供了取消任务,以及获取任务的执行状态等方法。例如:

ExecutorService service = anExecutorService;
Future future = service.submit(new Runnable(()->{}));
future.cancel();
future.isDone();

3 . AbstractExecutorService

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

4 . ThreadPoolExecutor

字段分析:

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

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

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

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;
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)方法:

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;
    }
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()方法,不会立即终结线程池,而是不再接受新的任务,如果任务队列中还有任务,会等待任务执行完毕。

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()方法则为

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()会将工作队列中的任务都移除。

如有错误,欢迎指正

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

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

我来说两句

0 条评论
登录 后参与评论

相关文章

  • dubbo源码——服务提供者的服务暴露过程(二)

    ##1将address进行分割,和属性填充例如在一个标签中中配置了多个地址,example:<dubbo:registry

    冰枫
  • JDK1.8 LinkedHashMap的实现原理

    LinkedHashMap,顾名思义连接的HashMap,它继承了HashMap,HashMap为了避免碰撞,因此用拉链法解决冲突,读过HashMap源码的读者...

    冰枫
  • BeanFactory与FactoryBean的区别与联系

    BeanFactory顾名思义:生产和管理bean的工厂。它是IOC容器的规范接口,Spring中有很多个BeanFactory的实现,如DefaultList...

    冰枫
  • 【JMeter-3】JMeter参数化4种实现方式

    什么是参数化?从字面上去理解的话,就是事先准备好数据(广义上来说,可以是具体的数据值,也可以是数据生成规则),而非在脚本中写死,脚本执行时从准备好的数据中取值。

    云深i不知处
  • Thread也会OOM吗?

    由于国内手机厂商的奇奇怪怪的优化,特别是华为,其对于线程的构建有特别严苛的要求,当进程内总线程数量达到一定的量级的情况下就会发生线程OOM问题。

    逮虾户
  • Java并发编程:线程控制

    在上一篇文章中(Java并发编程:线程的基本状态)我们介绍了线程状态的 5 种基本状态以及线程的声明周期。这篇文章将深入讲解Java如何对线程进行状态控制,比如...

    陈树义
  • 多线程编程必备技术—— volatile,synchronized,lock

    volatile: volatile是一个类型修饰符(type specifier)。它是被设计用来修饰被不同线程访问和修改的变量。确保本条指令不会...

    Java深度编程
  • 第37节:多线程安全问题

    创建线程的方法 继承类Thread并重写run(),run()称为线程体;用这种方法定义的类不能再继承其他类。

    达达前端
  • Netty中的线程处理EventLoop

    运行任务处理的在编程上的构造通常称作事件循环,Netty使用EventLoop来描述。一个EventLoop将由一个永远不会变的Thread驱动,它可以被指派给...

    爬蜥
  • windows 常用thread方法

    1.HANDLE CreateThread( _In_opt_LPSECURITY_ATTRIBUTES lpThreadAttributes, _In_S...

    战神伽罗

扫码关注云+社区

领取腾讯云代金券