@(Executors)[newFixedThreadPool]

[TOC]

java線程池

在面向?qū)ο缶幊讨?，?chuàng)建和銷毀對(duì)象是很費(fèi)時(shí)間的，因?yàn)閯?chuàng)建一個(gè)對(duì)象要獲取內(nèi)存資源或者其它更多資源。在Java中更是如此，虛擬機(jī)將試圖跟蹤每一個(gè)對(duì)象，以便能夠在對(duì)象銷毀后進(jìn)行垃圾回收。所以提高服務(wù)程序效率的一個(gè)手段就是盡可能減少創(chuàng)建和銷毀對(duì)象的次數(shù)，特別是一些很耗資源的對(duì)象創(chuàng)建和銷毀。如何利用已有對(duì)象來服務(wù)就是一個(gè)需要解決的關(guān)鍵問題，其實(shí)這就是一些"池化資源"技術(shù)產(chǎn)生的原因。

newFixedPool作用

創(chuàng)建一個(gè)固定線程數(shù)的線程池，在任何時(shí)候最多只有nThreads個(gè)線程被創(chuàng)建。如果在所有線程都處于活動(dòng)狀態(tài)時(shí)，有其他任務(wù)提交，他們將等待隊(duì)列中直到線程可用。如果任何線程由于執(zhí)行過程中的故障而終止，將會(huì)有一個(gè)新線程將取代這個(gè)線程執(zhí)行后續(xù)任務(wù)。

構(gòu)造方法

newFixedPool擁有兩個(gè)構(gòu)造方法：

參數(shù)為nThreads的構(gòu)造方法:

 public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

構(gòu)造方法中的nThreads代表固定線程池中的線程數(shù)，當(dāng)使用此構(gòu)造方法創(chuàng)建線程池后，就會(huì)創(chuàng)建nThreads個(gè)線程在線程池內(nèi)。
參數(shù)為： nThreads,ThreadFactory的構(gòu)造方法:

  public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>(),
                                      threadFactory);
    }

ThreadPoolExecutor構(gòu)造方法

newFixedThreadPool(int nThreads) 使用的構(gòu)造方法:

    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }

此ThreadPoolExecutor構(gòu)造方法的參數(shù)有5個(gè)：
corePoolSize：線程池中所保存的線程數(shù)，包括空閑線程，newFixedThreadPool中傳入nThreads
maximumPoolSize：池中允許的最大線程數(shù)，newFixedThreadPool中傳入nThreads，使線程池的最大線程數(shù)與線程池中保存的線程數(shù)一致，使保證線程池的線程數(shù)是固定的
TimeUnit：參數(shù)的時(shí)間單位
keepAliveTime：當(dāng)線程數(shù)大于corePoolSize時(shí)，此為終止前多余的空閑線程等待新任務(wù)的最長時(shí)間
LinkedBlockingQueue workQueuqe：當(dāng)任務(wù)被執(zhí)行前，放到此阻塞隊(duì)列中，任務(wù)將被放在阻塞隊(duì)列中直到使用execute方法提交Runnable任務(wù),不同的線程池主要是這個(gè)參數(shù)的不通，比如scheduledThreadPoolExecutor這邊使用的就是DelayedWorkQueue這個(gè)隊(duì)列

**注意：
* The queue used for holding tasks and handing off to worker
* threads. We do not require that workQueue.poll() returning
* null necessarily means that workQueue.isEmpty(), so rely
* solely on isEmpty to see if the queue is empty (which we must
* do for example when deciding whether to transition from
* SHUTDOWN to TIDYING). This accommodates special-purpose
* queues such as DelayQueues for which poll() is allowed to
* return null even if it may later return non-null when delays
* expire.
*/

    private final BlockingQueue<Runnable> workQueue;

newFixedThreadPool(in nThreads,ThreadFactory threadFactory)使用的構(gòu)造方法:

 public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>(),
                                      threadFactory);
    }

    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);   //創(chuàng)建默認(rèn)線程工廠，并設(shè)置駁回的異常處理
    }

創(chuàng)建ThreadPoolExecutor時(shí)傳入Executors.defaultThreadFactory()這個(gè)默認(rèn)線程工廠實(shí)例。
在代碼中調(diào)用ThreadPoolExecutor中的內(nèi)部類DefaultThreadFactory工廠，在包中的注釋是這么寫的：threadFactory the factory to use when the executor creates a new thread 。這是用來使用executor創(chuàng)建新線程的工廠。

static class DefaultThreadFactory implements ThreadFactory {
    private static final AtomicInteger poolNumber = new AtomicInteger(1);
    private final ThreadGroup group;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;

    DefaultThreadFactory() {
        SecurityManager s = System.getSecurityManager();
        group = (s != null) ? s.getThreadGroup() :
                              Thread.currentThread().getThreadGroup();
        namePrefix = "pool-" +
                      poolNumber.getAndIncrement() +
                     "-thread-";
    }

    public Thread newThread(Runnable r) {
        Thread t = new Thread(group, r,
                              namePrefix + threadNumber.getAndIncrement(),
                              0);
        if (t.isDaemon())
            t.setDaemon(false);
        if (t.getPriority() != Thread.NORM_PRIORITY)
            t.setPriority(Thread.NORM_PRIORITY);
        return t;
    }
}

調(diào)用DefaultThreadFactory的構(gòu)造方法初始化線程組和生成線程前綴。在創(chuàng)建完線程工廠后，需要使用線程工廠來創(chuàng)建線程并啟動(dòng)。

使用線程池中的線程執(zhí)行Runnable任務(wù)

當(dāng)之前的初始化（corePoolSize等于maxiumPoolSize等于nThreads，并創(chuàng)建defaultThreadFactory）就可以運(yùn)行定義好的線程了。
執(zhí)行線程的代碼如下：

    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，其方法目的是創(chuàng)建Worker對(duì)象，并將Worker對(duì)象加入到HashSet<Worker> workers中。在加入workers之前，先創(chuàng)建ReentrantLock排它鎖，將worker同步的加入到workers中。當(dāng)worker成功被加入后，啟動(dòng)本次放入set中的線程。

  private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);

            // Check if queue empty only if necessary.
            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;
                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
            }
        }

        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            w = new Worker(firstTask);
            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();
                        workers.add(w);
                        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對(duì)象的構(gòu)造方法：

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

通過線程工廠 來創(chuàng)建線程。

 public Thread newThread(Runnable r) {
        Thread t = new Thread(group, r,
                              namePrefix + threadNumber.getAndIncrement(),
                              0);
        if (t.isDaemon())
            t.setDaemon(false);
        if (t.getPriority() != Thread.NORM_PRIORITY)
            t.setPriority(Thread.NORM_PRIORITY);
        return t;
    }