本文目錄：

1. 概述

我們這里的隊(duì)列都指線程池使用的阻塞隊(duì)列 BlockingQueue 的實(shí)現(xiàn)。

什么是有界隊(duì)列？就是有固定大小的隊(duì)列。比如設(shè)定了固定大小的 LinkedBlockingQueue，又或者大小為 0，只是在生產(chǎn)者和消費(fèi)者中做中轉(zhuǎn)用的 SynchronousQueue。

什么是無界隊(duì)列？指的是沒有設(shè)置固定大小的隊(duì)列。這些隊(duì)列的特點(diǎn)是可以直接入列，直到溢出。當(dāng)然現(xiàn)實(shí)幾乎不會(huì)有到這么大的容量（超過 Integer.MAX_VALUE），所以從使用者的體驗(yàn)上，就相當(dāng)于 “無界”。比如沒有設(shè)定固定大小的 LinkedBlockingQueue。

所以無界隊(duì)列的特點(diǎn)就是可以一直入列，不存在隊(duì)列滿負(fù)荷的現(xiàn)象。

這個(gè)特性，在我們自定義線程池的使用中非常容易出錯(cuò)。而出錯(cuò)的根本原因是對(duì)線程池內(nèi)部原理的不了解。

比如有這么一個(gè)案例，我們使用了無界隊(duì)列創(chuàng)建了這樣一個(gè)線程池：

ExecutorService executor =  new ThreadPoolExecutor(2, 4, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());

配置的參數(shù)如下：

• 核心線程數(shù) 2
• 最大線程數(shù) 4
• 空閑線程?；顣r(shí)間 60s
• 使用無界隊(duì)列 LinkedBlockingQueue

然后對(duì)這個(gè)線程池我們提出一個(gè)問題：使用過程中，是否會(huì)達(dá)到最大線程數(shù) 4？

2. 驗(yàn)證

我們寫了個(gè) Demo 驗(yàn)證一下，設(shè)定有 10 個(gè)任務(wù)，每個(gè)任務(wù)執(zhí)行 10s。

任務(wù)的執(zhí)行代碼如下，用 Thread.sleep 操作模擬執(zhí)行任務(wù)的阻塞耗時(shí)。

/**
 * @author lidiqing
 * @since 17/9/17.
 */
public class BlockRunnable implements Runnable {
    private final String mName;

    public BlockRunnable(String name) {
        mName = name;
    }

    public void run() {
        System.out.println(String.format("[%s] %s 執(zhí)行", Thread.currentThread().getName(), mName));
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

然后在 main 方法中把這 10 個(gè)任務(wù)扔進(jìn)剛剛設(shè)計(jì)好的線程池中：

 public static void main(String[] args) {
        ExecutorService executor =  new ThreadPoolExecutor(2, 4, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
        for (int i = 0; i < 10; i++) {
            executor.execute(new BlockRunnable(String.valueOf(i)));
        }
    }

結(jié)果輸出如下：

[pool-1-thread-2] 1 執(zhí)行
[pool-1-thread-1] 0 執(zhí)行
[pool-1-thread-2] 2 執(zhí)行
[pool-1-thread-1] 3 執(zhí)行
[pool-1-thread-1] 5 執(zhí)行
[pool-1-thread-2] 4 執(zhí)行
[pool-1-thread-2] 7 執(zhí)行
[pool-1-thread-1] 6 執(zhí)行
[pool-1-thread-1] 8 執(zhí)行
[pool-1-thread-2] 9 執(zhí)行

發(fā)現(xiàn)了什么問題？這里最多出現(xiàn)兩個(gè)線程。當(dāng)放開到更多的任務(wù)時(shí)，也依然是這樣。

3. 剖析

我們回到線程池 ThreadPoolExecutor 的 execute 方法來找原因。

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);
    }

上面代碼的核心就是任務(wù)進(jìn)入等待隊(duì)列 workQueue 的時(shí)機(jī)。答案就是，執(zhí)行 execute 方法時(shí)，如果發(fā)現(xiàn)核心線程數(shù)已滿，是會(huì)先執(zhí)行 workQueue.offer(command) 來入列。

也就是 當(dāng)核心線程數(shù)滿了后，任務(wù)優(yōu)先進(jìn)入等待隊(duì)列。如果等待隊(duì)列也滿了后，才會(huì)去創(chuàng)建新的非核心線程 。

所以我們上面設(shè)計(jì)的線程池，使用了無界隊(duì)列，會(huì)直接導(dǎo)致最大線程數(shù)的配置失效。

可以用一張圖來展示整個(gè) execute 階段的過程：

所以上面的線程池，實(shí)際使用的線程數(shù)的最大值始終是 corePoolSize ，即便設(shè)置了 maximumPoolSize 也沒有生效。 要用上 maximumPoolSize ，允許在核心線程滿負(fù)荷下，繼續(xù)創(chuàng)建新線程來工作 ，就需要選用有界任務(wù)隊(duì)列?？梢越o LinkedBlockingQueue 設(shè)置容量，比如 new LinkedBlockingQueue(128) ，也可以換成 SynchronousQueue。

舉個(gè)例子，用來做異步任務(wù)的 AsyncTask 的內(nèi)置并發(fā)執(zhí)行器的線程池設(shè)計(jì)如下：

public abstract class AsyncTask<Params, Progress, Result> {     
    private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
    // We want at least 2 threads and at most 4 threads in the core pool,
    // preferring to have 1 less than the CPU count to avoid saturating
    // the CPU with background work
    private static final int CORE_POOL_SIZE = Math.max(2, Math.min(CPU_COUNT - 1, 4));
    private static final int MAXIMUM_POOL_SIZE = CPU_COUNT * 2 + 1;
    private static final int KEEP_ALIVE_SECONDS = 30;

    private static final ThreadFactory sThreadFactory = new ThreadFactory() {
        private final AtomicInteger mCount = new AtomicInteger(1);

        public Thread newThread(Runnable r) {
            return new Thread(r, "AsyncTask #" + mCount.getAndIncrement());
        }
    };

    private static final BlockingQueue<Runnable> sPoolWorkQueue =
            new LinkedBlockingQueue<Runnable>(128);

    /**
     * An {@link Executor} that can be used to execute tasks in parallel.
     */
    public static final Executor THREAD_POOL_EXECUTOR;

    static {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE_SECONDS, TimeUnit.SECONDS,
                sPoolWorkQueue, sThreadFactory);
        threadPoolExecutor.allowCoreThreadTimeOut(true);
        THREAD_POOL_EXECUTOR = threadPoolExecutor;
    }

    ...
}

我們可以看到，AsyncTask 的這個(gè)線程池設(shè)計(jì)，是希望在達(dá)到核心線程數(shù)之后，能夠繼續(xù)增加工作線程，最大達(dá)到 CPU_COUNT * 2 + 1 個(gè)線程，所以使用了有界隊(duì)列，限制了任務(wù)隊(duì)列最大數(shù)量為 128 個(gè)。

所以使用 AsyncTask 的并發(fā)線程池的時(shí)候要注意，不適宜短時(shí)間同時(shí)大量觸發(fā)大量任務(wù)的場(chǎng)景。

因?yàn)楫?dāng)核心線程、任務(wù)隊(duì)列、非核心線程全部滿負(fù)荷工作的情況下，下一個(gè)進(jìn)來的任務(wù)會(huì)觸發(fā) ThreaPoolExecutor 的 reject 操作，默認(rèn)會(huì)使用 AbortPolicy 策略，拋出 RejectedExecutionException 異常。