1. ThreadPool線程池

1.1 線程池的使用

線程復(fù)用、控制最大并發(fā)數(shù)、管理線程
- 降低資源消耗廉嚼。通過重復(fù)利用已創(chuàng)建的線程降低線程創(chuàng)建和銷毀造成的銷耗矗烛。
- 提高響應(yīng)速度辅柴。當(dāng)任務(wù)到達(dá)時(shí)，任務(wù)可以不需要等待線程創(chuàng)建就能立即執(zhí)行瞭吃。
- 提高線程的可管理性碌嘀。線程是稀缺資源，如果無限制的創(chuàng)建歪架，不僅會(huì)銷耗系統(tǒng)資源股冗，還會(huì)降低系統(tǒng)的穩(wěn)定性，使用線程池可以進(jìn)行統(tǒng)一的分配和蚪，調(diào)優(yōu)和監(jiān)控止状。
線程池架構(gòu)

1608252489878.png

? Executor，Executors攒霹，ExecutorService怯疤，ThreadPoolExecutor

1.1.1 三大方法

Executors.newFixedThreadPool(int)
- 執(zhí)行長期任務(wù)性能好，創(chuàng)建一個(gè)線程池催束，一池有N個(gè)固定的線程集峦，有固定線程數(shù)的線程
```
public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}
```
- newFixedThreadPool創(chuàng)建的線程池corePoolSize和maximumPoolSize值是相等的，它使用的是LinkedBlockingQueue

Executors.newSingleThreadExecutor()

一個(gè)任務(wù)一個(gè)任務(wù)的執(zhí)行，一池一線程

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}

newSingleThreadExecutor 創(chuàng)建的線程池corePoolSize和maximumPoolSize值都是1少梁，它使用的是LinkedBlockingQueue

Executors.newCachedThreadPool()
- 執(zhí)行很多短期異步任務(wù)洛口，線程池根據(jù)需要?jiǎng)?chuàng)建新線程，但在先前構(gòu)建的線程可用時(shí)將重用它們凯沪〉谘妫可擴(kuò)容，遇強(qiáng)則強(qiáng)
```
public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}
```
- newCachedThreadPool創(chuàng)建的線程池將corePoolSize設(shè)置為0妨马，將maximumPoolSize設(shè)置為Integer.MAX_VALUE挺举，它使用的是SynchronousQueue，也就是說來了任務(wù)就創(chuàng)建線程運(yùn)行烘跺，當(dāng)線程空閑超過60秒湘纵，就銷毀線程。
從以上源碼可以看出滤淳，三大方法底層均是使用ThreadPoolExecutor()來創(chuàng)建線程池
代碼

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 線程池
 * Arrays
 * Collections
 * Executors
 */
public class MyThreadPoolDemo {

    public static void main(String[] args) {
        //List list = new ArrayList();
        //List list = Arrays.asList("a","b");
        //固定數(shù)的線程池梧喷，一池五線程
    //一個(gè)銀行網(wǎng)點(diǎn)，5個(gè)受理業(yè)務(wù)的窗口
    //ExecutorService threadPool =  Executors.newFixedThreadPool(5); 
        //一個(gè)銀行網(wǎng)點(diǎn)脖咐，1個(gè)受理業(yè)務(wù)的窗口
    //ExecutorService threadPool =  Executors.newSingleThreadExecutor(); 
        //一個(gè)銀行網(wǎng)點(diǎn)铺敌，可擴(kuò)展受理業(yè)務(wù)的窗口
        ExecutorService threadPool =  Executors.newCachedThreadPool(); 

        //10個(gè)顧客請求
        try {
            for (int i = 1; i <=10; i++) {
                threadPool.execute(()->{
                    System.out.println(Thread.currentThread().getName()+"\t 辦理業(yè)務(wù)");
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            threadPool.shutdown();
        }
    }
}

工作中均不使用以上三大方法來創(chuàng)建線程池，而是直接使用ThreadPoolExecutor()來創(chuàng)建線程池

1608256649626.png

public static void main(String[] args) {
    ExecutorService threadPool = new ThreadPoolExecutor(
        2,
        5,
        2L,
        TimeUnit.SECONDS,
        new ArrayBlockingQueue<Runnable>(3),
        Executors.defaultThreadFactory(),
        //new ThreadPoolExecutor.AbortPolicy()
        //new ThreadPoolExecutor.CallerRunsPolicy()
        //new ThreadPoolExecutor.DiscardOldestPolicy()
        new ThreadPoolExecutor.DiscardOldestPolicy()
    );
    //10個(gè)顧客請求
    try {
        for (int i = 1; i <= 10; i++) {
            threadPool.execute(() -> {
                System.out.println(Thread.currentThread().getName() + "\t 辦理業(yè)務(wù)");
            });
        }
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        threadPool.shutdown();
    }
}

1.2 七大參數(shù)

ThreadPoolExecutor()源碼

/**
 * Creates a new {@code ThreadPoolExecutor} with the given initial
 * parameters.
 *
 * @param corePoolSize the number of threads to keep in the pool, even
 *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
 * @param maximumPoolSize the maximum number of threads to allow in the
 *        pool
 * @param keepAliveTime when the number of threads is greater than
 *        the core, this is the maximum time that excess idle threads
 *        will wait for new tasks before terminating.
 * @param unit the time unit for the {@code keepAliveTime} argument
 * @param workQueue the queue to use for holding tasks before they are
 *        executed.  This queue will hold only the {@code Runnable}
 *        tasks submitted by the {@code execute} method.
 * @param threadFactory the factory to use when the executor
 *        creates a new thread
 * @param handler the handler to use when execution is blocked
 *        because the thread bounds and queue capacities are reached
 * @throws IllegalArgumentException if one of the following holds:<br>
 *         {@code corePoolSize < 0}<br>
 *         {@code keepAliveTime < 0}<br>
 *         {@code maximumPoolSize <= 0}<br>
 *         {@code maximumPoolSize < corePoolSize}
 * @throws NullPointerException if {@code workQueue}
 *         or {@code threadFactory} or {@code handler} is null
 */
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;
}

七大參數(shù)
- corePoolSize：線程池中的常駐核心線程數(shù)
- maximumPoolSize：線程池中能夠容納同時(shí)執(zhí)行的最大線程數(shù)屁擅，此值必須大于等于1
- keepAliveTime：多余的空閑線程的存活時(shí)間偿凭，當(dāng)前池中線程數(shù)量超過corePoolSize時(shí)，當(dāng)空閑時(shí)間達(dá)到keepAliveTime時(shí)派歌，多余線程會(huì)被銷毀直到只剩下corePoolSize個(gè)線程為止
- unit：keepAliveTime的單位
- workQueue：任務(wù)隊(duì)列弯囊，被提交但尚未被執(zhí)行的任務(wù)
- threadFactory：表示生成線程池中工作線程的線程工廠，用于創(chuàng)建線程胶果，一般默認(rèn)的即可
- handler：拒絕策略匾嘱，表示當(dāng)隊(duì)列滿了，并且工作線程大于等于線程池的最大線程數(shù)（maximumPoolSize）時(shí)如何來拒絕請求執(zhí)行的runnable的策略

1.3 四種拒絕策略

拒絕策略：等待隊(duì)列已經(jīng)排滿了早抠，再也塞不下新任務(wù)了奄毡，同時(shí)，線程池中的max線程也達(dá)到了贝或，無法繼續(xù)為新任務(wù)服務(wù)。這個(gè)是時(shí)候我們就需要拒絕策略機(jī)制合理的處理這個(gè)問題锐秦。
JDK內(nèi)置四種拒絕策略

1608256190928.png

AbortPolicy(默認(rèn))：直接拋出RejectedExecutionException異常阻止系統(tǒng)正常運(yùn)行
CallerRunsPolicy：“調(diào)用者運(yùn)行”一種調(diào)節(jié)機(jī)制咪奖，該策略既不會(huì)拋棄任務(wù)，也不會(huì)拋出異常酱床，而是將某些任務(wù)回退到調(diào)用者羊赵，從而降低新任務(wù)的流量。
DiscardOldestPolicy：拋棄隊(duì)列中等待最久的任務(wù)，然后把當(dāng)前任務(wù)加人隊(duì)列中嘗試再次提交當(dāng)前任務(wù)昧捷。
DiscardPolicy：該策略默默地丟棄無法處理的任務(wù)闲昭，不予任何處理也不拋出異常。如果允許任務(wù)丟失靡挥，這是最好的一種策略序矩。

/**
* new ThreadPoolExecutor.AbortPolicy() // 銀行滿了，還有人進(jìn)來跋破，不處理這個(gè)人的簸淀，拋出異常
* new ThreadPoolExecutor.CallerRunsPolicy() // 哪來的去哪里！
* new ThreadPoolExecutor.DiscardPolicy() //隊(duì)列滿了毒返，丟掉任務(wù)租幕，不會(huì)拋出異常！
* new ThreadPoolExecutor.DiscardOldestPolicy() //隊(duì)列滿了拧簸，嘗試去和最早的競爭劲绪，也不會(huì)拋出異常！
*/

以上內(nèi)置拒絕策略均實(shí)現(xiàn)了RejectedExecutionHandle接口

1.4 線程池底層運(yùn)行原理

1608262423930.png

1608262155163.png

在創(chuàng)建了線程池后盆赤，開始等待請求贾富。
當(dāng)調(diào)用execute()方法添加一個(gè)請求任務(wù)時(shí)，線程池會(huì)做出如下判斷：
- 1如果正在運(yùn)行的線程數(shù)量小于corePoolSize弟劲，那么馬上創(chuàng)建線程運(yùn)行這個(gè)任務(wù)祷安；
- 2如果正在運(yùn)行的線程數(shù)量大于或等于corePoolSize，那么將這個(gè)任務(wù)放入隊(duì)列兔乞；
- 3如果這個(gè)時(shí)候隊(duì)列滿了且正在運(yùn)行的線程數(shù)量還小于maximumPoolSize汇鞭，那么還是要?jiǎng)?chuàng)建非核心線程立刻運(yùn)行這個(gè)任務(wù)(隊(duì)列中的依舊在等待)；
- 4如果隊(duì)列滿了且正在運(yùn)行的線程數(shù)量大于或等于maximumPoolSize庸追，那么線程池會(huì)啟動(dòng)飽和拒絕策略來執(zhí)行霍骄。
當(dāng)一個(gè)線程完成任務(wù)時(shí)，它會(huì)從隊(duì)列中取下一個(gè)任務(wù)來執(zhí)行淡溯。
當(dāng)一個(gè)線程無事可做超過一定的時(shí)間（keepAliveTime）時(shí)读整，線程會(huì)判斷：
- 如果當(dāng)前運(yùn)行的線程數(shù)大于corePoolSize，那么這個(gè)線程就被停掉咱娶。所以線程池的所有任務(wù)完成后米间，它最終會(huì)收縮到corePoolSize的大小。

舉例：

銀行有5個(gè)窗口(maximumPoolSize)膘侮，2個(gè)啟用(corePoolSize)屈糊，3個(gè)暫停服務(wù)，且等待區(qū)有5張椅子供等待使用(workQueue)琼了，開始時(shí)前面進(jìn)來的2個(gè)客戶直接到啟用的2個(gè)窗口辦理業(yè)務(wù)逻锐，后面來的客戶，先到等待區(qū)椅子上等待，當(dāng)?shù)却齾^(qū)5張椅子坐滿后昧诱，又有人進(jìn)來辦業(yè)務(wù)晓淀，于是銀行就啟用另外3個(gè)窗口進(jìn)行服務(wù)，辦理完業(yè)務(wù)的窗口盏档，直接喊等待區(qū)的人去那個(gè)窗口辦理凶掰，當(dāng)5個(gè)窗口都在服務(wù)，且等待區(qū)也滿的時(shí)候妆丘，銀行只能讓保安在門口堵著(RejectedExecutionHandler)锄俄，拒絕后面的人員進(jìn)入(畢竟疫情期間不能擠在一起嘛！Ｉ准稹奶赠！)

import java.util.concurrent.*;

/**
 * @Description TODO
 * @Package: juc.juc
 * @ClassName ThreadPoolDemo
 * @author: wuwb
 * @date: 2020/12/18 9:12
 */
public class ThreadPoolDemo {
    public static void main(String[] args) {
        ExecutorService threadPool = new ThreadPoolExecutor(
                2,
                5,
                2L,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(5),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );
        try {
            for (int i = 1; i <= 10; i++) {
                final int j = i;
                threadPool.execute(()->{
                    System.out.println(Thread.currentThread().getName() + " run " + j);
                    try {
                        TimeUnit.SECONDS.sleep(2);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            threadPool.shutdown();
        }
    }
}
/*
pool-1-thread-1 run 1
pool-1-thread-2 run 2
pool-1-thread-3 run 8
pool-1-thread-4 run 9
pool-1-thread-5 run 10
pool-1-thread-1 run 3
pool-1-thread-4 run 4
pool-1-thread-2 run 5
pool-1-thread-3 run 6
pool-1-thread-5 run 7

1、2進(jìn)核心線程药有，3毅戈、4、5愤惰、6苇经、7進(jìn)隊(duì)列等待，8宦言、9扇单、10啟用非核心線程先于隊(duì)列中任務(wù)運(yùn)行
*/

*CPU密集型最大線程數(shù)為CPU核數(shù)，CPU核數(shù)Runtime.getRuntime().availableProcessors()奠旺；

Java多線程之線程池

Java多線程之線程池

1. ThreadPool線程池

1.1 線程池的使用

1.1.1 三大方法

1.2 七大參數(shù)

1.3 四種拒絕策略

1.4 線程池底層運(yùn)行原理