摘要：實(shí)現(xiàn)初始化線(xiàn)程池?cái)?shù)量策略并執(zhí)行線(xiàn)程。二簡(jiǎn)述下從定義線(xiàn)程池，到其底層執(zhí)行的簡(jiǎn)單流程簡(jiǎn)單定義了如上兩句代碼，自動(dòng)創(chuàng)建個(gè)固定大小的線(xiàn)程的線(xiàn)程池，實(shí)現(xiàn)了接口的對(duì)象以后，自動(dòng)啟動(dòng)線(xiàn)程并執(zhí)行對(duì)象的方法。關(guān)鍵的調(diào)用類(lèi)方法，管理線(xiàn)程池并執(zhí)行任務(wù)。

一.類(lèi)圖說(shuō)明

Executor接口類(lèi)，執(zhí)行Runnable接口execute。

ExecutorService接口類(lèi)繼承Executor接口，包含提交執(zhí)行Runnable和Callable接口submit以及shutdown，invokeAll接口。

ScheduledExecutorService接口類(lèi)繼承ExecutorService接口，主要包含計(jì)劃執(zhí)行接口schedule，scheduleAtFixedRate以及scheduleWithFixedDelay。

虛類(lèi)AbstractExecutorService，實(shí)現(xiàn)ExecutorService接口。實(shí)現(xiàn)ExecutorService相關(guān)接口。

ThreadPoolExecutor類(lèi)，繼承虛類(lèi)AbstractExecutorService。實(shí)現(xiàn)初始化線(xiàn)程池?cái)?shù)量、策略并執(zhí)行線(xiàn)程。

ScheduledThreadPoolExecutor類(lèi)，繼承ThreadPoolExecutor類(lèi)并實(shí)現(xiàn)ScheduledExecutorService接口。實(shí)現(xiàn)計(jì)劃執(zhí)行相關(guān)接口。

執(zhí)行類(lèi)，定義ThreadPoolExecutor和ScheduledThreadPoolExecutor類(lèi)，并使用相關(guān)concurrent類(lèi)方法。

二.簡(jiǎn)述下從定義線(xiàn)程池，到其底層執(zhí)行的簡(jiǎn)單流程

ExecutorService executorService = Executors.newFixedThreadPool(3);
executorService.submit(new NewTask());

簡(jiǎn)單定義了如上兩句代碼，JDK自動(dòng)創(chuàng)建3個(gè)固定大小的線(xiàn)程的線(xiàn)程池，submit實(shí)現(xiàn)了Runnable接口的NewTask對(duì)象以后，JDK自動(dòng)啟動(dòng)線(xiàn)程并執(zhí)行NewTask對(duì)象的run方法。流程是如何的呢？

1.Executors的newFixedThreadPool方法new了一個(gè)ThreadPoolExecutor對(duì)象，且new了一個(gè)LinkedBlockingQueue對(duì)象。

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

2.調(diào)用ThreadPoolExecutor的構(gòu)造函數(shù)，其中ThreadFactory默認(rèn)使用defaultThreadFactory，defaultHandler為AbortPolicy。

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

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }
    

3.初始化以后，調(diào)用AbstractExecutorService類(lèi)的submit方法，執(zhí)行Runnable對(duì)象。

public Future submit(Runnable task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture ftask = newTaskFor(task, null);
        execute(ftask);
        return ftask;
    }
    

4.RunnableAdapter實(shí)現(xiàn)了Callable接口，F(xiàn)utureTask類(lèi)包含callable類(lèi)成員對(duì)象。
FutureTask實(shí)現(xiàn)RunnableFuture接口，RunnableFuture繼承Runnable接口。所以newTaskFor(task, null)方法返回一個(gè)FutureTask對(duì)象。

public class FutureTask implements RunnableFuture

public interface RunnableFuture extends Runnable, Future

RunnableFuture ftask = newTaskFor(task, null);

protected  RunnableFuture newTaskFor(Runnable runnable, T value) {
    return new FutureTask(runnable, value);
}

public FutureTask(Runnable runnable, V result) {
    this.callable = Executors.callable(runnable, result);
    this.state = NEW;       // ensure visibility of callable
}

public static  Callable callable(Runnable task, T result) {
    if (task == null)
        throw new NullPointerException();
    return new RunnableAdapter(task, result);
}

static final class RunnableAdapter implements Callable {
    final Runnable task;
    final T result;
    RunnableAdapter(Runnable task, T result) {
        this.task = task;
        this.result = result;
    }
    public T call() {
        task.run();
        return result;
    }
}

5.關(guān)鍵的調(diào)用ThreadPoolExecutor類(lèi)execute方法，管理線(xiàn)程池并執(zhí)行Runnable任務(wù)。主要邏輯如下：
1）如果線(xiàn)程數(shù)量小于corePoolSize，新來(lái)一個(gè)任務(wù)，創(chuàng)建一個(gè)新線(xiàn)程。
2）如果線(xiàn)程數(shù)量等于corePoolSize，則將任務(wù)緩存到workQueue中。
3）如果線(xiàn)程數(shù)量等于corePoolSize并且workQueue隊(duì)列已滿(mǎn)，則繼續(xù)創(chuàng)建線(xiàn)程直到等于maximumPoolSize。
4）如果線(xiàn)程數(shù)量等于maximumPoolSize，且workQueue隊(duì)列已滿(mǎn)，則根據(jù)defaultHandler策略執(zhí)行相應(yīng)措施。默認(rèn)是AbortPolicy，拋出一個(gè)運(yùn)行時(shí)異常RejectedExecutionException。另外還有3種策略：CallerRunsPolicy->如果線(xiàn)程池沒(méi)有shutdown，則直接調(diào)用Runnable的run方法執(zhí)行。如果線(xiàn)程池shutdown，則直接丟棄；DiscardPolicy->直接丟棄，沒(méi)有任何異常；DiscardOldestPolicy->丟棄最老的任務(wù)，并調(diào)用線(xiàn)程池的execute方法執(zhí)行，如果線(xiàn)程池沒(méi)有shutdown。

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        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);
    }