知其然而知所以然

Java線程池早已稱為面試官必考的知識(shí)，作為被面試的你想必不想被這個(gè)簡(jiǎn)單的問題難倒吧豁状，來一起學(xué)習(xí)探索吧~

線程池是什么知态？

通俗可以理解為一個(gè)維護(hù)線程的池子。

為什么需要線程池牲阁？為什么不直接創(chuàng)建使用線程?

進(jìn)程需要執(zhí)行任務(wù)時(shí)固阁，會(huì)將任務(wù)交給線程來處理，所以我們需要?jiǎng)?chuàng)建一個(gè)線程城菊，處理完任務(wù)后备燃，線程需要銷毀，完成一個(gè)線程的生命周期凌唬。 如果需要執(zhí)行多個(gè)任務(wù)并齐，那么就需要?jiǎng)?chuàng)建多個(gè)線程去執(zhí)行，由于創(chuàng)建線程和銷毀的過程比較耗時(shí)客税，所以我們可以用一個(gè)池子來維護(hù)這些線程况褪，需要的時(shí)候去池子里拿，用完了還給池子更耻，線程不會(huì)銷毀测垛，省去了線程重復(fù)創(chuàng)建和銷毀的耗時(shí)操作。

基本使用

Executors
JDK為我們提供了Executors工具來創(chuàng)建和使用線程池秧均，我們可以通過以下方法來創(chuàng)建不同的線程池：

1. newFixedThreadPool

可以設(shè)置固定大小線程的線程池食侮，也是就核心線程數(shù)是固定的

/**
 * 創(chuàng)建一個(gè)固定大小的線程池
 * @param nThreads 線程個(gè)數(shù)
 */
public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

2. newSingleThreadExecutor

/**
 * 創(chuàng)建只有一個(gè)線程的線程池
 */
public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

3. newCachedThreadPool

線程緩存器号涯，因?yàn)榫€程池沒有初始核心線程，同時(shí)使用的是SynchronousQueue阻塞隊(duì)列锯七，所以在有任務(wù)處理的時(shí)候才會(huì)創(chuàng)建線程链快，線程空閑后還能存活60秒

/**
 * 創(chuàng)建一個(gè)類似緩存的線程池
 */
public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

細(xì)心的我們發(fā)現(xiàn)以上三個(gè)方法創(chuàng)建線程池最終都會(huì)通過ThreadPoolExecutor的構(gòu)造方法來創(chuàng)建

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

所以搞清楚這個(gè)構(gòu)造函數(shù)每個(gè)參數(shù)的作用非常重要。

實(shí)現(xiàn)原理

1. 原理流程

線程池的組成

image.png

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

以下流程圖解讀了以上代碼執(zhí)行的流程，這也是線程池執(zhí)行流程的核心

image.png

生命周期

image.png

2. 參數(shù)介紹

以下為創(chuàng)建線程池最終的構(gòu)造方法彩倚，我們想創(chuàng)建一個(gè)符合業(yè)務(wù)的線程池，就需要對(duì)下列的參數(shù)了如指掌扶平。

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

corePoolSize 核心線程數(shù)
maximumPoolSize 線程池最大線程數(shù)
keepAliveTime 非核心線程數(shù)非活躍時(shí)存活的時(shí)間
unit keepAliveTime值的單位
workQueue 阻塞隊(duì)列

threadFactory 線程工廠
handler 拒絕策略

1. AbortPolicy（默認(rèn)策略） 不在接受新的任務(wù)，當(dāng)新的任務(wù)來時(shí)總是拋異常麻献。

/**
     * A handler for rejected tasks that throws a
     * {@code RejectedExecutionException}.
     */
    public static class AbortPolicy implements RejectedExecutionHandler {
        /**
         * Creates an {@code AbortPolicy}.
         */
        public AbortPolicy() { }

        /**
         * Always throws RejectedExecutionException.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         * @throws RejectedExecutionException always
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            throw new RejectedExecutionException("Task " + r.toString() +
                                                 " rejected from " +
                                                 e.toString());
        }
    }

2. CallerRunsPolicy 由調(diào)用線程去執(zhí)行任務(wù)

/**
     * A handler for rejected tasks that runs the rejected task
     * directly in the calling thread of the {@code execute} method,
     * unless the executor has been shut down, in which case the task
     * is discarded.
     */
    public static class CallerRunsPolicy implements RejectedExecutionHandler {
        /**
         * Creates a {@code CallerRunsPolicy}.
         */
        public CallerRunsPolicy() { }

        /**
         * Executes task r in the caller's thread, unless the executor
         * has been shut down, in which case the task is discarded.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                r.run();
            }
        }
    }

3. DiscardPolicy 默默的將任務(wù)丟棄呼巷，啥也不做

/**
     * A handler for rejected tasks that silently discards the
     * rejected task.
     */
    public static class DiscardPolicy implements RejectedExecutionHandler {
        /**
         * Creates a {@code DiscardPolicy}.
         */
        public DiscardPolicy() { }

        /**
         * Does nothing, which has the effect of discarding task r.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        }
    }

4. DiscardOldestPolicy 丟棄隊(duì)列中長時(shí)間未處理的任務(wù)（靠近表頭的任務(wù)），并嘗試將新的任務(wù)放入隊(duì)列中

/**
     * A handler for rejected tasks that discards the oldest unhandled
     * request and then retries {@code execute}, unless the executor
     * is shut down, in which case the task is discarded.
     */
    public static class DiscardOldestPolicy implements RejectedExecutionHandler {
        /**
         * Creates a {@code DiscardOldestPolicy} for the given executor.
         */
        public DiscardOldestPolicy() { }

        /**
         * Obtains and ignores the next task that the executor
         * would otherwise execute, if one is immediately available,
         * and then retries execution of task r, unless the executor
         * is shut down, in which case task r is instead discarded.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                e.getQueue().poll();
                e.execute(r);
            }
        }
    }

閱讀到這赎瑰，你對(duì)線程池一定有深刻的印象，接下來我們看一下實(shí)際開發(fā)正確使用的姿勢(shì)破镰。

實(shí)際應(yīng)用

對(duì)于新手來說餐曼，使用線程池的方式可能如下：

public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 100; i++) {
            int finalI = i;
            executorService.submit(new Runnable() {
                @Override
                public void run() {
                    System.out.println(finalI);
                }
            });
        }
    }

我們通過Executors工具創(chuàng)建一個(gè)固定大小為10的線程池压储，使用的是無界的LinkedBlockingQueue阻塞隊(duì)列，以上代碼看起來沒有任何問題源譬，很快執(zhí)行結(jié)束集惋。
我們稍微改動(dòng)一下：
新增sleep模擬耗時(shí)的操作。

public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        final int[] i = {0};
        while (true){
            executorService.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        Thread.sleep(5000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println(i[0]++);
                }
            });
        }
    }

以上代碼一直執(zhí)行踩娘，最終會(huì)出現(xiàn)產(chǎn)生內(nèi)存溢出的問題刮刑。
因?yàn)閚ewFixedThreadPool使用的是LinkedBlockingQueue（無界阻塞隊(duì)列），當(dāng)線程池源源不斷的接受新的任務(wù)养渴，自己出來不過來的時(shí)候雷绢，會(huì)將任務(wù)暫存到隊(duì)列里，由于處理耗時(shí)理卑，所以阻塞隊(duì)列里的數(shù)據(jù)會(huì)越來越多翘紊，最終把內(nèi)存打爆。

根據(jù)阿里的代碼規(guī)范藐唠，不允許通過Executors來創(chuàng)建線程池帆疟，必須通過new ThreadPoolExcutor構(gòu)造方法來創(chuàng)建線程池，如果你的Idea安裝了阿里的代碼檢查插件宇立，會(huì)看到如下提示：

image.png

所以正確的姿勢(shì)是：

public static void main(String[] args) {

        ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 10, 60,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(100),
                new ThreadPoolExecutor. DiscardPolicy());

        while (true){
            executor.submit(new Runnable() {
                @Override
                public void run() {
                    System.out.println("running");
                }
            });
        }

    }

我們創(chuàng)建了核心線程數(shù)為10踪宠，且最大線程數(shù)為10，阻塞隊(duì)列容量為100妈嘹，拒絕策略為丟棄新的任務(wù) 的線程池柳琢，實(shí)際使用時(shí)還得需要根據(jù)業(yè)務(wù)場(chǎng)景和部署的機(jī)器來設(shè)置線程數(shù)。

不推薦使用Executors來創(chuàng)建線程池蟋滴，通過ThreadPoolExecutor來創(chuàng)建染厅。

題外話

當(dāng)一個(gè)服務(wù)里有一些比較耗時(shí)的操作，比如批量導(dǎo)入津函、導(dǎo)出等耗時(shí)的IO操作肖粮，我們需要將這些操作的線程池和其他業(yè)務(wù)的線程池區(qū)分開，否則我們?cè)谶M(jìn)行導(dǎo)入導(dǎo)出操作時(shí)尔苦，會(huì)影響到其他的業(yè)務(wù)涩馆。所以我們會(huì)定制多個(gè)不同的線程池來處理不同類型的任務(wù)。

以上為個(gè)人的理解允坚，如果差錯(cuò)魂那，煩請(qǐng)不吝嗇指出。