圖靈學院Java架構師-VIP-並發編程（線程池）

1、Executor

public interface Executor {

    // 執行Runnable任務
    void execute(Runnable command);
}

ExecutorService：

1. 提供了多種提交任務的方式，支持Callable
2. 定義了線程池相關操作的接口

public interface ExecutorService extends Executor {
  
    // 啟動有序關閉，其中先前提交的任務將被執行，但不會接受任何新任務。
    void shutdown();

    /**
     * 立即關閉執行器, 主要有以下特點:
     * 1. 嘗試停止所有正在執行的任務, 無法保證能夠停止成功, 但會盡力嘗試(例如, 通過 Thread.interrupt中斷任務, 但是不響應中斷的任務可能無法終止);
     * 2. 暫停處理已經提交但未執行的任務;
     *
     * @return 返回已經提交但未執行的任務列表
     */
    List<Runnable> shutdownNow();

    // 如果此執行程序已關閉，則返回true。
    boolean isShutdown();

    // 僅當執行器已關閉且所有任務都已經執行完成, 才返回true.
    boolean isTerminated();

    // 阻止所有任務在關閉請求之后完成執行，或者發生超時，或者當前線程被中斷，以先發生者為准
    boolean awaitTermination(long timeout, TimeUnit unit)
        throws InterruptedException;

    // 接收Callable實現，返回Future對象
    <T> Future<T> submit(Callable<T> task);

    // 提交Runnable任務以執行並返回表示該任務的Future。
    <T> Future<T> submit(Runnable task, T result);

    // 提交Runnable任務以執行並返回表示該任務的Future
    // 注意: Future的get方法在成功完成時將會返回null.
    Future<?> submit(Runnable task);

    // 執行給定集合中的所有任務, 當所有任務都執行完成后, 返回保持任務狀態和結果的 Future 列表.
    // 注意: 該方法為同步方法. 返回列表中的所有元素的Future.isDone() 為 true.
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException;

	// 執行給定集合中的所有任務, 當所有任務都執行完成后或超時期滿時（無論哪個首先發生）
    // 返回保持任務狀態和結果的 Future 列表.    
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                  long timeout, TimeUnit unit)
        throws InterruptedException;

    // 執行給定集合中的任務, 只有其中某個任務率先成功完成（未拋出異常）, 則返回其結果
    // 一旦正常或異常返回后, 則取消尚未完成的任務.
    <T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException;

    // 執行給定集合中的任務, 如果在給定的超時期滿前, 某個任務已成功完成（未拋出異常）, 則返回其結果.
    // 一旦正常或異常返回后, 則取消尚未完成的任務.
    <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                    long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

AbstractExecutorService：

1. 聚合各種提交方式的入口，將各種任務包裝為RunnableFuture，統一調用 execute(RunnableFuture ftask);
2. 實現各種invoke*的調用邏輯，聚合為doInvokeAny

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

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

public <T> Future<T> submit(Callable<T> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task);
    execute(ftask);
    return ftask;
}

ScheduledExecutorService：

交給執行器的某些任務能夠定時執行或周期性地執行

public interface ScheduledExecutorService extends ExecutorService {

    
    public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit);

    
    public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit);

    
    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit);

   
    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                     long initialDelay,
                                                     long delay,
                                                     TimeUnit unit);

}

Executors：

1. 提供線程池的工廠函數
2. 操作線程池創建

核心創建方式：

• newCachedThreadPool創建一個可緩存線程池，如果線程池長度超過處理需要，可靈活回收空閑線程，若無可回收，則新建線程。
• newFixedThreadPool 創建一個定長線程池，可控制線程最大並發數，超出的線程會在隊列中等待。
• newScheduledThreadPool 創建一個定長線程池，支持定時及周期性任務執行
• newSingleThreadExecutor 創建一個單線程化的線程池，它只會用唯一的工作線程來執行任務，保證所有任務按照指定順序(FIFO, LIFO, 優先級)執行。

核心API：

1、execute(Runnable command):履行Ruannable類型的任務,
2、submit(task):可用來提交Callable或Runnable任務，並返回代表此任務的Future 對象 
3、shutdown():在完成已提交的任務后封閉辦事，不再接管新任務,
4、shutdownNow():停止所有正在履行的任務並封閉辦事。
5、isTerminated():測試是否所有任務都履行完畢了。
6、isShutdown():測試是否該ExecutorService已被關閉。

核心參數：

• corePoolSize :池中所保存的線程數，包括空閑線程
• maximumPoolSize:池中允許的最大線程數
• keepAliveTime: 當線程數大於核心時，此為終止前多余的空閑線程等待新任務的最長時間
• unit:keepAliveTime 參數的時間單位
• workQueue :執行前用於保持任務的隊列。此隊列僅保持由 execute 方法提交的 Runnable 任務
• threadFactory:執行程序創建新線程時使用的工廠
• handler :由於超出線程范圍和隊列容量而使執行

運行邏輯：

• 如果當前池大小，poolSize 小於 corePoolSize ，則創建新線程執行任務
• 如果當前池大小，poolSize 大於 corePoolSize ，且等待隊列未滿，則進入等待隊列
• 如果當前池大小，poolSize 大於 corePoolSize 且小於 maximumPoolSize ，且等待隊列已滿，則創建新線程，執行任務
• 如果當前池大小，poolSize 大於 corePoolSize 且大於 maximumPoolSize ，且等待隊列已滿，則調用拒絕策
• 線程池里的每個線程執行完任務后不會立刻退出，而是會去檢查下等待隊列里是否還有線程任務需要執行， 如果在keepAliveTime 里等不到新的任務了，那么線程就會退出

2、執行原理

構造參數：

/**
 * 使用給定的參數創建ThreadPoolExecutor.
 *
 * @param corePoolSize    核心線程池中的最大線程數
 * @param maximumPoolSize 總線程池中的最大線程數
 * @param keepAliveTime   空閑線程的存活時間
 * @param unit            keepAliveTime的單位
 * @param workQueue       任務隊列, 保存已經提交但尚未被執行的線程
 * @param threadFactory   線程工廠(用於指定如何創建一個線程)
 * @param handler         拒絕策略 (當任務太多導致工作隊列滿時的處理策略)
 */

ThreadPoolExecutor在邏輯上將自身管理的線程池划分為兩部分：

• 核心線程池（大小對應為corePoolSize）、
• 非核心線程池（大小對應為maximumPoolSize-corePoolSize）。

向線程池提交一個任務時，將創建一個工作線程——稱之為Worker；Worker在邏輯上從屬於下圖中的【核心線程池】或【非核心線程池】，具體屬於哪一種，要根據corePoolSize、maximumPoolSize、Worker總數進行判斷

【核心線程池】【非核心線程池】都是邏輯上的概念，ThreadPoolExecutor在任務調度過程中會根據corePoolSize和maximumPoolSize的大小，判斷應該如何調度任務

工作原理：

ThreadPoolExecutor內部定義了一個AtomicInteger變量——ctl，通過按位划分的方式，在一個變量中記錄線程池狀態和工作線程數——低29位保存線程數，高3位保存線程池狀態：

/**
 * 保存線程池狀態和工作線程數:
 * 低29位: 工作線程數
 * 高3位 : 線程池狀態
 */
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
 
private static final int COUNT_BITS = Integer.SIZE - 3;
 
// 最大線程數: 2^29-1
private static final int CAPACITY = (1 << COUNT_BITS) - 1;  // 00011111 11111111 11111111 11111111
 
// 線程池狀態
// 11100000 00000000 00000000 00000000
private static final int RUNNING = -1 << COUNT_BITS;   
// 00000000 00000000 00000000 00000000
private static final int SHUTDOWN = 0 << COUNT_BITS;   
// 00100000 00000000 00000000 00000000
private static final int STOP = 1 << COUNT_BITS;     
// 01000000 00000000 00000000 00000000
private static final int TIDYING = 2 << COUNT_BITS;    
// 01100000 00000000 00000000 00000000
private static final int TERMINATED = 3 << COUNT_BITS;      

可以看到，ThreadPoolExecutor一共定義了5種線程池狀態：

• RUNNING : 接受新任務, 且處理已經進入阻塞隊列的任務
• SHUTDOWN : 不接受新任務, 但處理已經進入阻塞隊列的任務
• STOP : 不接受新任務, 且不處理已經進入阻塞隊列的任務, 同時中斷正在運行的任務
• TIDYING : 所有任務都已終止, 工作線程數為0, 線程轉化為TIDYING狀態並准備調用terminated方法
• TERMINATED : terminated方法已經執行完成

內部通過HashSet維護Worker

/**
 * 工作線程集合.
 */
private final HashSet<Worker> workers = new HashSet<Worker>();

Worker：

/**
 * Worker表示線程池中的一個工作線程, 可以與任務相關聯.
 * 由於實現了AQS框架, 其同步狀態值的定義如下:
 * -1: 初始狀態
 * 0:  無鎖狀態
 * 1:  加鎖狀態
 */
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
 
    /**
     * 與該Worker關聯的線程.
     */
    final Thread thread;
    /**
     * Initial task to run.  Possibly null.
     */
    Runnable firstTask;
    /**
     * Per-thread task counter
     */
    volatile long completedTasks;
 
 
    Worker(Runnable firstTask) {
        setState(-1); // 初始的同步狀態值
        this.firstTask = firstTask;
      	// 每個worker對象會調用線程工廠，
        this.thread = getThreadFactory().newThread(this);
    }
 
    /**
     * 執行任務
     */
    public void run() {
        runWorker(this);
    }
 
    /**
     * 是否加鎖
     */
    protected boolean isHeldExclusively() {
        return getState() != 0;
    }
 
    /**
     * 嘗試獲取鎖
     */
    protected boolean tryAcquire(int unused) {
        if (compareAndSetState(0, 1)) {
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }
 
    /**
     * 嘗試釋放鎖
     */
    protected boolean tryRelease(int unused) {
        setExclusiveOwnerThread(null);
        setState(0);
        return true;
    }
 
    public void lock() {
        acquire(1);
    }
 
    public boolean tryLock() {
        return tryAcquire(1);
    }
 
    public void unlock() {
        release(1);
    }
 
    public boolean isLocked() {
        return isHeldExclusively();
    }
 
    /**
     * 中斷線程(僅任務非初始狀態)
     */
    void interruptIfStarted() {
        Thread t;
        if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
            try {
                t.interrupt();
            } catch (SecurityException ignore) {
            }
        }
    }
}

線程工廠：

以下為默認的線程工廠

public static ThreadFactory defaultThreadFactory() {
    return new DefaultThreadFactory();
}

/**
 * 默認的線程工廠.
 */
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;
    }
}

execute的整個執行流程關鍵是下面兩點：

1. 如果工作線程數小於核心線程池上限（CorePoolSize），則直接新建一個工作線程並執行任務；
2. 如果工作線程數大於等於CorePoolSize，則嘗試將任務加入到隊列等待以后執行。如果加入隊列失敗了（比如隊列已滿的情況），則在總線程池未滿的情況下（CorePoolSize ≤ 工作線程數 ＜ maximumPoolSize）新建一個工作線程立即執行任務，否則執行拒絕策略。

Worker線程的生命周期

拒絕策略：

ThreadPoolExecutor在以下兩種情況下會執行拒絕策略：

1. 當核心線程池滿了以后，如果任務隊列也滿了，首先判斷非核心線程池有沒滿，沒有滿就創建一個工作線程（歸屬非核心線程池）， 否則就會執行拒絕策略；
2. 提交任務時，ThreadPoolExecutor已經關閉了。

四種拒絕策略：

• AbortPolicy（默認）：AbortPolicy策略其實就是拋出一個RejectedExecutionException異常
• DiscardPolicy：DiscardPolicy策略其實就是無為而治，什么都不做，等任務自己被回收
• DiscardOldestPolicy：DiscardOldestPolicy策略是丟棄任務隊列中的最近一個任務，並執行當前任務
• CallerRunsPolicy：CallerRunsPolicy策略相當於以自身線程來執行任務，這樣可以減緩新任務提交的速度

3、源碼解析：

構造：

ExecutorService executorService = Executors.newFixedThreadPool(100);

創建ThreadPoolExecutor實例：

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

// 指定默認的線程創建工廠和拒絕策略
public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), defaultHandler);
}

/**
 * 使用給定的參數創建ThreadPoolExecutor.
 *
 * @param corePoolSize    核心線程池中的最大線程數
 * @param maximumPoolSize 總線程池中的最大線程數
 * @param keepAliveTime   空閑線程的存活時間
 * @param unit            keepAliveTime的單位
 * @param workQueue       任務隊列, 保存已經提交但尚未被執行的線程
 * @param threadFactory   線程工廠(用於指定如何創建一個線程)
 * @param handler         拒絕策略 (當任務太多導致工作隊列滿時的處理策略)
 */
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> 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.acc = System.getSecurityManager() == null ? 
      	null : AccessController.getContext();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

執行任務：

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
  
    int c = ctl.get();
  	// CASE1: 工作線程數 < 核心線程池上限
    if (workerCountOf(c) < corePoolSize) {
      	// 添加工作線程並執行
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    // 如果工作隊列未滿，再次檢查運行狀態，並插入到任務
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        // 再次check判斷運行狀態如果是非運行狀態就移除出去&reject掉
        if (!isRunning(recheck) && remove(command))
            reject(command);
        // 發現可能運行線程數是0那么增加一個null的worker
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 直接增加worker如果不成功直接reject
    else if (!addWorker(command, false))
        reject(command);
}

addWorker()

/**
 * 添加工作線程並執行任務
 *
 * @param firstTask 如果指定了該參數, 表示將立即創建一個新工作線程執行該firstTask任務; 
 										否則復用已有的工作線程，從工作隊列中獲取任務並執行
 * @param core      執行任務的工作線程歸屬於哪個線程池:  true-核心線程池  false-非核心線程池
 */
private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        int c = ctl.get();
      	// 獲取線程池狀態
        int rs = runStateOf(c);

        /**
         * 這個if主要是判斷哪些情況下, 線程池不再接受新任務執行
         * 1. 線程池狀態為 STOP 或 TIDYING 或 TERMINATED: 線程池狀態為上述任一一種
         * 2. 線程池狀態 ≥ SHUTDOWN 且 firstTask != null: 
         			因為當線程池狀態 ≥ SHUTDOWN時, 不再接受新任務的提交，所以直接返回
         * 3. 線程池狀態 ≥ SHUTDOWN 且 隊列為空
         			隊列中已經沒有任務了, 所以也就不需要執行任何任務了，可以直接返回
         */
        if (rs >= SHUTDOWN &&
            !(rs == SHUTDOWN && firstTask == null && !workQueue.isEmpty()))
            return false;

        for (;;) {
            int wc = workerCountOf(c);
            // 檢查容量是否超出
          	// 1. 超出最大容量；2、core為true表示核心線程數量，為false表示非核心線程數量
            if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            // 增加worker數量
            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 {
        // 創建worker(AQS)
        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();
                    // 添加到worker隊列中
                    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)
            // 添加worker失敗處理
            addWorkerFailed(w);
    }
    return workerStarted;
}

啟動Worker后，會調用run()

public void run() {
    runWorker(this);
}

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
      	// 當task==null時會通過getTask從隊列取任務
        while (task != null || (task = getTask()) != null) {
            w.lock();
            /**
             * 下面這個if判斷的作用如下:
             * 1.保證當線程池狀態為STOP/TIDYING/TERMINATED時，
             		當前執行任務的線程wt是中斷狀態(因為線程池處於上述任一狀態時，均不能再執行新任務)
             * 2.保證當線程池狀態為RUNNING/SHUTDOWN時，當前執行任務的線程wt不是中斷狀態
             */
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();
            try {
                // 執行前置處理, 1.8版本為空實現
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    // 執行任務
                    task.run();
                } catch (RuntimeException x) {
                    thrown = x; throw x;
                } catch (Error x) {
                    thrown = x; throw x;
                } catch (Throwable x) {
                    thrown = x; throw new Error(x);
                } finally {
                    // 執行后置處理, 1.8版本為空實現
                    afterExecute(task, thrown);
                }
            } finally {
                task = null;
                // 更新該worker完成狀態
                w.completedTasks++;
                w.unlock();
            }
        }
      	// 執行到此處, 說明該工作線程自身既沒有攜帶任務, 也沒從任務隊列中獲取到任務
      	// 正常退出時，為false, 如果拋出異常退出，則為true
        completedAbruptly = false;
    } finally {
	      // 處理工作線程的退出工作
        processWorkerExit(w, completedAbruptly);
    }
}  

processWorkerExit(Worker w, boolean completedAbruptly)

private void processWorkerExit(Worker w, boolean completedAbruptly) {
	  // 工作線程因異常情況而退出
    if (completedAbruptly) 
      	// 工作線程數減1(如果工作線程執行時沒有出現異常, 在getTask()方法中已經對線程數減1了)
        decrementWorkerCount();

    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
      	// completedTaskCount記錄線程池完成的總任務數
        completedTaskCount += w.completedTasks;
      	// 從工作線程集合中移除(該工作線程會自動被GC回收)
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }
		// 根據線程池狀態, 判斷是否需要終止線程池
    tryTerminate();

    int c = ctl.get();
    if (runStateLessThan(c, STOP)) {
      	// 工作線程為正常退出
        if (!completedAbruptly) {
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
      	// 新建一個工作線程
        addWorker(null, false);
    }
}

4、ScheduledThreadPoolExecutor

要滿足任務的延遲/周期調度功能，它會對所有的Runnable任務都進行包裝，包裝成一個RunnableScheduledFuture任務。

• ScheduledThreadPoolExecutor中的任務隊列是一種特殊的延時隊列
• DelayedWorkQueue：它和DelayQueue類似，只不過要求所有入隊元素必須實現RunnableScheduledFuture接口。

執行原理

構造：

ScheduledThreadPoolExecutor的構造器，內部其實都是調用了父類ThreadPoolExecutor的構造器，這里最需要注意的就是任務隊列的選擇——DelayedWorkQueue

調度：

核心調度方法是schedule、scheduleAtFixedRate、scheduleWithFixedDelay

1、將任務包裝為ScheduledFutureTask

2、判斷狀態，將任務添加到DelayedWorkQueue

1. 首先，任務被提交到線程池后，會判斷線程池的狀態，如果不是RUNNING狀態會執行拒絕策略。
2. 然后，將任務添加到阻塞隊列中。（注意，由於DelayedWorkQueue是無界隊列，所以一定會add成功）
3. 然后，會創建一個工作線程，加入到核心線程池或者非核心線程池

如果核心線程池未滿，則新建的工作線程會被放到核心線程池中。如果核心線程池已經滿了，ScheduledThreadPoolExecutor不會像ThreadPoolExecutor那樣再去創建歸屬於非核心線程池的工作線程，而是直接返回。也就是說，在ScheduledThreadPoolExecutor中，一旦核心線程池滿了，就不會再去創建工作線程。

生產實踐

如何合理的設置線程池大小？

分析任務的特性：

1. 任務的性質：CPU密集型任務、IO密集型任務、混合型任務。
2. 任務的優先級：高、中、低。
3. 任務的執行時間：長、中、短。
4. 任務的依賴性：是否依賴其他系統資源，如數據庫連接等。

• CPU密集型任務應配置盡可能小的線程，如配置CPU個數+1的線程數
• IO密集型任務應配置盡可能多的線程，因為IO操作不占用CPU，不要讓CPU閑下來，應加大線程數量，如配置兩倍CPU個數+1
• 而對於混合型的任務，如果可以拆分，拆分成IO密集型和CPU密集型分別處理，前提是兩者運行的時間是差不多的，如果處理時間相差很大，則沒必要拆分
• 若任務對其他系統資源有依賴，如某個任務依賴數據庫的連接返回的結果，這時候等待的時間越長，則CPU空閑的時間越長，那么線程數量應設置得越大，才能更好的利用CPU。

合理設置線程池大小公式：

最佳線程數目 = （（線程等待時間+線程CPU時間）/ 線程CPU時間 ）* CPU數目

轉化為：

最佳線程數目 = （線程等待時間與線程CPU時間之比 + 1）* CPU數目

結論：線程等待時間所占比例越高，需要越多線程。線程CPU時間所占比例越高，需要越少線程

並發情況下線程池配置：

1. 並發高，執行時間短的任務配置盡可能少的線程：CPU核數 + 1
2. 並發高、業務執行時間長的任務，對於系統的壓力很大，應盡可能通過架構的優化，而不是線程池的配置解決。例如轉換為異步削峰解藕
3. 並發不高，業務時間長：
   1. 如果任務長時間消耗在IO操作上，應加大線程池數量，不讓CPU閑下來，盡量執行更多的任務
   2. 如果任務長時間消耗在計算上，應減少CPU線程的切換，設置和CPU核數一致的線程數量。

《Java並發編程實戰》

Nthreads = NCPU * UCPU * (1 + W/C)

其中：

NCPU是處理器的核的數目，可以通過Runtime.getRuntime().availableProcessors()得到

UCPU是期望的CPU利用率（該值應該介於0和1之間）

W/C是等待時間與計算時間的比率

CPU密集型，設置核心線程永久存在：因為執行任務優先判斷核心線程數量，不足則創建，避免資源浪費

CPU密集型任務，隊列選擇阻塞隊列，高並發場景大量CAS資源占用大量CPU資源，影響性能。