Java并發(fā)之線程池Executor框架的深入理解
線程池
無限制的創(chuàng)建線程
若采用"為每個(gè)任務(wù)分配一個(gè)線程"的方式會(huì)存在一些缺陷,尤其是當(dāng)需要?jiǎng)?chuàng)建大量線程時(shí):
- 線程生命周期的開銷非常高
- 資源消耗
- 穩(wěn)定性
引入線程池
任務(wù)是一組邏輯工作單元,線程則是使任務(wù)異步執(zhí)行的機(jī)制。當(dāng)存在大量并發(fā)任務(wù)時(shí),創(chuàng)建、銷毀線程需要很大的開銷,運(yùn)用線程池可以大大減小開銷。
Executor框架
說明:
- Executor 執(zhí)行器接口,該接口定義執(zhí)行Runnable任務(wù)的方式。
- ExecutorService 該接口定義提供對(duì)Executor的服務(wù)。
- ScheduledExecutorService 定時(shí)調(diào)度接口。
- AbstractExecutorService 執(zhí)行框架抽象類。
- ThreadPoolExecutor JDK中線程池的具體實(shí)現(xiàn)。
- Executors 線程池工廠類。
ThreadPoolExecutor線程池類
線程池是一個(gè)復(fù)雜的任務(wù)調(diào)度工具,它涉及到任務(wù)、線程池等的生命周期問題。要配置一個(gè)線程池是比較復(fù)雜的,尤其是對(duì)于線程池的原理不是很清楚的情況下,很有可能配置的線程池不是較優(yōu)的。
JDK中的線程池均由ThreadPoolExecutor類實(shí)現(xiàn)。其構(gòu)造方法如下:
參數(shù)說明:
- corePoolSize:核心線程數(shù)。
- maximumPoolSize:最大線程數(shù)。
- keepAliveTime:線程存活時(shí)間。當(dāng)線程數(shù)大于core數(shù),那么超過該時(shí)間的線程將會(huì)被終結(jié)。
- unit:keepAliveTime的單位。java.util.concurrent.TimeUnit類存在靜態(tài)靜態(tài)屬性:NANOSECONDS、MICROSECONDS、MILLISECONDS、SECONDS
- workQueue:Runnable的阻塞隊(duì)列。若線程池已經(jīng)被占滿,則該隊(duì)列用于存放無法再放入線程池中的Runnable。
另一個(gè)構(gòu)造方法:
該方法在下面的擴(kuò)展部分有更深入的講解。其中handler表示線程池對(duì)拒絕任務(wù)的處理策略。
ThreadPoolExecutor的使用需要注意以下概念:
- 若線程池中的線程數(shù)量小于corePoolSize,即使線程池中的線程都處于空閑狀態(tài),也要?jiǎng)?chuàng)建新的線程來處理被添加的任務(wù)。
- 若線程池中的線程數(shù)量等于 corePoolSize且緩沖隊(duì)列 workQueue未滿,則任務(wù)被放入緩沖隊(duì)列。
- 若線程池中線程的數(shù)量大于corePoolSize且緩沖隊(duì)列workQueue滿,且線程池中的數(shù)量小于maximumPoolSize,則建新的線程來處理被添加的任務(wù)。
- 若線程池中線程的數(shù)量大于corePoolSize且緩沖隊(duì)列workQueue滿,且線程池中的數(shù)量等于maximumPoolSize,那么通過 handler所指定的策略來處理此任務(wù)。
- 當(dāng)線程池中的線程數(shù)量大于corePoolSize時(shí),如果某線程空閑時(shí)間超過keepAliveTime,線程將被終止。
Executors 工廠方法
JDK內(nèi)部提供了五種最常見的線程池。由Executors類的五個(gè)靜態(tài)工廠方法創(chuàng)建。
- newFixedThreadPool(...)
- newSingleThreadExecutor(...)
- newCachedThreadPool(...)
- newScheduledThreadPool(...)
- newSingleThreadScheduledExecutor()
單線程的線程池newSingleThreadExecutor
這個(gè)線程池只有一個(gè)線程在工作,也就是相當(dāng)于單線程串行執(zhí)行所有任務(wù)。
返回單線程的Executor,將多個(gè)任務(wù)交給此Exector時(shí),這個(gè)線程處理完一個(gè)任務(wù)后接著處理下一個(gè)任務(wù),若該線程出現(xiàn)異常,將會(huì)有一個(gè)新的線程來替代。此線程池保證所有任務(wù)的執(zhí)行順序按照任務(wù)的提交順序執(zhí)行。
說明:LinkedBlockingQueue會(huì)無限的添加需要執(zhí)行的Runnable。
創(chuàng)建固定大小的線程池newFixedThreadPool
每次提交一個(gè)任務(wù)就創(chuàng)建一個(gè)線程,直到線程達(dá)到線程池的最大大小。線程池的大小一旦達(dá)到最大值就會(huì)保持不變,如果某個(gè)線程因?yàn)閳?zhí)行異常而結(jié)束,那么線程池會(huì)補(bǔ)充一個(gè)新線程。
public static ExecutorSevice newFixedThreadPool()
返回一個(gè)包含指定數(shù)目線程的線程池,如果任務(wù)數(shù)量多于線程數(shù)目,那么沒有沒有執(zhí)行的任務(wù)必須等待,直到有任務(wù)完成為止。
可緩存的線程池newCachedThreadPool
如果線程池的大小超過了處理任務(wù)所需要的線程,那么就會(huì)回收部分空閑(60秒不執(zhí)行任務(wù))的線程,當(dāng)任務(wù)數(shù)增加時(shí),此線程池又可以智能的添加新線程來處理任務(wù)。此線程池不會(huì)對(duì)線程池大小做限制,線程池大小完全依賴于操作系統(tǒng)(或者說JVM)能夠創(chuàng)建的最大線程大小。
newCachedThreadPool方法創(chuàng)建的線程池可以自動(dòng)的擴(kuò)展線程池的容量。核心線程數(shù)量為0。
SynchronousQueue是個(gè)特殊的隊(duì)列。SynchronousQueue隊(duì)列的容量為0。當(dāng)試圖為SynchronousQueue添加Runnable,則執(zhí)行會(huì)失敗。只有當(dāng)一邊從SynchronousQueue取數(shù)據(jù),一邊向SynchronousQueue添加數(shù)據(jù)才可以成功。SynchronousQueue僅僅起到數(shù)據(jù)交換的作用,并不保存線程。但newCachedThreadPool()方法沒有線程上限。Runable添加到SynchronousQueue會(huì)被立刻取出。
根據(jù)用戶的任務(wù)數(shù)創(chuàng)建相應(yīng)的線程來處理,該線程池不會(huì)對(duì)線程數(shù)目加以限制,完全依賴于JVM能創(chuàng)建線程的數(shù)量,可能引起內(nèi)存不足。
定時(shí)任務(wù)調(diào)度的線程池newScheduledThreadPool
創(chuàng)建一個(gè)大小無限的線程池。此線程池支持定時(shí)以及周期性執(zhí)行任務(wù)的需求。
例:
public class ScheduledThreadPoolTest { public static void main(String[] args) { ScheduledExecutorService ses = Executors.newScheduledThreadPool(10); ses.scheduleWithFixedDelay(new Runnable() { @Override public void run() { try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(new Date()); } }, 1000, 2000, TimeUnit.MILLISECONDS); } }
單線程的定時(shí)任務(wù)調(diào)度線程池newSingleThreadScheduledExecutor
此線程池支持定時(shí)以及周期性執(zhí)行任務(wù)的需求。
Executor接口
Executor是一個(gè)線程執(zhí)行接口。任務(wù)執(zhí)行的主要抽象不是Thead,而是Executor。
public interface Executor{ void executor(Runnable command); }
Executor將任務(wù)的提交過程與執(zhí)行過程解耦,并用Runnable來表示任務(wù)。執(zhí)行的任務(wù)放入run方法中即可,將Runnable接口的實(shí)現(xiàn)類交給線程池的execute方法,作為它的一個(gè)參數(shù)。如果需要給任務(wù)傳遞參數(shù),可以通過創(chuàng)建一個(gè)Runnable接口的實(shí)現(xiàn)類來完成。
Executor可以支持多種不同類型的任務(wù)執(zhí)行策略。
Executor基于生產(chǎn)者消費(fèi)者模式,提交任務(wù)的操作相當(dāng)于生產(chǎn)者,執(zhí)行任務(wù)的線程則相當(dāng)于消費(fèi)者。
ExecutorService接口
線程池接口。ExecutorService在Executor的基礎(chǔ)上增加了一些方法,其中有兩個(gè)核心的方法:
Future<?> submit(Runnable task) <T> Future<T> submit(Callable<T> task)
這兩個(gè)方法都是向線程池中提交任務(wù),它們的區(qū)別在于Runnable在執(zhí)行完畢后沒有結(jié)果,Callable執(zhí)行完畢后有一個(gè)結(jié)果。這在多個(gè)線程中傳遞狀態(tài)和結(jié)果是非常有用的。另外他們的相同點(diǎn)在于都返回一個(gè)Future對(duì)象。Future對(duì)象可以阻塞線程直到運(yùn)行完畢(獲取結(jié)果,如果有的話),也可以取消任務(wù)執(zhí)行,當(dāng)然也能夠檢測(cè)任務(wù)是否被取消或者是否執(zhí)行完畢。
在沒有Future之前我們檢測(cè)一個(gè)線程是否執(zhí)行完畢通常使用Thread.join()或者用一個(gè)死循環(huán)加狀態(tài)位來描述線程執(zhí)行完畢。現(xiàn)在有了更好的方法能夠阻塞線程,檢測(cè)任務(wù)執(zhí)行完畢甚至取消執(zhí)行中或者未開始執(zhí)行的任務(wù)。
ScheduledExecutorService接口
ScheduledExecutorService描述的功能和Timer/TimerTask類似,解決那些需要任務(wù)重復(fù)執(zhí)行的問題。這包括延遲時(shí)間一次性執(zhí)行、延遲時(shí)間周期性執(zhí)行以及固定延遲時(shí)間周期性執(zhí)行等。當(dāng)然了繼承ExecutorService的ScheduledExecutorService擁有ExecutorService的全部特性。
線程池生命周期
線程是有多種執(zhí)行狀態(tài)的,同樣管理線程的線程池也有多種狀態(tài)。JVM會(huì)在所有線程(非后臺(tái)daemon線程)全部終止后才退出,為了節(jié)省資源和有效釋放資源關(guān)閉一個(gè)線程池就顯得很重要。有時(shí)候無法正確的關(guān)閉線程池,將會(huì)阻止JVM的結(jié)束。
線程池Executor是異步的執(zhí)行任務(wù),因此任何時(shí)刻不能夠直接獲取提交的任務(wù)的狀態(tài)。這些任務(wù)有可能已經(jīng)完成,也有可能正在執(zhí)行或者還在排隊(duì)等待執(zhí)行。因此關(guān)閉線程池可能出現(xiàn)一下幾種情況:
- 平緩關(guān)閉:已經(jīng)啟動(dòng)的任務(wù)全部執(zhí)行完畢,同時(shí)不再接受新的任務(wù)。
- 立即關(guān)閉:取消所有正在執(zhí)行和未執(zhí)行的任務(wù)。
另外關(guān)閉線程池后對(duì)于任務(wù)的狀態(tài)應(yīng)該有相應(yīng)的反饋信息。
啟動(dòng)線程池
線程池在構(gòu)造前(new操作)是初始狀態(tài),一旦構(gòu)造完成線程池就進(jìn)入了執(zhí)行狀態(tài)RUNNING。嚴(yán)格意義上講線程池構(gòu)造完成后并沒有線程被立即啟動(dòng),只有進(jìn)行"預(yù)啟動(dòng)"或者接收到任務(wù)的時(shí)候才會(huì)啟動(dòng)線程。
線程池是處于運(yùn)行狀態(tài),隨時(shí)準(zhǔn)備接受任務(wù)來執(zhí)行。
關(guān)閉線程池
線程池運(yùn)行中可以通過shutdown()和shutdownNow()來改變運(yùn)行狀態(tài)。
- shutdown():平緩的關(guān)閉線程池。線程池停止接受新的任務(wù),同時(shí)等待已經(jīng)提交的任務(wù)執(zhí)行完畢,包括那些進(jìn)入隊(duì)列還沒有開始的任務(wù)。shutdown()方法執(zhí)行過程中,線程池處于SHUTDOWN狀態(tài)。
- shutdownNow():立即關(guān)閉線程池。線程池停止接受新的任務(wù),同時(shí)線程池取消所有執(zhí)行的任務(wù)和已經(jīng)進(jìn)入隊(duì)列但是還沒有執(zhí)行的任務(wù)。shutdownNow()方法執(zhí)行過程中,線程池處于STOP狀態(tài)。shutdownNow方法本質(zhì)是調(diào)用Thread.interrupt()方法。但我們知道該方法僅僅是讓線程處于interrupted狀態(tài),并不會(huì)讓線程真正的停止!所以若只調(diào)用或只調(diào)用一次shutdownNow()方法,不一定會(huì)讓線程池中的線程都關(guān)閉掉,線程中必須要有處理interrupt事件的機(jī)制。
線程池結(jié)束
一旦shutdown()或者shutdownNow()執(zhí)行完畢,線程池就進(jìn)入TERMINATED狀態(tài),即線程池就結(jié)束了。
- isTerminating() 如果關(guān)閉后所有任務(wù)都已完成,則返回true。
- isShutdown() 如果此執(zhí)行程序已關(guān)閉,則返回true。
例:使用固定大小的線程池。并將任務(wù)添加到線程池。
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
public class JavaThreadPool {
public static void main(String[] args) {
// 創(chuàng)建一個(gè)可重用固定線程數(shù)的線程池
ExecutorService pool = Executors.newFixedThreadPool(2);
// 創(chuàng)建實(shí)現(xiàn)了Runnable接口對(duì)象,Thread對(duì)象當(dāng)然也實(shí)現(xiàn)了Runnable接口
Thread t1 = new MyThread();
Thread t2 = new MyThread();
Thread t3 = new MyThread();
Thread t4 = new MyThread();
Thread t5 = new MyThread();
// 將線程放入池中進(jìn)行執(zhí)行
pool.execute(t1);
pool.execute(t2);
pool.execute(t3);
pool.execute(t4);
pool.execute(t5);
// 關(guān)閉線程池
pool.shutdown();
}
}
class MyThread extends Thread {
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + "正在執(zhí)行。。。");
}
}
Java線程池?cái)U(kuò)展
ThreadPoolExecutor線程池的執(zhí)行監(jiān)控
ThreadPoolExecutor中定義了三個(gè)空方法,用于監(jiān)控線程的執(zhí)行情況。
ThreadPoolExecutor源碼:
protected void beforeExecute(Thread t, Runnable r) { } protected void afterExecute(Runnable r, Throwable t) { } protected void terminated() { }
例:使用覆蓋方法,定義新的線程池。
public class ExtThreadPoolTest { static class MyTask implements Runnable { public String name; public MyTask(String name) { super(); this.name = name; } @Override public void run() { try { Thread.sleep(500); System.out.println("執(zhí)行中:"+this.name); Thread.sleep(500); } catch (InterruptedException e) { e.printStackTrace(); } } } public static void main(String[] args) throws InterruptedException { ExecutorService es = new ThreadPoolExecutor(5,5,0,TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()){ @Override protected void beforeExecute(Thread t, Runnable r) { System.out.println("準(zhǔn)備執(zhí)行:" + ((MyTask)r).name); } @Override protected void afterExecute(Runnable r, Throwable t) { System.out.println("執(zhí)行完成:" + ((MyTask)r).name); } @Override protected void terminated() { System.out.println("執(zhí)行退出"); } }; for(int i=0;i<5;i++){ MyTask task = new MyTask("Task-"+i); es.execute(task); } Thread.sleep(10); // 等待terminated()執(zhí)行 es.shutdown(); // 若無該方法,主線程不會(huì)結(jié)束。 } }
ThreadPoolExecutor的拒絕策略
線程池不可能處理無限多的線程。所以一旦線程池中中需要執(zhí)行的任務(wù)過多,線程池對(duì)于某些任務(wù)就無法處理了。拒絕策略即對(duì)這些無法處理的任務(wù)進(jìn)行處理??赡軄G棄掉這些不能處理的任務(wù),也可能用其他方式。
ThreadPoolExecutor類還有另一個(gè)構(gòu)造方法。該構(gòu)造方法中的RejectedExecutionHandler用于定義拒絕策略。
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { ..... }
JDK內(nèi)部已經(jīng)提供一些拒絕策略。
AbortPolicy一旦線程不能處理,則拋出異常。
AbortPolicy源碼:
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()); } }
DiscardPolicy 一旦線程不能處理,則丟棄任務(wù)。
DiscardPolicy源碼:
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) { } }
CallerRunsPolicy 一旦線程不能處理,則將任務(wù)返回給提交任務(wù)的線程處理。
CallerRunsPolicy源碼:
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(); } } }
DiscardOldestPolicy 一旦線程不能處理,丟棄掉隊(duì)列中最老的任務(wù)。
DiscardOldestPolicy源碼:
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); } } }
例:自定義拒絕策略。打印并丟棄無法處理的任務(wù)。
public class RejectedPolicyHandleTest { public static void main(String[] args) throws InterruptedException { ExecutorService es = new ThreadPoolExecutor(5,5,0,TimeUnit.MILLISECONDS,new SynchronousQueue<Runnable>(),Executors.defaultThreadFactory(),new RejectedExecutionHandler() { @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { // 打印并丟棄。 System.out.println(r.toString()+" is discard"); } }); for(int i=0;i<Integer.MAX_VALUE;i++){ MyTask task = new MyTask("Task-"+i); es.execute(task); Thread.sleep(10); } es.shutdown(); // 若無該方法,主線程不會(huì)結(jié)束。 } }
ThreadFactory 線程工廠
ThreadPoolExecutor類構(gòu)造器的參數(shù)其中之一即為ThreadFactory線程工廠。
ThreadFactory用于創(chuàng)建線程池中的線程。
public interface ThreadFactory { Thread newThread(Runnable r); }
ThreadFactory的實(shí)現(xiàn)類中一般定義線程了線程組,線程數(shù)與線程名稱。
DefaultThreadFactory源碼:
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; } }
CompletionService接口
這里需要稍微提一下的是CompletionService接口,它是用于描述順序獲取執(zhí)行結(jié)果的一個(gè)線程池包裝器。它依賴一個(gè)具體的線程池調(diào)度,但是能夠根據(jù)任務(wù)的執(zhí)行先后順序得到執(zhí)行結(jié)果,這在某些情況下可能提高并發(fā)效率。
以上就是本文的全部?jī)?nèi)容,希望對(duì)大家的學(xué)習(xí)有所幫助,也希望大家多多支持腳本之家。
相關(guān)文章
使用springboot打包成zip部署,并實(shí)現(xiàn)優(yōu)雅停機(jī)
這篇文章主要介紹了使用springboot打包成zip部署,并實(shí)現(xiàn)優(yōu)雅停機(jī),具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2021-08-08java中for循環(huán)執(zhí)行的順序圖文詳析
關(guān)于java的for循環(huán)想必大家非常熟悉,它是java常用的語句之一,這篇文章主要給大家介紹了關(guān)于java中for循環(huán)執(zhí)行順序的相關(guān)資料,需要的朋友可以參考下2021-06-06java面試題之try中含return語句時(shí)代碼的執(zhí)行順序詳解
這篇文章主要介紹了關(guān)于java中的一道面試題,這套題就是在try中含return語句時(shí)代碼的執(zhí)行順序,這個(gè)問題看似簡(jiǎn)單,卻暗藏殺機(jī)??!文中通過一個(gè)個(gè)例子詳細(xì)介紹了其中玄機(jī),需要的朋友可以參考學(xué)習(xí),下面來一起看看吧。2017-04-04關(guān)于JDK15的新特性之TextBlocks文本塊的引入和使用
這篇文章主要介紹了關(guān)于JDK15的新特性之文本塊的引入和使用,如果具有一種語言學(xué)機(jī)制,可以比多行文字更直觀地表示字符串,而且可以跨越多行,而且不會(huì)出現(xiàn)轉(zhuǎn)義的視覺混亂,那么這將提高廣泛Java類程序的可讀性和可寫性,需要的朋友可以參考下2023-07-07springboot登錄攔截器+ThreadLocal實(shí)現(xiàn)用戶信息存儲(chǔ)的實(shí)例代碼
ThreadLocal 為變量在每個(gè)線程中創(chuàng)建了一個(gè)副本,這樣每個(gè)線程都可以訪問自己內(nèi)部的副本變量,這篇文章主要介紹了springboot登錄攔截器+ThreadLocal實(shí)現(xiàn)用戶信息存儲(chǔ)的實(shí)例代碼,需要的朋友可以參考下2024-03-03Java實(shí)現(xiàn)的計(jì)算稀疏矩陣余弦相似度示例
這篇文章主要介紹了Java實(shí)現(xiàn)的計(jì)算稀疏矩陣余弦相似度功能,涉及java基于HashMap的數(shù)值計(jì)算相關(guān)操作技巧,需要的朋友可以參考下2018-07-07