Java工程師知識樹 / Java基礎(chǔ)

為什么使用線程池

池化技術(shù)相比大家已經(jīng)屢見不鮮了种蘸，線程池、數(shù)據(jù)庫連接池、Http 連接池等等都是對這個思想的應(yīng)用次舌。池化技術(shù)的思想主要是為了減少每次獲取資源的消耗，提高對資源的利用率。

線程池提供了一種限制和管理資源（包括執(zhí)行一個任務(wù)）。每個線程池還維護(hù)一些基本統(tǒng)計(jì)信息睹逃，例如已完成任務(wù)的數(shù)量。

使用線程池的好處：

• 降低資源消耗祷肯。通過重復(fù)利用已創(chuàng)建的線程降低線程創(chuàng)建和銷毀造成的消耗沉填。
• 提高響應(yīng)速度。當(dāng)任務(wù)到達(dá)時佑笋，任務(wù)可以不需要的等到線程創(chuàng)建就能立即執(zhí)行翼闹。
• 提高線程的可管理性。線程是稀缺資源蒋纬，如果無限制的創(chuàng)建猎荠，不僅會消耗系統(tǒng)資源，還會降低系統(tǒng)的穩(wěn)定性蜀备，使用線程池可以進(jìn)行統(tǒng)一的分配关摇，調(diào)優(yōu)和監(jiān)控。

四種線程池

線程池創(chuàng)建最終是通過ThreadPoolExecutor創(chuàng)建的碾阁。根據(jù)構(gòu)造方法傳參數(shù)的不同又可以分為不同的線程池输虱。

ThreadPoolExecutor 類圖結(jié)構(gòu)：

ThreadPoolExecutor構(gòu)造方法為：

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

參數(shù)說明

• corePoolSize：線程池中核心線程數(shù)的數(shù)量

• maximumPoolSize：在線程池中允許存在的最大線程數(shù)

• keepAliveTime：當(dāng)存在的線程數(shù)大于corePoolSize，那么會找到空閑線程去銷毀脂凶，此參數(shù)是設(shè)置空閑多久的線程才被銷毀宪睹。

• unit：時間單位 TimeUnit工具類

• workQueue：工作隊(duì)列，線程池中的當(dāng)前線程數(shù)大于核心線程的話艰猬，那么接下來的任務(wù)會放入到隊(duì)列中

  • ArrayBlockingQueue：基于數(shù)組有界阻塞隊(duì)列
  • LinkedBlockingQueue：基于鏈表阻塞隊(duì)列
  • SynchronousQueue：不存儲元素的阻塞隊(duì)列（讀寫須等待一并進(jìn)行）
  • PriorityBlockingQueue：支持優(yōu)先級的無界隊(duì)列

• threadFactory：在創(chuàng)建線程的時候横堡，通過工廠模式來生產(chǎn)線程埋市。這個參數(shù)就是設(shè)置我們自定義的線程創(chuàng)建工廠冠桃。

• handler：拒絕策略，如果超過了最大線程數(shù)道宅，那么就會執(zhí)行我們設(shè)置的拒絕策略

  • AbortPolicy：直接拋出異常食听。
  • CallerRunsPolicy：只用調(diào)用者所在線程來運(yùn)行任務(wù)胸蛛。
  • DiscardOldestPolicy：丟棄隊(duì)列里最近的一個任務(wù)，并執(zhí)行當(dāng)前任務(wù)樱报。
  • DiscardPolicy：不處理葬项，丟棄掉。

線程池處理邏輯：

1. 前corePoolSize個任務(wù)時迹蛤，來一個任務(wù)就創(chuàng)建一個線程
2. 如果當(dāng)前線程池的線程數(shù)大于了corePoolSize那么接下來再來的任務(wù)就會放入到我們上面設(shè)置的workQueue隊(duì)列中
3. 如果此時workQueue也滿了民珍，那么再來任務(wù)時，就會新建臨時線程盗飒，那么此時如果我們設(shè)置了keepAliveTime或者設(shè)置了allowCoreThreadTimeOut嚷量，那么系統(tǒng)就會進(jìn)行線程的活性檢查，一旦超時便銷毀線程
4. 如果此時線程池中的當(dāng)前線程大于了maximumPoolSize最大線程數(shù)逆趣，那么就會執(zhí)行我們剛才設(shè)置的handler拒絕策略

線程池分類：

newFixedThreadPool---固定大小的線程池

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

使用newFixedThreadPool方法創(chuàng)建出來的線程池為固定大小的線程池蝶溶，可以通過第一個靜態(tài)方法指定線程池的大小，該線程池corePoolSize和maximumPoolSize相等宣渗，阻塞隊(duì)列使用的是LinkedBlockingQueue抖所，理論上大小為整數(shù)最大值。

使用newFixedThreadPool方法創(chuàng)建出來的線程池中的線程數(shù)量始終不變痕囱，當(dāng)有新任務(wù)提交時田轧，線程池中有空閑線程則會立即執(zhí)行，如果沒有咐蝇，則會暫存到阻塞隊(duì)列涯鲁。

對于固定大小的線程池，不存在線程數(shù)量的變有序，同時使用無界的LinkedBlockingQueue來存放執(zhí)行的任務(wù)抹腿。

存在的問題：

當(dāng)任務(wù)提交十分頻繁的時候，LinkedBlockingQueue迅速增大旭寿，存在著耗盡系統(tǒng)資源的問題警绩。而且在線程池空閑時，即線程池中沒有可運(yùn)行任務(wù)時盅称，它也不會釋放工作線程肩祥，還會占用一定的系統(tǒng)資源，需要shutdown缩膝。

ExecutorService fixPool = Executors.newFixedThreadPool(5);
for (int i = 0; i < 10; i++) {
    fixPool.execute(() -> System.out.println(Thread.currentThread().getName() + " : " + atomicInteger.getAndIncrement()));
}
fixPool.shutdown();
// 打印結(jié)果 5個線程
pool-1-thread-1 : 0
pool-1-thread-5 : 4
pool-1-thread-4 : 3
pool-1-thread-3 : 2
pool-1-thread-2 : 1
pool-1-thread-3 : 8
pool-1-thread-4 : 7
pool-1-thread-5 : 6
pool-1-thread-1 : 5
pool-1-thread-2 : 9

newSingleThreadExecutor---單個線程的線程池

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

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

使用newSingleThreadExecutor方法創(chuàng)建的線程池為單個線程線程池混狠，只有一個線程的線程池，阻塞隊(duì)列使用的是LinkedBlockingQueue,若有多余的任務(wù)提交到線程池中疾层，則會被暫存到阻塞隊(duì)列将饺，待空閑時再去執(zhí)行。按照先入先出的順序執(zhí)行任務(wù)。

ExecutorService singlePool = Executors.newSingleThreadExecutor();
for (int i = 0; i < 10; i++) {
    singlePool.execute(() -> System.out.println(Thread.currentThread().getName() + " : " + atomicInteger.getAndIncrement()));
}
singlePool.shutdown();
// 打印結(jié)果  只有一個線程
pool-1-thread-1 : 0
pool-1-thread-1 : 1
pool-1-thread-1 : 2
pool-1-thread-1 : 3
pool-1-thread-1 : 4
pool-1-thread-1 : 5
pool-1-thread-1 : 6
pool-1-thread-1 : 7
pool-1-thread-1 : 8
pool-1-thread-1 : 9

newCachedThreadPool---緩存線程池

java.util.concurrent.Executors
    
public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS, // 緩存的線程默認(rèn)存活時間
                                  new SynchronousQueue<Runnable>());
}

public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,// 緩存的線程默認(rèn)存活時間
                                  new SynchronousQueue<Runnable>(),
                                  threadFactory);
}

使用newCachedThreadPool方法創(chuàng)建的線程池為緩存線程池予弧。

緩存線程池刮吧，緩存的線程默認(rèn)存活60秒。線程的核心池corePoolSize大小為0掖蛤，核心池最大為Integer.MAX_VALUE,阻塞隊(duì)列使用的是SynchronousQueue杀捻。

SynchronousQueue是一個直接提交的阻塞隊(duì)列， SynchronousQueue總會迫使線程池增加新的線程去執(zhí)行新的任務(wù)蚓庭。在沒有任務(wù)執(zhí)行時致讥，當(dāng)線程的空閑時間超過keepAliveTime（60秒），則工作線程將會終止被回收器赞，當(dāng)提交新任務(wù)時拄踪，如果沒有空閑線程，則創(chuàng)建新線程執(zhí)行任務(wù)拳魁，會導(dǎo)致一定的系統(tǒng)開銷惶桐。

存在的問題：

如果同時又大量任務(wù)被提交，而且任務(wù)執(zhí)行的時間不是特別快潘懊，那么線程池便會新增出等量的線程池處理任務(wù)姚糊，這很可能會很快耗盡系統(tǒng)的資源。

ExecutorService cachedPool = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
    cachedPool.execute(() -> System.out.println(Thread.currentThread().getName() + " : " + atomicInteger.getAndIncrement()));
}
//打印結(jié)果
pool-1-thread-1 : 0
pool-1-thread-4 : 3
pool-1-thread-3 : 2
pool-1-thread-2 : 1
pool-1-thread-6 : 5
pool-1-thread-5 : 4
pool-1-thread-7 : 6
pool-1-thread-8 : 7
pool-1-thread-8 : 8     
pool-1-thread-2 : 9

newScheduledThreadPool---定時線程池

java.util.concurrent.Executors
    
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

public static ScheduledExecutorService newScheduledThreadPool(
        int corePoolSize, ThreadFactory threadFactory) {
    return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
}

使用newScheduledThreadPool方法創(chuàng)建的線程池為定時線程池授舟。

定時線程池救恨，可用于周期性地去執(zhí)行任務(wù)，通常用于周期性的同步數(shù)據(jù)释树。

ScheduledExecutorService ses = Executors.newScheduledThreadPool(5);
//5個線程每兩秒執(zhí)行一次 隨機(jī)哪個線程執(zhí)行
ses.scheduleAtFixedRate(() -> System.out.println(Thread.currentThread().getName() + " : " + "執(zhí)行了"), 0, 2, TimeUnit.SECONDS);
打印結(jié)果:  5個線程每兩秒執(zhí)行一次 隨機(jī)哪個線程執(zhí)行
pool-1-thread-1 : 執(zhí)行了
pool-1-thread-1 : 執(zhí)行了
pool-1-thread-2 : 執(zhí)行了
pool-1-thread-1 : 執(zhí)行了
pool-1-thread-3 : 執(zhí)行了
pool-1-thread-3 : 執(zhí)行了
pool-1-thread-3 : 執(zhí)行了
... ...

線程池的使用

線程池不允許使用Executors去創(chuàng)建肠槽，而是通過ThreadPoolExecutor的方式，這樣的處理方式讓寫的同學(xué)更加明確線程池的運(yùn)行規(guī)則奢啥，規(guī)避資源耗盡的風(fēng)險秸仙。

說明：Executors各個方法的弊端：

• 1：newFixedThreadPool和newSingleThreadExecutor:
  ??主要問題是堆積的請求處理隊(duì)列可能會耗費(fèi)非常大的內(nèi)存，甚至OOM桩盲。
• 2：newCachedThreadPool和newScheduledThreadPool:
  ??主要問題是線程數(shù)最大數(shù)是Integer.MAX_VALUE寂纪，可能會創(chuàng)建數(shù)量非常多的線程，甚至OOM赌结。

使用實(shí)例：

package com.thread.study.pool;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class ThreadTest {

    public static void main(String[] args) throws InterruptedException {
        int corePoolSize = 3; // 線程池中核心線程數(shù)的數(shù)量
        int maximumPoolSize = 6;// 在線程池中允許存在的最大線程數(shù)
        long keepAliveTime = 10;// 存活時間
        TimeUnit unit = TimeUnit.SECONDS; // 時間單位 秒
        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(3); // 工作隊(duì)列 基于鏈表阻塞隊(duì)列
        ThreadFactory threadFactory = new NameTreadFactory();// 自定義線程工廠
        RejectedExecutionHandler handler = new MyIgnorePolicy(); // 自定義拒絕策略
        /*
            public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
                              long keepAliveTime, TimeUnit unit,
                              BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)
         */
        ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit,
                workQueue, threadFactory, handler);
        executor.prestartAllCoreThreads(); // 預(yù)啟動所有核心線程
        for (int i = 1; i <= 10; i++) {
            executor.execute(new MyTask(String.valueOf(i)));
        }
        System.out.println("當(dāng)前線程池中存活的線程數(shù)為: "+executor.getActiveCount());
        Thread.sleep(5000);
        System.out.println("當(dāng)前線程池中存活的線程數(shù)為: "+executor.getActiveCount());
        executor.shutdown();
    }

    // 自定義線程工廠
    static class NameTreadFactory implements ThreadFactory {

        private final AtomicInteger mThreadNum = new AtomicInteger(1);

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r, "my-thread-" + mThreadNum.getAndIncrement());
            System.out.println(t.getName() + " has been created");
            return t;
        }
    }

    //拒絕策略
    public static class MyIgnorePolicy implements RejectedExecutionHandler {

        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            doLog(r, e);
        }

        private void doLog(Runnable r, ThreadPoolExecutor e) {
            // 可做日志記錄等
            System.err.println(r.toString() + " rejected!" + " 當(dāng)前線程池中存活的線程數(shù)為: " + e.getActiveCount());
        }
    }

    // 線程
    static class MyTask implements Runnable {
        private String name;

        public MyTask(String name) {
            this.name = name;
        }

        @Override
        public void run() {
            try {
                System.out.println(Thread.currentThread().getName() +" : "+ this.toString() + " is running!");
                Thread.sleep(1000); //讓任務(wù)執(zhí)行慢點(diǎn)
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        public String getName() {
            return name;
        }

        @Override
        public String toString() {
            return "MyTask [name=" + name + "]";
        }
    }
}

打印結(jié)果：

my-thread-1 has been created
my-thread-2 has been created
my-thread-3 has been created
my-thread-4 has been created
my-thread-3 : MyTask [name=3] is running!
my-thread-1 : MyTask [name=1] is running!
my-thread-2 : MyTask [name=2] is running!
my-thread-5 has been created
my-thread-6 has been created
my-thread-4 : MyTask [name=4] is running!
my-thread-5 : MyTask [name=8] is running!
my-thread-6 : MyTask [name=9] is running!
當(dāng)前線程池中存活的線程數(shù)為: 6
MyTask [name=10] rejected! 當(dāng)前線程池中存活的線程數(shù)為: 6
my-thread-3 : MyTask [name=7] is running!
my-thread-2 : MyTask [name=5] is running!
my-thread-1 : MyTask [name=6] is running!
當(dāng)前線程池中存活的線程數(shù)為: 0

線程池使用場景

使用線程池異步操作常見場景

批量操作
批量操作不是實(shí)時顯示效果的操作捞蛋，比如批量導(dǎo)入配置，批量刪除或作廢柬姚，批量導(dǎo)出查詢結(jié)果等拟杉，只需要搭建一個任務(wù)，在任務(wù)中處理即可量承，有錯誤對應(yīng)處理錯誤

日志
異步寫操作對功能影響不大的業(yè)務(wù)邏輯是常見的場景搬设，日志對與功能來說重要性并么有那么高

使用第三方接口
比如發(fā)郵箱啼染，發(fā)短信，發(fā)消息隊(duì)列信息焕梅，推送搜索引擎數(shù)據(jù)，異步調(diào)用外圍接口等處理數(shù)據(jù)卦洽。