一、概念
線程池：創(chuàng)建并維護(hù)一定數(shù)量的空閑線程，當(dāng)有需要執(zhí)行的任務(wù)，就交付給線程池中的一個(gè)線程，任務(wù)執(zhí)行結(jié)束后，該線程也不會(huì)死亡，而是回到線程池中重新變?yōu)榭臻e狀態(tài)。
線程池優(yōu)點(diǎn)：
1、重用線程池中的線程，避免頻繁創(chuàng)建和銷毀線程所帶來的內(nèi)存開銷。
2、有效控制線程的最大并發(fā)數(shù)，避免因線程之間搶占資源而導(dǎo)致的阻塞現(xiàn)象。
3、能夠?qū)€程進(jìn)行簡單的管理，提供定時(shí)執(zhí)行以及指定時(shí)間間隔循環(huán)執(zhí)行等功能。
二、線程池使用
ThreadPoolExecutor是線程池的實(shí)現(xiàn)，先來個(gè)小demo，核心代碼如下：
private void threadPoolTest() {
//創(chuàng)建線程池
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4 , 3, TimeUnit.SECONDS,
new LinkedBlockingDeque<>(3), new ThreadPoolExecutor.DiscardOldestPolicy());
for (int i = 0; i < 5; i++) {
String data = "task@" + i;
Log.i(TAG, data);
//提交任務(wù)
threadPoolExecutor.execute(new ThreadPoolRunnable(data));
}
Log.i(TAG, "start test ---------------");
}
class ThreadPoolRunnable implements Runnable {
private Object taskData;
public ThreadPoolRunnable(Object taskData) {
this.taskData = taskData;
}
@Override
public void run() {
Log.i(TAG, "run - runnableData: " + taskData + " | " + Thread.currentThread());
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
taskData = null;
}
}
三、線程池創(chuàng)建參數(shù)說明
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
}
corePoolSize (int )：核心線程的個(gè)數(shù)。線程池中有兩類線程，核心線程和非核心線程。核心線程默認(rèn)情況下會(huì)一直存在于線程池中，即使這個(gè)核心線程什么都不干（鐵飯碗），而非核心線程如果長時(shí)間的閑置，就會(huì)被銷毀（臨時(shí)工）。如果調(diào)用了線程池的prestartAllCoreThreads()或prestartCoreThread()方法，線程池會(huì)提前創(chuàng)建并啟動(dòng)所有核心線程。
maximumPoolSize (int )：線程池所能容納的最大線程數(shù)。當(dāng)活動(dòng)線程數(shù)達(dá)到這個(gè)數(shù)值后，后續(xù)的新任務(wù)將會(huì)被阻塞。該值等于核心線程數(shù)量 + 非核心線程數(shù)量。如果任務(wù)隊(duì)列使用無界的阻塞隊(duì)列，該參數(shù)沒有什么效果。
有界阻塞隊(duì)列和無界阻塞隊(duì)列含義和區(qū)別：
阻塞隊(duì)列有一個(gè)非常重要的屬性：容量的大小，分為有界和無界兩種。
無界阻塞隊(duì)列：隊(duì)列容量很大，近似無上界，例如 LinkedBlockingQueue 的上限是 Integer.MAX_VALUE，約為 2 的 31 次方，是非常大的一個(gè)數(shù)，可以近似認(rèn)為是無限容量。
有界阻塞隊(duì)列：隊(duì)列容量有上界，例如 ArrayBlockingQueue 如果容量滿了，也不會(huì)擴(kuò)容，所以一旦滿了就無法再往里放數(shù)據(jù)了。
keepAliveTime (long)：非核心線程閑置時(shí)的超時(shí)時(shí)長，超過這個(gè)時(shí)長，非核心線程就會(huì)被回收。如果設(shè)置allowCoreThreadTimeOut(true)，則會(huì)也作用于核心線程。
unit (TimeUnit )：keepAliveTime的時(shí)間單位。
workQueue(BlockingQueue)：線程池中的任務(wù)隊(duì)列，通過線程池的execute方法提交的Runnable對(duì)象存儲(chǔ)在這個(gè)參數(shù)中，遵循先進(jìn)先出原則。
threadFactory(ThreadFactory )：創(chuàng)建線程的工廠 ，用于批量創(chuàng)建線程，統(tǒng)一在創(chuàng)建線程時(shí)設(shè)置一些參數(shù)，如是否守護(hù)線程、線程的優(yōu)先級(jí)等。如果不指定，默認(rèn)使用Executors.defaultThreadFactory() 來創(chuàng)建線程，線程具有相同的NORM_PRIORITY優(yōu)先級(jí)并且是非守護(hù)線程。
private static class DefaultThreadFactory implements ThreadFactory {
private static final AtomicInteger poolNumber = new AtomicInteger(1);
private final ThreadGroup group;
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final String namePrefix;
DefaultThreadFactory() {
SecurityManager s = System.getSecurityManager();
group = (s != null) ? s.getThreadGroup() :
Thread.currentThread().getThreadGroup();
namePrefix = "pool-" +
poolNumber.getAndIncrement() +
"-thread-";
}
public Thread newThread(Runnable r) {
Thread t = new Thread(group, r,
namePrefix + threadNumber.getAndIncrement(),
0);
if (t.isDaemon())
t.setDaemon(false);
if (t.getPriority() != Thread.NORM_PRIORITY)
t.setPriority(Thread.NORM_PRIORITY);
return t;
}
}
- handler(RejectedExecutionHandler )：拒絕處理策略，線程數(shù)量大于最大線程數(shù)就會(huì)采用拒絕處理策略。
四種拒絕處理的策略如下:
- ThreadPoolExecutor.AbortPolicy：默認(rèn)拒絕處理策略，丟棄任務(wù)并拋出RejectedExecutionException異常。
- ThreadPoolExecutor.DiscardPolicy：丟棄新來的任務(wù)，但是不拋出異常。
- ThreadPoolExecutor.DiscardOldestPolicy：丟棄隊(duì)列頭部（最舊的）的任務(wù)，然后重新嘗試執(zhí)行程序（如果再次失敗，重復(fù)此過程）。
- ThreadPoolExecutor.CallerRunsPolicy：由調(diào)用線程處理該任務(wù)。
使用總結(jié)：
demo中的構(gòu)造代碼如下，聲明了2個(gè)核心進(jìn)程，最大進(jìn)程數(shù)為4，非核心進(jìn)程空閑超時(shí)時(shí)長是3s。阻塞隊(duì)列使用LinkedBlockingDeque，并指定大小為3，防止隊(duì)列無限膨脹，拒絕處理策略為DiscardOldestPolicy，丟棄隊(duì)列頭部（最舊的）的任務(wù)。
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4 , 3, TimeUnit.SECONDS,
new LinkedBlockingDeque<>(3), new ThreadPoolExecutor.DiscardOldestPolicy());
四、線程池常用阻塞隊(duì)列(BlockingQueue)
1、ArrayBlockingQueue（常用）
- 基于數(shù)組的阻塞隊(duì)列。ArrayBlockingQueue 內(nèi)部，維護(hù)了一個(gè)定長數(shù)組，以便緩存隊(duì)列中的數(shù)據(jù)對(duì)象，這是一個(gè)常用的阻塞隊(duì)列。
- ArrayBlockingQueue 在生產(chǎn)者放入數(shù)據(jù)和消費(fèi)者獲取數(shù)據(jù)，都是共用同一個(gè)鎖對(duì)象，由此也意味著兩者無法真正并行運(yùn)行，沒有實(shí)現(xiàn)讀寫分離。
- 在創(chuàng)建 ArrayBlockingQueue 時(shí)，我們還可以控制對(duì)象的內(nèi)部鎖是否采用公平鎖，默認(rèn)采用非公平鎖。
ArrayBlockingQueue 使用 ReentrantLock 的公平鎖和非公平鎖來實(shí)現(xiàn)該功能。簡單理解就是，ReentrantLock 內(nèi)部會(huì)維護(hù)一個(gè)有先后順序的等待隊(duì)列，假如有五個(gè)任務(wù)一起過來，都被阻塞了。如果是公平的，則等待隊(duì)列中等待最久的任務(wù)就會(huì)先進(jìn)入阻塞隊(duì)列。如果是非公平的，那么這五個(gè)線程就需要搶鎖，誰先搶到，誰就先進(jìn)入阻塞隊(duì)列。
2、LinkedBlockingQueue（常用）
- 由鏈表結(jié)構(gòu)組成的有界阻塞隊(duì)列。LinkedBlockingQueue 不指定隊(duì)列的大小時(shí)，默認(rèn)值是 Integer.MAX_VALUE 。但是，建議指定一個(gè)固定大小。因?yàn)?，如果生產(chǎn)者的速度比消費(fèi)者的速度大的多的情況下，這會(huì)導(dǎo)致阻塞隊(duì)列一直膨脹，直到系統(tǒng)內(nèi)存被耗盡。
- LinkedBlockingQueue 實(shí)現(xiàn)了讀寫分離，可以實(shí)現(xiàn)數(shù)據(jù)的讀和寫互不影響，這在高并發(fā)的場景下，對(duì)于效率的提高無疑是非常巨大的。
3、SynchronousQueue
- 這是一個(gè)沒有緩沖的無界隊(duì)列，size為0。當(dāng)執(zhí)行插入元素的操作時(shí)，必須等待一個(gè)取出操作。也就是說，put元素的時(shí)候，必須等待 take 操作。
- 適用于并發(fā)任務(wù)不大，而且生產(chǎn)者和消費(fèi)者的速度相差不多的場景下，直接把生產(chǎn)者和消費(fèi)者對(duì)接，不用經(jīng)過隊(duì)列的入隊(duì)出隊(duì)這一系列操作，效率上會(huì)高一些。
- Excutors.newCachedThreadPool 方法用的就是這種隊(duì)列，我們還可以控制對(duì)象的內(nèi)部鎖是否采用公平鎖，默認(rèn)采用非公平鎖。
4、PriorityBlockingQueue
- 這是一個(gè)支持優(yōu)先級(jí)排序的無界隊(duì)列。
- 可以指定初始容量大?。ㄗ⒁獬跏既萘坎⒉淮碜畲笕萘浚?，或者不指定，默認(rèn)大小為 11。也可以傳入一個(gè)比較器，把元素按一定的規(guī)則排序，不指定比較器的話，默認(rèn)是自然順序。
- PriorityBlockingQueue 是基于二叉樹最小堆實(shí)現(xiàn)的，每當(dāng)取元素的時(shí)候，就會(huì)把優(yōu)先級(jí)最高的元素取出來。
5、DelayQueue
- 一個(gè)帶有延遲時(shí)間的無界阻塞隊(duì)列，隊(duì)列中的元素，只有等延時(shí)時(shí)間到了，才能取出來。此隊(duì)列一般用于過期數(shù)據(jù)的刪除，或任務(wù)調(diào)度。
五、線程池常用方法
- execute(Runnable run)：提交任務(wù)，交由線程池調(diào)度；
- shutdown()：關(guān)閉線程池，等待任務(wù)執(zhí)行完成；
- shutdownNow()：關(guān)閉線程池，不等待任務(wù)執(zhí)行完成；
- getTaskCount()：返回線程池找中所有任務(wù)的數(shù)量 (已完成的任務(wù)+阻塞隊(duì)列中的任務(wù))；
- getCompletedTaskCount()：返回線程池中已執(zhí)行完成的任務(wù)數(shù)量 (已完成的任務(wù))；
- getPoolSize()：返回線程池中已創(chuàng)建線程數(shù)量；
- getActiveCount()：返回當(dāng)前正在運(yùn)行的線程數(shù)量；
- terminated()：線程池終止時(shí)執(zhí)行的策略。
六、線程池狀態(tài)
ThreadPoolExecutor類中使用了一些final int常量變量來表示線程池的狀態(tài) :
private static final int RUNNING = -1 << COUNT_BITS;
private static final int SHUTDOWN = 0 << COUNT_BITS;
private static final int STOP = 1 << COUNT_BITS;
private static final int TIDYING = 2 << COUNT_BITS;
private static final int TERMINATED = 3 << COUNT_BITS;
1、RUNNING
- 線程池創(chuàng)建后處于RUNNING狀態(tài)。
- 線程池能夠接收新任務(wù)，也能夠?qū)σ呀?jīng)添加的任務(wù)進(jìn)行處理。
2、SHUTDOWN
- 線程池已經(jīng)被關(guān)閉了，不再接收新任務(wù)；但是，其還是會(huì)處理隊(duì)列中的剩余的任務(wù)。
- 調(diào)用線程池的shutdown()方法后，線程池的狀態(tài)就會(huì)由RUNNING轉(zhuǎn)為SHUTDOWN。
3、STOP：
- 線程池處于STOP狀態(tài)，此時(shí)線程池不再接收新任務(wù)，不處理已經(jīng)添加進(jìn)來的任務(wù)，并且會(huì)中斷正在處理的任務(wù)。
- 調(diào)用線程池的shutdownNow()方法后，線程池的狀態(tài)就會(huì)由RUNNING或SHUTDOWN轉(zhuǎn)為STOP；
4、TIDYING：
- workCount（有效線程數(shù)）為0。
- 線程池被下達(dá)關(guān)閉命令后，如果當(dāng)前所有的任務(wù)都已經(jīng)終止了（這個(gè)終止可以表示執(zhí)行結(jié)束，也可以表示強(qiáng)制中斷，也可以表示被丟棄） ，那么線程就會(huì)進(jìn)入TIDYING狀態(tài)；當(dāng)線程池變?yōu)?strong>TIDYING狀態(tài)時(shí)，會(huì)執(zhí)行鉤子函數(shù)terminated()。terminated()在ThreadPoolExecutor類中是空的，若用戶想在線程池變?yōu)?strong>TIDYING時(shí)進(jìn)行相應(yīng)的處理，可以通過重載terminated()函數(shù)來實(shí)現(xiàn)。
- 如果線程狀態(tài)已經(jīng)是SHUTDOWN了，并且線程中以及隊(duì)列中都沒有任務(wù)時(shí)，線程池就會(huì)由SHUTDOWN轉(zhuǎn)為TIDYING；如果線程池狀態(tài)為STOP，那么當(dāng)線程池把所有的任務(wù)都給清理干凈時(shí)，線程池就會(huì)由STOP轉(zhuǎn)為TIDYING。
5、TERMINATED：
- 線程池就結(jié)束了；線程池就不能重新啟動(dòng)了。
- 如果線程池處于TIDYING狀態(tài)，那么當(dāng)線程池執(zhí)行完terminated()方法后，線程池狀態(tài)就會(huì)由TIDYING轉(zhuǎn)為TERMINTED。
七、線程池分類
1、newFixedThreadPool()
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
- 線程數(shù)量固定且都是核心線程：核心線程數(shù)量和最大線程數(shù)量都是nThreads。
- keepAliveTime = 0L，即使線程池中的線程空閑，也不會(huì)被回收。除非調(diào)用shutDown()或shutDownNow()去關(guān)閉線程池。
- 可以更快響應(yīng)外界的請(qǐng)求，且任務(wù)阻塞隊(duì)列為無邊界隊(duì)列(LinkedBlockingQueue()鏈表結(jié)構(gòu)的阻塞隊(duì)列)，意味著任務(wù)可以無限加入線程池。
- 可控制線程最大并發(fā)數(shù)。
2、newScheduledThreadPool()
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
}
//默認(rèn)閑置超時(shí)回收時(shí)常
private static final long DEFAULT_KEEPALIVE_MILLIS = 10L;
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE,
DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
new DelayedWorkQueue());
}
- 核心線程數(shù)量固定，非核心線程數(shù)量無限制。
- 非核心線程閑置超過10s會(huì)被回收。
- 主要用于執(zhí)行定時(shí)任務(wù)和具有固定周期的重復(fù)任務(wù)。
3、newSingleThreadExecutor()
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
- 核心線程數(shù)為1的線程池。確保所有任務(wù)在同一線程執(zhí)行，因此不用考慮線程同步問題，從命名也可以看出來，其他為ThreadPool線程池，當(dāng)前類型為ThreadExecutor，執(zhí)行線程。（相當(dāng)于newFixedThreadPool傳入線程個(gè)數(shù)是1）。
- 阻塞隊(duì)列是無界的，可以一直接收任務(wù) 。
- 常用于不適合并發(fā)但可能引起IO阻塞性及影響UI線程響應(yīng)的操作，如數(shù)據(jù)庫操作、文件操作等。
4、newCachedThreadPool()
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
- 沒有核心線程，非核心線程無限，但執(zhí)行完任務(wù)后，空閑超過60s則回收空閑線程。
- 使用SynchronousQueue() 隊(duì)列，有新任務(wù)時(shí)使用空閑線程執(zhí)行，沒有空閑線程則創(chuàng)建新的線程來處理。
- 適合執(zhí)行大量的耗時(shí)較少的任務(wù)，當(dāng)所有線程閑置超過60s都會(huì)被停止，所以這時(shí)幾乎不占用系統(tǒng)資源。
八、各類線程池使用
基礎(chǔ)知識(shí)差不多了，可以擼代碼了，以下為核心代碼：
public void fixedThreadPoolTest() {
Log.i(TAG, "----fixedThreadPoolTest----");
ExecutorService fixedThreadPool = Executors.newFixedThreadPool(2);
for (int i = 0; i < 5; i++) {
String data = "task@" + i;
Log.i(TAG, data);
fixedThreadPool.execute(new ThreadPoolRunnable(data));
}
}
public void cachedThreadPoolTest() {
Log.i(TAG, "----cachedThreadPoolTest----");
ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
for (int i = 0; i < 5; i++) {
String data = "task@" + i;
Log.i(TAG, data);
cachedThreadPool.execute(new ThreadPoolRunnable(data));
}
}
public void singleThreadExecutorTest() {
Log.i(TAG, "----singleThreadExecutorTest----");
ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
for (int i = 0; i < 5; i++) {
String data = "task@" + i;
Log.i(TAG, data);
singleThreadExecutor .execute(new ThreadPoolRunnable(data));
}
}
public void scheduledThreadPoolTest() {
Log.i(TAG, "----scheduledThreadPoolTest----");
//這里創(chuàng)建的是ScheduledExecutorService對(duì)象，ScheduledExecutorService是ExecutorService的子類
mScheduledThreadPool = Executors.newScheduledThreadPool(2);
for (int i = 0; i < 5; i++) {
String data = "task@" + i;
Log.i(TAG, data);
// 1000ms后執(zhí)行runnable
mScheduledThreadPool.schedule(new ThreadPoolRunnable(data), 1000, TimeUnit.MILLISECONDS);
}
}
public void scheduledAtFixedRateTest() {
Log.i(TAG, "----scheduledAtFixedRateTest----");
//這里創(chuàng)建的是ScheduledExecutorService對(duì)象，ScheduledExecutorService是ExecutorService的子類
mScheduledThreadPool = Executors.newScheduledThreadPool(2);
// 1s后，每2s執(zhí)行一次runnable
mScheduledThreadPool.scheduleAtFixedRate(new ThreadPoolRunnable(666), 1000, 2000, TimeUnit.MILLISECONDS);
}
public void scheduledCancelTest() {
Log.i(TAG, "----scheduledCancelTest----");
if (mScheduledThreadPool != null) {
mScheduledThreadPool.shutdown();
}
}
class ThreadPoolRunnable implements Runnable {
private Object taskData;
public ThreadPoolRunnable(Object taskData) {
this.taskData = taskData;
}
@Override
public void run() {
Log.i(TAG, "run - runnableData: " + taskData + " | " + Thread.currentThread());
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
taskData = null;
}
}
運(yùn)行結(jié)果：
cachedThreadPoolTest函數(shù)，看下圖結(jié)果可知，一共新建了5個(gè)線程來執(zhí)行這5個(gè)任務(wù)，所以newCachedThreadPoo()只適合耗時(shí)較短的任務(wù)，否則會(huì)一直創(chuàng)建新線程。
參考文章：
線程池10：線程池的5種狀態(tài)；
常用阻塞隊(duì)列 BlockingQueue 有哪些？
Android多線程：理解和簡單使用總結(jié)
Android 線程池
全方位解析-Android中的線程池
