execute()原理

执行步骤如下:

如果workerCount小于corePoolSize,将任务作为first Task新建线程来执行任务。 将任务添加到队列中,如果添加成功依然需要检查,在进入任务之前,

如果线程池被关闭,那么将任务从队列中移除;

如果当前线程池中没有工作线程,而刚刚在队列中加入了任务,要保证线程池中至少有一个工作线程可以处理任务。

如果不能将任务加入队列中,尝试新加一个线程来执行任务,但是并不定会成功,可能是线程池被shut down或者线程池达到了饱和(maximumPoolSize),如果失败了执行拒绝策略。

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        
        int c = ctl.get();
        // 高3位表示状态,低29位任务数量。
    //工作线程小于核心线程数,创建新的线程。
        if (workerCountOf(c) < corePoolSize) {
        //创建新的worker立即执行command,轮询workQueue处理task。
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        //线程池在运行状态且可以将task插入队列
    //第一次校验线程池在运行状态
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            //第二次校验,防止在第一次校验通过后线程池关闭。如果线程池关闭,在队列中删除task并拒绝task
            if (! isRunning(recheck) && remove(command))
                reject(command);
                //如果线程数=0(线程都死掉了,比如:corePoolSize=0),新建线程且未指定firstTask,仅仅去轮训workQueue
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        //线程队列已满,尝试创建新线程执行task,创建失败后拒绝task,创建失败原因:1.线程池关闭;2.线程数已经达到maxPoolSize
        else if (!addWorker(command, false))
            reject(command);
    }
复制代码

addWorker方法

addWorker方法主要有两部分

一:判断是否可以创建worker。根据自旋、CAS、ctl等,判断继续创建还是返回false,自旋周期一般很短。

二:同步创建workder,启动线程。 addWorker会根据当前线程池的工作状态和给定的界限限制(corePoolSize 和maximumPoolSize)判断是否可以添加新的工作线程。 其中每一个Worker对象都是一个AQS队列。


    private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
         //外层循环判断线程池的状态
         for (;;) {
            //在进行获取一次上下文操作机制
            int c = ctl.get();
            int rs = runStateOf(c);
             //线程池状态
            // Check if queue empty only if necessary.
          // 线程池状态:RUNNING = -1、SHUTDOWN = 0、 STOP = 1、TIDYING = 2、TERMINATED = 3 
        //线程池至少是shutdown状态
        if (rs >= SHUTDOWN &&
          // 除了线程池正在关闭(shutdown),
          // 队列里还有未处理的task的情况,其他都不能添加
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;
            //内层循环判断是否到达容量上限,worker+1
            for (;;) {
                int wc = workerCountOf(c);//worker数量
                //worker大于Integer最大上限
                //或到达边界上限
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                //CAS worker+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 
         // CAS失败workerCount被其他线程改变,
         // 重新尝试内层循环CAS对workerCount+1

            }
        }
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            final ReentrantLock mainLock = this.mainLock;
            w = new Worker(firstTask); 
            //1.state置为-1,Worker继承了AbstractQueuedSynchronizer.
            //2.设置firstTask属性.
            //3.Worker实现了Runable接口,将this作为入参创建线程.
            final Thread t = w.thread;
            if (t != null) {
          //addWorker需要加锁
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    int c = ctl.get();
                    int rs = runStateOf(c);
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        workers.add(w);//workers是HashSet<Worker>
              //设置最大线程池大小
                        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);
        }
        return workerStarted;
    }
复制代码

Worker

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

复制代码

Worker方法实现了Runnable,它将构造的Thread赋值给thread。第一步中的setState(-1),即将AQS中的同步状态设置为-1,线程池使用AQS中的同步状态来判断该工作线程是否可以被中断。-1:初始化值,此时工作线程还没有启动,也没有中断的必要;0:表示接受中断,此时工作线程为空闲状态;1:表示此时工作线程正在执行任务。它实现了非重入互斥锁,非重入是为了避免线程池的一些控制方法获得重入锁。注意Worker实现锁的目的与传统锁的意义不太一样。其主要是为了控制线程是否可interrupt,以及其他的监控,如线程是否active(正在执行任务)。

private final class Worker extends AbstractQueuedSynchronizer implements Runnable
{
    private static final long serialVersionUID = 6138294804551838833L;

    //每个worker有自己的内部线程,ThreadFactory创建失败时是null
    //封装任务线程机制
    final Thread thread;
    //初始化任务,可能是null
    Runnable firstTask;
    //每个worker的完成任务数
    volatile long completedTasks;

    Worker(Runnable firstTask) {
        // inhibit interrupts until runWorker //状态置为-1,如果中断线程需要CAS将state 从0-
        >1,以此来保证能只中断从workerQueue getTask的线程
        setState(-1); // 禁止线程在启动前被打断
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }

    //核心方法
    public void run() {
    //首先执行w.unlock,就是把state置为0,对该线程的中断就可以进行了
        runWorker(this);
    }

    // state = 0 代表未锁;state = 1 代表已锁

    protected boolean isHeldExclusively() {
        return getState() != 0;
    }

    // 在setCorePoolSize/shutdown等方法中断worker线程时需要调用该方法,
    // 确保中断的是从workerQueue getTask的线程
    protected boolean tryAcquire(int unused) {
        if (compareAndSetState(0, 1)) {
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }

    protected boolean tryRelease(int unused) {
        setExclusiveOwnerThread(null);
        setState(0);
        return true;
    }

    public void lock()        { acquire(1); }
    public boolean tryLock()  { return tryAcquire(1); }
    public void unlock()      { release(1); }
    //调用tryRelease修改state=0,LockSupport.unpark(thread) 下一个等待锁的线程
    public boolean isLocked() { return isHeldExclusively(); }
    // interrupt已启动线程
    void interruptIfStarted() {
        Thread t;
        // 初始化是 state = -1,不会被interrupt
        if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
            try {
                t.interrupt();
            } catch (SecurityException ignore) {
            }
        }
    }
}
复制代码

Worker的run()方法实际调用的是runWorker(this),启动工作线程(注意这里不是直接启动的任务),在工作线程中执行任务。工作线程在这循环,反复的从队列中获取任务并执行它们。

在运行所有任务之前,需要获取锁,来保证当任务在运行的时候不会被中断,除非线程池正在停止(Stop) 总结如下: 1.Worker类主要负责运行线程状态的控制。 2.Worker继承了AQS实现了简单的获取锁和释放所的操作。来避免中断等待执行任务的线 程时,中断正在运行中的线程(线程刚启动,还没开始执行任务)。 3.自己实现不可重入锁,是为了避免在实现线程池控状态控制的方法,例如:setCorePoolSize的时候中断正在开始运行的线程。setCorePoolSize可能会调用interruptIdleWorkers(),该方法中会调用worker的tryLock()方法中断线程,自己实现锁可以确保工作线程启动之前不会被中断

runWorker的主要任务就是一直loop循环处理任务,没有任务就去getTask(), 代码如下:

final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // 允许被 interrupt
        boolean completedAbruptly = true;
        try {
            //循环获取任务直至 task = null (线程池关闭、超时等)
            // 注意这里的getTask()方法,我们配置的阻塞队列会在这里起作用
            while (task != null || (task = getTask()) != null) {
                w.lock();  // 执行任务前上锁               
                // 如果线程池停止(处于STOP状态或者TIDYING、TERMINATED状态时),设置当前线程处于中断状态如果不是(线程就处于RUNNING或者SHUTDOWN状态),确保线程不中断,重新检查当前线程池的状态是否大于等于STOP状态
                if ((runStateAtLeast(ctl.get(), STOP) ||
                     (Thread.interrupted() &&
                      runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                    wt.interrupt();
                try {
                    //在线程运行前调用
                    beforeExecute(wt, task); 
                    Throwable thrown = null;
                    try {
                        //这里执行run方法
                        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 {
            processWorkerExit(w, completedAbruptly); // 线程退出工作
        }
    }
复制代码

getTask()

从队列中获取Task,大概分为以下几个步骤。

判断线程池以及队列的状态,如果线程池状态在STOP以上,此时线程池不处理队列中的任务;或者线程池处于SHUTDOWN但是队列为空(SHUTDOWN不再接受新的任务),workerCount减1,返回null,注意此时只是将变量减1,其实工作线程并没有终止真正的终止在 processWorkerExit(w, completedAbruptly);中。 如果通过了状态检查,判断是否要进行线程回收,如果需要workerCount数量减1,成功后返回null。 根据timed(timed表示需要进行超时闲置线程回收),选择是限时等待还是阻塞的方式从队列中获取任务。

方法依据配置的workQueue来工作,其阻塞原理与超时原理基于阻塞队列实现,不再详述。

private Runnable getTask() {
    boolean timedOut = false; // Did the last poll() time out?
	//循环
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        //线程线程池状态和队列是否为空
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }
		//线程数量
        int wc = workerCountOf(c);

        
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

        //(当前线程数是否大于最大线程数或者)
        //且(线程数大于1或者任务队列为空)
        //这里有个问题(timed && timedOut)timedOut = false,好像(timed && timedOut)一直都是false吧
        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;
        }
    }
}