使用 Executors 創(chuàng)建線程池

1.newFixedThreadPool()

由于使用了LinkedBlockingQueue所以maximumPoolSize沒用恨胚，當(dāng)corePoolSize滿了之后就加入到LinkedBlockingQueue隊(duì)列中瘤缩。
每當(dāng)某個線程執(zhí)行完成之后就從LinkedBlockingQueue隊(duì)列中取一個。
所以這個是創(chuàng)建固定大小的線程池衫冻。

源碼分析:

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

2.newSingleThreadPool()

創(chuàng)建線程數(shù)為1的線程池钮追，由于使用了LinkedBlockingQueue所以maximumPoolSize 沒用晃财，corePoolSize為1表示線程數(shù)大小為1,滿了就放入隊(duì)列中，執(zhí)行完了就從隊(duì)列取一個建炫。

源碼分析

public static ExecutorService newSingleThreadExecutor() {
    return new Executors.FinalizableDelegatedExecutorService
            (
                    new ThreadPoolExecutor(
                            1,
                            1,
                            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);
}

3.newCachedThreadPool()

創(chuàng)建可緩沖的線程池畦韭。沒有大小限制。由于corePoolSize為0所以任務(wù)會放入SynchronousQueue隊(duì)列中肛跌，SynchronousQueue只能存放大小為1艺配，所以會立刻新起線程察郁，由于maxumumPoolSize為Integer.MAX_VALUE所以可以認(rèn)為大小為2147483647。受內(nèi)存大小限制妒挎。

源碼分析

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(
            0,
            Integer.MAX_VALUE,
            60L,
            TimeUnit.SECONDS,
            new SynchronousQueue<Runnable>());
}

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

使用 ThreadPoolExecutor 創(chuàng)建線程池

源碼分析 ,ThreadPoolExecutor 的構(gòu)造函數(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.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

構(gòu)造函數(shù)參數(shù)

1绳锅、corePoolSize 核心線程數(shù)大小，當(dāng)線程數(shù) < corePoolSize 酝掩，會創(chuàng)建線程執(zhí)行 runnable

2鳞芙、maximumPoolSize 最大線程數(shù)， 當(dāng)線程數(shù) >= corePoolSize的時候期虾，會把 runnable 放入 workQueue中

3原朝、keepAliveTime 保持存活時間，當(dāng)線程數(shù)大于corePoolSize的空閑線程能保持的最大時間镶苞。

4喳坠、unit 時間單位

5、workQueue 保存任務(wù)的阻塞隊(duì)列

6茂蚓、threadFactory 創(chuàng)建線程的工廠

7壕鹉、handler 拒絕策略

任務(wù)執(zhí)行順序

1、當(dāng)線程數(shù)小于 corePoolSize時聋涨，創(chuàng)建線程執(zhí)行任務(wù)晾浴。

2、當(dāng)線程數(shù)大于等于 corePoolSize并且 workQueue 沒有滿時牍白，放入workQueue中

3脊凰、線程數(shù)大于等于 corePoolSize并且當(dāng) workQueue 滿時，新任務(wù)新建線程運(yùn)行茂腥，線程總數(shù)要小于 maximumPoolSize

4狸涌、當(dāng)線程總數(shù)等于 maximumPoolSize 并且 workQueue 滿了的時候執(zhí)行 handler 的 rejectedExecution。也就是拒絕策略最岗。

JDK7提供了7個阻塞隊(duì)列帕胆。（也屬于并發(fā)容器）

1、 ArrayBlockingQueue ：一個由數(shù)組結(jié)構(gòu)組成的有界阻塞隊(duì)列仑性。
2惶楼、LinkedBlockingQueue ：一個由鏈表結(jié)構(gòu)組成的有界阻塞隊(duì)列。
3诊杆、PriorityBlockingQueue ：一個支持優(yōu)先級排序的無界阻塞隊(duì)列歼捐。
4、DelayQueue：一個使用優(yōu)先級隊(duì)列實(shí)現(xiàn)的無界阻塞隊(duì)列晨汹。
5豹储、SynchronousQueue：一個不存儲元素的阻塞隊(duì)列。
6淘这、LinkedTransferQueue：一個由鏈表結(jié)構(gòu)組成的無界阻塞隊(duì)列剥扣。
7巩剖、LinkedBlockingDeque：一個由鏈表結(jié)構(gòu)組成的雙向阻塞隊(duì)列。

什么是阻塞隊(duì)列钠怯？

阻塞隊(duì)列是一個在隊(duì)列基礎(chǔ)上又支持了兩個附加操作的隊(duì)列佳魔。

2個附加操作：

支持阻塞的插入方法：隊(duì)列滿時，隊(duì)列會阻塞插入元素的線程晦炊，直到隊(duì)列不滿鞠鲜。
支持阻塞的移除方法：隊(duì)列空時，獲取元素的線程會等待隊(duì)列變?yōu)榉强铡?/p>

阻塞隊(duì)列的應(yīng)用場景

阻塞隊(duì)列常用于生產(chǎn)者和消費(fèi)者的場景断国，生產(chǎn)者是向隊(duì)列里添加元素的線程贤姆，消費(fèi)者是從隊(duì)列里取元素的線程。簡而言之稳衬，阻塞隊(duì)列是生產(chǎn)者用來存放元素霞捡、消費(fèi)者獲取元素的容器。

幾個方法

在阻塞隊(duì)列不可用的時候薄疚，上述2個附加操作提供了四種處理方法

圖片.png

JAVA里的阻塞隊(duì)列

JDK 7 提供了7個阻塞隊(duì)列碧信，如下

1、ArrayBlockingQueue 數(shù)組結(jié)構(gòu)組成的有界阻塞隊(duì)列街夭。

此隊(duì)列按照先進(jìn)先出（FIFO）的原則對元素進(jìn)行排序音婶，但是默認(rèn)情況下不保證線程公平的訪問隊(duì)列，即如果隊(duì)列滿了莱坎，那么被阻塞在外面的線程對隊(duì)列訪問的順序是不能保證線程公平（即先阻塞，先插入）的寸士。

2檐什、LinkedBlockingQueue一個由鏈表結(jié)構(gòu)組成的有界阻塞隊(duì)列

此隊(duì)列按照先出先進(jìn)的原則對元素進(jìn)行排序

3、PriorityBlockingQueue支持優(yōu)先級的無界阻塞隊(duì)列

4弱卡、DelayQueue支持延時獲取元素的無界阻塞隊(duì)列乃正，即可以指定多久才能從隊(duì)列中獲取當(dāng)前元素

5、SynchronousQueue不存儲元素的阻塞隊(duì)列婶博，每一個put必須等待一個take操作瓮具，否則不能繼續(xù)添加元素。并且他支持公平訪問隊(duì)列凡人。

6名党、LinkedTransferQueue由鏈表結(jié)構(gòu)組成的無界阻塞TransferQueue隊(duì)列。相對于其他阻塞隊(duì)列挠轴，多了tryTransfer和transfer方法

transfer方法

如果當(dāng)前有消費(fèi)者正在等待接收元素（take或者待時間限制的poll方法）传睹，transfer可以把生產(chǎn)者傳入的元素立刻傳給消費(fèi)者。如果沒有消費(fèi)者等待接收元素岸晦，則將元素放在隊(duì)列的tail節(jié)點(diǎn)欧啤，并等到該元素被消費(fèi)者消費(fèi)了才返回睛藻。

tryTransfer方法

用來試探生產(chǎn)者傳入的元素能否直接傳給消費(fèi)者。邢隧，如果沒有消費(fèi)者在等待店印，則返回false。和上述方法的區(qū)別是該方法無論消費(fèi)者是否接收倒慧，方法立即返回按摘。而transfer方法是必須等到消費(fèi)者消費(fèi)了才返回。

7迫靖、LinkedBlockingDeque鏈表結(jié)構(gòu)的雙向阻塞隊(duì)列院峡，優(yōu)勢在于多線程入隊(duì)時，減少一半的競爭系宜。

四個拒絕策略

ThreadPoolExecutor默認(rèn)有四個拒絕策略：

1照激、ThreadPoolExecutor.AbortPolicy() 直接拋出異常RejectedExecutionException

2、ThreadPoolExecutor.CallerRunsPolicy() 直接調(diào)用run方法并且阻塞執(zhí)行

3盹牧、ThreadPoolExecutor.DiscardPolicy() 直接丟棄后來的任務(wù)

4俩垃、ThreadPoolExecutor.DiscardOldestPolicy() 丟棄在隊(duì)列中隊(duì)首的任務(wù)

當(dāng)然可以自己繼承RejectedExecutionHandler來寫拒絕策略.
TestThreadPoolExecutor 示例

TestThreadPoolExecutor.java

package io.ymq.thread.TestThreadPoolExecutor;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 描述:
 *
 * @author yanpenglei
 * @create 2017-10-12 15:39
 **/
public class TestThreadPoolExecutor {
    public static void main(String[] args) {

        long currentTimeMillis = System.currentTimeMillis();

        // 構(gòu)造一個線程池
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(5, 6, 3,
                TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(3)
        );

        for (int i = 1; i <= 10; i++) {
            try {
                String task = "task=" + i;
                System.out.println("創(chuàng)建任務(wù)并提交到線程池中：" + task);
                threadPool.execute(new ThreadPoolTask(task));

                Thread.sleep(100);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        try {
            //等待所有線程執(zhí)行完畢當(dāng)前任務(wù)。
            threadPool.shutdown();

            boolean loop = true;
            do {
                //等待所有線程執(zhí)行完畢當(dāng)前任務(wù)結(jié)束
                loop = !threadPool.awaitTermination(2, TimeUnit.SECONDS);//等待2秒
            } while (loop);

            if (loop != true) {
                System.out.println("所有線程執(zhí)行完畢");
            }

        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            System.out.println("耗時：" + (System.currentTimeMillis() - currentTimeMillis));
        }


    }
}

ThreadPoolTask.java

package io.ymq.thread.TestThreadPoolExecutor;

import java.io.Serializable;

/**
 * 描述:
 *
 * @author yanpenglei
 * @create 2017-10-12 15:40
 **/
public class ThreadPoolTask implements Runnable, Serializable {

    private Object attachData;

    ThreadPoolTask(Object tasks) {
        this.attachData = tasks;
    }

    public void run() {

        try {

            System.out.println("開始執(zhí)行任務(wù)：" + attachData + "任務(wù)汰寓，使用的線程池口柳，線程名稱：" + Thread.currentThread().getName());

            System.out.println();

        } catch (Exception e) {
            e.printStackTrace();
        }
        attachData = null;
    }

}

遇到j(luò)ava.util.concurrent.RejectedExecutionException

第一

你的線程池 ThreadPoolExecutor 顯示的 shutdown() 之后，再向線程池提交任務(wù)的時候有滑。 如果你配置的拒絕策略是 AbortPolicy 的話跃闹，這個異常就會拋出來。

第二

當(dāng)你設(shè)置的任務(wù)緩存隊(duì)列過小的時候毛好，或者說望艺， 你的線程池里面所有的線程都在干活（線程數(shù)== maxPoolSize),并且你的任務(wù)緩存隊(duì)列也已經(jīng)充滿了等待的隊(duì)列， 這個時候肌访，你再向它提交任務(wù)找默，則會拋出這個異常。

響應(yīng)

可以看到線程 pool-1-thread-1 到5 循環(huán)使用

創(chuàng)建任務(wù)并提交到線程池中：task=1
開始執(zhí)行任務(wù)：task=1任務(wù)吼驶，使用的線程池惩激，線程名稱：pool-1-thread-1

創(chuàng)建任務(wù)并提交到線程池中：task=2
開始執(zhí)行任務(wù)：task=2任務(wù)，使用的線程池蟹演，線程名稱：pool-1-thread-2

創(chuàng)建任務(wù)并提交到線程池中：task=3
開始執(zhí)行任務(wù)：task=3任務(wù)风钻，使用的線程池，線程名稱：pool-1-thread-3

創(chuàng)建任務(wù)并提交到線程池中：task=4
開始執(zhí)行任務(wù)：task=4任務(wù)酒请，使用的線程池魄咕，線程名稱：pool-1-thread-4

創(chuàng)建任務(wù)并提交到線程池中：task=5
開始執(zhí)行任務(wù)：task=5任務(wù)，使用的線程池蚌父，線程名稱：pool-1-thread-5

創(chuàng)建任務(wù)并提交到線程池中：task=6
開始執(zhí)行任務(wù)：task=6任務(wù)哮兰，使用的線程池毛萌，線程名稱：pool-1-thread-1

創(chuàng)建任務(wù)并提交到線程池中：task=7
開始執(zhí)行任務(wù)：task=7任務(wù)，使用的線程池喝滞，線程名稱：pool-1-thread-2

創(chuàng)建任務(wù)并提交到線程池中：task=8
開始執(zhí)行任務(wù)：task=8任務(wù)阁将，使用的線程池，線程名稱：pool-1-thread-3

創(chuàng)建任務(wù)并提交到線程池中：task=9

轉(zhuǎn)自：https://segmentfault.com/a/1190000011527245