這篇文章大部分都是直接摘抄自《實(shí)戰(zhàn)Java高并發(fā)程序設(shè)計(jì)》,基本上就是一篇筆記,用于以后忘了的時(shí)候可以回顧。
框架提供的ExecutorService
Executors框架提供了各種類型的線程池,主要有以下工廠方法:
public static ExecutorService newFixedThreadPool(int nThreads)
public static ExecutorService newSingleThreadExecutor()
public static ExecutorService newCachedThreadPool()
public static ScheduledExecutorService newSingleThreadScheduledExecutor()
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize)
newFixedThreadPool()方法. 該方法返回一個(gè)固定線程數(shù)量的線程池,該線程池中的線程數(shù)量始終不變,當(dāng)有一個(gè)新任務(wù)時(shí),線程池中若有空閑線程,則立即執(zhí)行,若沒有,則新任務(wù)會(huì)被暫時(shí)存在一個(gè)隊(duì)列中,得有空閑線程時(shí),便處理在任務(wù)隊(duì)列中的任務(wù)
newSingleThreadExecutor()方法,改方法返回一個(gè)只有一個(gè)線程的線程池,若多余一個(gè)任務(wù)被提交到該線程池,任務(wù)會(huì)被保存在一個(gè)隊(duì)伍隊(duì)列,帶線程空閑,按先入先出的順序執(zhí)行隊(duì)列中的任務(wù),
newCachedThreadPool()方法,該方法返回一個(gè)可根據(jù)實(shí)際情況調(diào)整線程數(shù)量的線程池.線程池?cái)?shù)量是不確定的,但若有空閑線程可以復(fù)用,則會(huì)優(yōu)先使用可以復(fù)用的線程,若所有線程均在工作,又有新的任務(wù)提交,則會(huì)創(chuàng)建新的線程處理任務(wù),所有線程在當(dāng)前任務(wù)執(zhí)行完畢后,將返回線程池進(jìn)行復(fù)用,
newSingleThreadScheduledExecutor()方法: 改方法返回一個(gè)ScheduledExecutorService對(duì)象,線程池大小為1 這個(gè)接口在ExecutorService接口之上拓展了在給定時(shí)間執(zhí)行某任務(wù)的功能,如在某個(gè)固定的延時(shí)之后執(zhí)行,或者周期性執(zhí)行某個(gè)任務(wù).
newScheduledThreadPool()方法:改方法也返回一個(gè)ScheduledExecutorService對(duì)象 但改線程池可以指定線程數(shù)量
前面三個(gè)工廠方法創(chuàng)建的ExecutorService只需要使用ExecutorService.execute()方法或者submit()方法將需要執(zhí)行的任務(wù)傳入即可,這里就不細(xì)講了逊桦。關(guān)于這兩個(gè)方法的差異我會(huì)在后面細(xì)說,這里也不展開討論了毙玻。
后面兩個(gè)工廠方法會(huì)創(chuàng)建ScheduledExecutorService。它有會(huì)多出下面三個(gè)schedule方法用于延遲執(zhí)行任務(wù):
public ScheduledFuture<?> schedule(Runnable command,
long delay, TimeUnit unit);
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
long initialDelay,
long period,
TimeUnit unit);
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
long initialDelay,
long delay,
TimeUnit unit);
schedule()方法會(huì)在給定時(shí)間,對(duì)方法進(jìn)行一次調(diào)度缘揪。scheduleAtFixedRate()方法和scheduleWithFixedDelay()會(huì)對(duì)任務(wù)進(jìn)行周期性調(diào)度。但兩者有一點(diǎn)小小的差別:
對(duì)于FixedRate方式來說,任務(wù)調(diào)度的頻率是一樣的。它是以上一個(gè)任務(wù)開始執(zhí)行時(shí)間為起點(diǎn),之后的period時(shí)間,調(diào)度下一次任務(wù)涡真。而FixDelay則是在上一個(gè)任務(wù)結(jié)束后,再經(jīng)過delay時(shí)間進(jìn)行任務(wù)調(diào)度。
ThreadPoolExecutor
對(duì)于Executors.newFixedThreadPool()肾筐、Executors.newSingleThreadExecutor()哆料、Executors.newCachedThreadPool()這幾個(gè)方法雖然創(chuàng)建的線程池的功能特點(diǎn)完全不一樣,但是他們其實(shí)都是使用了ThreadPoolExecutor實(shí)現(xiàn):
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
ThreadPoolExecutor的最重要的構(gòu)造函數(shù)如下:
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler)
- corePoolSize: 指定了線程池中的線程數(shù)量
- maximumPoolSize: 指定了線程池中的最大線程數(shù)量
- keepAliveTime: 當(dāng)線程池線程數(shù)量超過corePoolSize時(shí),多余的空閑線程的存活時(shí)間。即,超過corePoolSize的空閑線程,在多長(zhǎng)的時(shí)間內(nèi),會(huì)被銷毀吗铐。
- unit: keepAliveTime的時(shí)間單位
- workQueue: 被提交但未被執(zhí)行的任務(wù)
- threadFactory: 線程工廠,用于創(chuàng)建線程,一般用默認(rèn)即可
- handler: 拒絕策略东亦。但任務(wù)太多來不及處理,如何拒絕任務(wù)
以上參數(shù)中,大部分都很簡(jiǎn)單,只有workQueue和handler需要說一下。
內(nèi)置的BlockingQueue有下面幾種:
SynchronousQueue: 一個(gè)沒有容量的隊(duì)列唬渗。使用SynchronousQueue,提交的任務(wù)不會(huì)真正的被保存,而總是將新任務(wù)提交給線程執(zhí)行典阵。如果沒有空閑線程,就創(chuàng)建新線程,如果線程數(shù)量已經(jīng)到達(dá)最大值,則執(zhí)行拒絕策略
ArrayBlockingQueue: 有界任務(wù)隊(duì)列,若有新的任務(wù)需要執(zhí)行,如果實(shí)際線程數(shù)少于corePoolSize則創(chuàng)建新的線程,如果大于corePoolSize,就會(huì)放入ArrayBlockingQueue中,如果ArrayBlockingQueue已滿,在總線程數(shù)不大于maximumPoolSize的情況下會(huì)創(chuàng)建新線程,否則就執(zhí)行拒絕策略
LinkedBlockingQueue: 無界任務(wù)隊(duì)列,若有新的任務(wù)需要執(zhí)行,如果實(shí)際線程數(shù)少于corePoolSize則創(chuàng)建新的線程,如果大于corePoolSize,就會(huì)放入LinkedBlockingQueue中等待
PriorityBlockingQueue: 它是一個(gè)特殊的無界隊(duì)列,可以設(shè)定任務(wù)的優(yōu)先級(jí)
而內(nèi)置的拒絕策略又有下面幾種:
- AbortPolicy策略: 該策略會(huì)直接拋出異常,阻止系統(tǒng)正常工作
- CallerRunsPolicy策略: 只要線程池沒有關(guān)閉,該策略直接在調(diào)用者線程中運(yùn)行被拒絕的任務(wù)。(使用這個(gè)策略可能導(dǎo)致在主線程執(zhí)行耗時(shí)操作)
- DiscardOldestPolicy策略: 該策略丟棄一個(gè)最老的任務(wù),并嘗試重新提交任務(wù)
- DiscardPolicy策略: 該策略默默丟棄拒絕的任務(wù),不做任何處理镊逝。
線程池任務(wù)調(diào)度的邏輯如下圖所示:
execute和submit的區(qū)別
ExecutorService.execute()和ExecutorService.submit()都可以提交任務(wù)去異步執(zhí)行,但是它們之間有什么區(qū)別呢壮啊？
void execute(Runnable command);
Future<?> submit(Runnable task);
<T> Future<T> submit(Callable<T> task);
- 返回值
ExecutorService.execute()沒有返回值,只能簡(jiǎn)單的提交Runnable給線程池去運(yùn)行
ExecutorService.submit(),有返回值,可以獲得一個(gè)Future
- 異常
ExecutorService.execute()的異常機(jī)制和普通線程的異常機(jī)制一樣,必須用try、catch來捕獲異常撑蒜。如果沒有捕獲一些運(yùn)行時(shí)異常,也會(huì)打印出堆棧信息:
Executors.newCachedThreadPool().execute(
new Runnable() {
@Override
public void run() {
int i = 1 / 0;
}
}
);
Exception in thread "pool-1-thread-1" java.lang.ArithmeticException: / by zero
ExecutorService.submit()的異常會(huì)被吃掉,下面的代碼的異常會(huì)被默默吃掉,沒有堆棧信息的打印:
Executors.newCachedThreadPool().submit(
new Runnable() {
@Override
public void run() {
int i = 1 / 0;
}
}
);
但是我們可以調(diào)用Future.get()方法,這樣當(dāng)拋出異常的時(shí)候系統(tǒng)也會(huì)打印堆棧:
Future future = Executors.newCachedThreadPool().submit(
new Runnable() {
@Override
public void run() {
int i = 1 / 0;
}
}
);
future.get();
需要注意的是Future.get()是阻塞的,需要需要等待線程執(zhí)行完畢才會(huì)返回,所以我們可以用這個(gè)方法獲得Callable.call()的返回值:
Future<Integer> future = Executors.newCachedThreadPool().submit(
new Callable<Integer>() {
@Override
public Integer call() throws Exception {
return 123;
}
}
);
System.out.println(future.get());
