ThreadPoolExecutor execute 方法分析

      分析完AbstractExecutorService异步任务提交之后，一直留着一个问题：就是任务提交之后的最终执行方法execute(Runnable)始终没有细究，只知道它会在将来某个时刻去执行任务，也就是所谓的异步执行。 现在可以揭开异步执行方法executor(Runnable command)的真面目了，回到线程池执行器ThreadPoolExecutor，乍看这个方法，蛮精干的：

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
            if (runState == RUNNING && workQueue.offer(command)) {
                if (runState != RUNNING || poolSize == 0)
                    ensureQueuedTaskHandled(command);
            }
            else if (!addIfUnderMaximumPoolSize(command))
                reject(command); // is shutdown or saturated
        }
}

      作为异步任务执行的核心方法，看上去短短10行，其实蕴含了许多依赖的组件，这个可以参考文章“ThreadPoolExecutor 分析之类基础架构 ”。

      在分析之前，需要再解释几个概念，ThreadPoolExecutor是一个线程池执行器，它里面维护了一个线程池，以及一个等待执行的任务队列。

      所谓线程池，其实就是一个Worker对象的集合：HashSet<Worker> workers = new HashSet<Worker>();至于这个Worker的基本类图如下：      

 

     从类图看(当然最好结合代码)，Worker本身是一个Runnable，它自己维护了执行它的线程对象thead，又维护了一个Runnable对象firstTask（这个对象就是ThreadPoolExecutor线程池的任务对象了），当ThreadPoolExecutor执行一个任务的时候，先获得 (最直接的方法就是new)一个可用的Thread对象，然后再获得(最直接方法就是new)一个Worker对象，并把Thread对象包装进这个Worker对象中，接着让这个thread对象start就开始执行这个Worker对象的run()方法，而run()方法中会去执行Worker.firstTask.run()方法。这就间接的的执行了目标任务，同时通过worker这个包装(或者说代理)之后，可用做很多额外的工作，比如中断自身执行线程，记录在该线程上执行过的任务数量等。分析addIfUnderCorePoolSize(command)方法时还会分析Worker类。 

    线程池数量poolSize指工作线程Worker对象的集合workers的实际大小，通过workers.size()可直接获得。            

      核心线程池数量corePoolSize，可理解为工作线程Worker对象的集合workers的目标大小。如果poolSize > corePoolSize，那么ThreadPoolExecutor就会有机制在适当的时候回收闲置的线程。

 

     最大线程池数量maxPoolSize，就是工作线程Worker对象的集合workers的大小上限。假如说任务队列满了，再来新任务时，若poolSize还没达到maxPoolSize，则继续创建新的线程来执行新任务，若不幸poolSize达到了上限maxPoolSize，那不能再创建新的线程了，只能采取reject策略来拒绝新任务。

    所谓任务队列，就是一个Runnable对象的阻塞队列：BlockingQueue<Runnable> workQueue; 可根据不同需求设置不同的队列类型。

     下面分析execute(command)执行流程：

1.  第四行if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command))

 

          解释：  如果当前线程池中线程数量poolSize >=  核心线程数量corePoolSize 成立，那么逻辑或运算符后面的方法addIfUnderCorePoolSize(command)就忽略不做，而直接进花括号内部；如果poolSize >= corePoolSize不成立，尝试调用addIfUnderCorePoolSize(command)方法，该方法返回true就进花括号，否则整个execute方法就结束。

          理解：这很好理解，如果当前线程数量poolSize>=核心线程数量corePoolSize，那当然无法再把当前任务加入到核心线程池中执行了，于是进花括号选择其他的策略执行；如果poolSize没有达到corePoolSize，那很自然是把当前任务放到核心线程池执行，也就是执行逻辑或运算符后的方法addIfUnderCorePoolSize(command)。“放到核心线程池执行”是什么意思呢？就是new 一个新工作线程放到workers集合中，让这个新线程来执行当前的任务command，而这个新线程可以认为是核心线程池中的其中一个线程。

 addIfUnderCorePoolSize(command)方法做了什么事情？请看代码：

 

private boolean addIfUnderCorePoolSize(Runnable firstTask) {
        Thread t = null;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (poolSize < corePoolSize && runState == RUNNING)
                t = addThread(firstTask);
        } finally {
            mainLock.unlock();
        }
        if (t == null)
            return false;
        t.start();
        return true;
    }

 

      主要就是在poolSize < corePoolSize并且当前状态runState == RUNNING时通过方法addThread(firstTask)返回一个线程t，然后马上t.start()执行任务，返回true。这里有失败的可能，因为虽然在前面execute方法中已经保证了poolSize < corePoolSize进入该方法，但是当时并没有加锁，很有可能到了当前的位置poolSize已经改变了，所以这里必须再次检查并且必须加锁访问。顺便简单看下addThread方法：

 

private Thread addThread(Runnable firstTask) {
        Worker w = new Worker(firstTask);
        Thread t = threadFactory.newThread(w);
        if (t != null) {
            w.thread = t;
            workers.add(w);
            int nt = ++poolSize;
            if (nt > largestPoolSize)
                largestPoolSize = nt;
        }
        return t;
    }

 

       直接new一个工作线程Worker对象，封装目标任务firstTask，用线程工厂创建一个新线程t,然后t又赋值给Worker的thread属性，这下对Worker有了新理解吧？线程t 用来执行Worker对象，而Worker对象的thread属性的值＝t ，他维护了执行它自己的那个线程。t.start()便开始执行worker.run()，而worker.run中会去执行封装目标任务firstTask.run()方法。当然Worker还没有完，后面还会分析到。

       addIfUnderCorePoolSize(command)会返回false吗？如果corePoolSize设置的足够大，基本就不会失败，那execute(command)方法做完这句if语句判断就结束了，压根不需要进入花括号继续。但是不幸的是，addIfUnderCorePoolSize还是经常会失败的。所以接下来还得看看if大括号里面的逻辑。

 

2. 第五行if (runState == RUNNING && workQueue.offer(command))

       解释：runState表示这个TheadPoolExecutor的状态，可以有4个状态

        a)      RUNNING可接收新任务并执行任务队列

        b)      SHUTDOWN不能接收新任务，但可以继续执行任务队列

        c)       STOP不能接收新任务，也不在处理任务队列，并且中断正在执行的任务

        d)      TERMINATED在STOP基础上，所有线程都已终止

    程序若到了这一步，说明当前线程数量poolSize >=核心线程数量corePoolSize，这里先判断是不是这个TheadPoolExecutor还是RUNNING状态，若是则试着加入到任务队列workQueue中，无法加入的唯一可能就是队列已经满了。先说没满的情况，就是任务加入到任务队列成功。按照常理，加入了队列以后，只要保证有工作线程就ok了，工作线程会自动去执行任务队列的。所以判断一下if ( runState != RUNNING || poolSize == 0),在这个if为true时候，去保证一下任务队列有线程会执行，即执行ensureQueuedTaskHandled(command)方法。这里有两种情况，情况一：runState != RUNNING，这种情况在ensureQueuedTaskHandled方法中会把任务丢给reject拒绝策略处理，情况二：poolSize == 0，这种情况是new一个新线程加入到工作线程集合workers中。

 

3. 第九行else if (!addIfUnderMaximumPoolSize(command))

       解释：程序执行到这个分支，说明上面第五行if条件为false，也就是说当前状态runState ！= RUNNING，或者任务队列workQueue已经满了。先看第一个条件下，前面解释过runState，除了RUNNING状态，其他三个状态都不能接收新任务，所以当runState ！= RUNNING时新任务只能根据reject策略拒绝，而这个拒绝的逻辑是在addIfUnderMaximumPoolSize方法中实现的；再看第二个条件下，workQueue已经满，潜在的条件是runState == RUNNING，这种情况怎么处理新任务呢？很简单，若当前线程数量已经poolSize没有达到最大线程数量maxPoolSize，则创建新的线程去执行这个无法加入任务队列的新任务，否则就根据reject策略拒绝，这里的拒绝逻辑就在这个else if条件成立的子句中做的，即第10行reject(command);

 

      到此，任务异步执行的整个过程execute(Runnable command)分析完毕，顺便画个流程图作为本文的结尾，本来应该呼应一下前文，完成对工作线程Worker类的分析，包括选取执行任务队列、及在任务队列为空时将Worker对象回收，但是看来得放到下一篇了。

评论

2 楼 秦时明月黑 2014-06-12  

三年了，距今三年了，三年前已经有人熟读源码，膜拜，邀想2012年刚毕业

1 楼 fengzhonghun102 2014-03-18  

太专业了！ 非常感谢！