最近工作中使用了Java線程池，然后想弄清楚新提交的任務(wù)是如何被分配給線程池中的空余線程的唧取，于是就看了下ThreadPoolExecutor的源碼铅鲤。以下面代碼為例，進(jìn)行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);
    }
}

首先當(dāng)我們提交任務(wù)時邢享，調(diào)用的是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;
    }

然后會執(zhí)行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);
    }

接著執(zhí)行addWorker方法，addWorker的大概邏輯就是淡诗，如果線程池中的線程數(shù)小于corePoolSize骇塘，那么就新建線程執(zhí)行任務(wù)，例子中當(dāng)a,b任務(wù)提交后韩容，我們的線程池中就保持了兩個可用的線程款违，當(dāng)c任務(wù)提交后，是如何處理的呢群凶？

在ThreadPoolExecutor里有runWorker方法插爹，在while循環(huán)里如果getTask()能取到任務(wù)，那么就讓W(xué)orker w不停的執(zhí)行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) {
                   ......
                   此處代碼部分省略
                    //任務(wù)執(zhí)行
                    task.run();
                   此處代碼部分省略
                    ......
                   }
        } 
    }

所以問題的關(guān)鍵就在于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是線程池用來存放任務(wù)的阻塞隊(duì)列毅弧，workQueue.take()在隊(duì)列中沒有任務(wù)的時候會等待气嫁，一旦有新的任務(wù)被提交，r就會獲取到這個最新的任務(wù)够坐，并且被返回給runWorker執(zhí)行任務(wù)寸宵。