最近工作中使用了Java线程池，然后想弄清楚新提交的任务是如何被分配给线程池中的空余线程的，于是就看了下ThreadPoolExecutor的源码。以下面代码为例，进行debug。

public class ThreadPoolDemo {


    public static void main(String args[]){
        ExecutorService pool = Executors.newFixedThreadPool(2);
        Printer a = new Printer("I am A");
        Printer b = new Printer("I am B");
        pool.submit(a);
        pool.submit(b);
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        Printer c = new Printer("I am C");
        pool.submit(c);

        pool.shutdown();
    }
}

class Printer implements Runnable {

    private String msg;

    public Printer(String msg){
        this.msg = msg;
    }

    public void run()  {

        System.out.println(msg);
    }
}

首先当我们提交任务时，调用的是AbstractExecutorService的submit方法

 /**
     * @throws RejectedExecutionException {@inheritDoc}
     * @throws NullPointerException       {@inheritDoc}
     */
    public Future<?> submit(Runnable task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<Void> ftask = newTaskFor(task, null);
        execute(ftask);
        return ftask;
    }

然后会执行ThreadPoolExecutor的execute方法

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        /*
         * Proceed in 3 steps:
         *
         * 1. If fewer than corePoolSize threads are running, try to
         * start a new thread with the given command as its first
         * task.  The call to addWorker atomically checks runState and
         * workerCount, and so prevents false alarms that would add
         * threads when it shouldn't, by returning false.
         *
         * 2. If a task can be successfully queued, then we still need
         * to double-check whether we should have added a thread
         * (because existing ones died since last checking) or that
         * the pool shut down since entry into this method. So we
         * recheck state and if necessary roll back the enqueuing if
         * stopped, or start a new thread if there are none.
         *
         * 3. If we cannot queue task, then we try to add a new
         * thread.  If it fails, we know we are shut down or saturated
         * and so reject the task.
         */
        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))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        else if (!addWorker(command, false))
            reject(command);
    }

接着执行addWorker方法，addWorker的大概逻辑就是，如果线程池中的线程数小于corePoolSize，那么就新建线程执行任务，例子中当a,b任务提交后，我们的线程池中就保持了两个可用的线程，当c任务提交后，是如何处理的呢？

在ThreadPoolExecutor里有runWorker方法，在while循环里如果getTask()能取到任务，那么就让Worker w不停的执行task。

    final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
            while (task != null || (task = getTask()) != null) {
                   ......
                   此处代码部分省略
                    //任务执行
                    task.run();
                   此处代码部分省略
                    ......
                   }
        } 
    }

所以问题的关键就在于getTask()方法。

private Runnable getTask() {
        boolean timedOut = false; // Did the last poll() time out?

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

            // Check if queue empty only if necessary.
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }

            int wc = workerCountOf(c);

            // Are workers subject to culling?
            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;
            }
        }
    }

其中下面这行代码揭示了最终问题的答案。

Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();

workQueue是线程池用来存放任务的阻塞队列，workQueue.take()在队列中没有任务的时候会等待，一旦有新的任务被提交，r就会获取到这个最新的任务，并且被返回给runWorker执行任务。