這篇文章大部分都是直接摘抄自《實戰(zhàn)Java高并發(fā)程序設(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()方法. 該方法返回一個固定線程數(shù)量的線程池,該線程池中的線程數(shù)量始終不變,當(dāng)有一個新任務(wù)時,線程池中若有空閑線程,則立即執(zhí)行,若沒有,則新任務(wù)會被暫時存在一個隊列中,得有空閑線程時,便處理在任務(wù)隊列中的任務(wù)
newSingleThreadExecutor()方法,改方法返回一個只有一個線程的線程池,若多余一個任務(wù)被提交到該線程池,任務(wù)會被保存在一個隊伍隊列,帶線程空閑,按先入先出的順序執(zhí)行隊列中的任務(wù),
newCachedThreadPool()方法,該方法返回一個可根據(jù)實際情況調(diào)整線程數(shù)量的線程池.線程池數(shù)量是不確定的,但若有空閑線程可以復(fù)用,則會優(yōu)先使用可以復(fù)用的線程,若所有線程均在工作,又有新的任務(wù)提交,則會創(chuàng)建新的線程處理任務(wù),所有線程在當(dāng)前任務(wù)執(zhí)行完畢后,將返回線程池進(jìn)行復(fù)用,
newSingleThreadScheduledExecutor()方法: 改方法返回一個ScheduledExecutorService對象,線程池大小為1 這個接口在ExecutorService接口之上拓展了在給定時間執(zhí)行某任務(wù)的功能,如在某個固定的延時之后執(zhí)行,或者周期性執(zhí)行某個任務(wù).
newScheduledThreadPool()方法:改方法也返回一個ScheduledExecutorService對象 但改線程池可以指定線程數(shù)量
前面三個工廠方法創(chuàng)建的ExecutorService只需要使用ExecutorService.execute()方法或者submit()方法將需要執(zhí)行的任務(wù)傳入即可,這里就不細(xì)講了。關(guān)于這兩個方法的差異我會在后面細(xì)說,這里也不展開討論了。
后面兩個工廠方法會創(chuàng)建ScheduledExecutorService。它有會多出下面三個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()方法會在給定時間,對方法進(jìn)行一次調(diào)度。scheduleAtFixedRate()方法和scheduleWithFixedDelay()會對任務(wù)進(jìn)行周期性調(diào)度。但兩者有一點小小的差別:
對于FixedRate方式來說,任務(wù)調(diào)度的頻率是一樣的。它是以上一個任務(wù)開始執(zhí)行時間為起點,之后的period時間,調(diào)度下一次任務(wù)。而FixDelay則是在上一個任務(wù)結(jié)束后,再經(jīng)過delay時間進(jìn)行任務(wù)調(diào)度。
ThreadPoolExecutor
對于Executors.newFixedThreadPool()、Executors.newSingleThreadExecutor()、Executors.newCachedThreadPool()這幾個方法雖然創(chuàng)建的線程池的功能特點完全不一樣,但是他們其實都是使用了ThreadPoolExecutor實現(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時,多余的空閑線程的存活時間。即,超過corePoolSize的空閑線程,在多長的時間內(nèi),會被銷毀。
- unit: keepAliveTime的時間單位
- workQueue: 被提交但未被執(zhí)行的任務(wù)
- threadFactory: 線程工廠,用于創(chuàng)建線程,一般用默認(rèn)即可
- handler: 拒絕策略。但任務(wù)太多來不及處理,如何拒絕任務(wù)
以上參數(shù)中,大部分都很簡單,只有workQueue和handler需要說一下。
內(nèi)置的BlockingQueue有下面幾種:
SynchronousQueue: 一個沒有容量的隊列。使用SynchronousQueue,提交的任務(wù)不會真正的被保存,而總是將新任務(wù)提交給線程執(zhí)行。如果沒有空閑線程,就創(chuàng)建新線程,如果線程數(shù)量已經(jīng)到達(dá)最大值,則執(zhí)行拒絕策略
ArrayBlockingQueue: 有界任務(wù)隊列,若有新的任務(wù)需要執(zhí)行,如果實際線程數(shù)少于corePoolSize則創(chuàng)建新的線程,如果大于corePoolSize,就會放入ArrayBlockingQueue中,如果ArrayBlockingQueue已滿,在總線程數(shù)不大于maximumPoolSize的情況下會創(chuàng)建新線程,否則就執(zhí)行拒絕策略
LinkedBlockingQueue: 無界任務(wù)隊列,若有新的任務(wù)需要執(zhí)行,如果實際線程數(shù)少于corePoolSize則創(chuàng)建新的線程,如果大于corePoolSize,就會放入LinkedBlockingQueue中等待
PriorityBlockingQueue: 它是一個特殊的無界隊列,可以設(shè)定任務(wù)的優(yōu)先級
而內(nèi)置的拒絕策略又有下面幾種:
- AbortPolicy策略: 該策略會直接拋出異常,阻止系統(tǒng)正常工作
- CallerRunsPolicy策略: 只要線程池沒有關(guān)閉,該策略直接在調(diào)用者線程中運行被拒絕的任務(wù)。(使用這個策略可能導(dǎo)致在主線程執(zhí)行耗時操作)
- DiscardOldestPolicy策略: 該策略丟棄一個最老的任務(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()沒有返回值,只能簡單的提交Runnable給線程池去運行
ExecutorService.submit(),有返回值,可以獲得一個Future
- 異常
ExecutorService.execute()的異常機(jī)制和普通線程的異常機(jī)制一樣,必須用try、catch來捕獲異常。如果沒有捕獲一些運行時異常,也會打印出堆棧信息:
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()的異常會被吃掉,下面的代碼的異常會被默默吃掉,沒有堆棧信息的打印:
Executors.newCachedThreadPool().submit(
new Runnable() {
@Override
public void run() {
int i = 1 / 0;
}
}
);
但是我們可以調(diào)用Future.get()方法,這樣當(dāng)拋出異常的時候系統(tǒng)也會打印堆棧:
Future future = Executors.newCachedThreadPool().submit(
new Runnable() {
@Override
public void run() {
int i = 1 / 0;
}
}
);
future.get();
需要注意的是Future.get()是阻塞的,需要需要等待線程執(zhí)行完畢才會返回,所以我們可以用這個方法獲得Callable.call()的返回值:
Future<Integer> future = Executors.newCachedThreadPool().submit(
new Callable<Integer>() {
@Override
public Integer call() throws Exception {
return 123;
}
}
);
System.out.println(future.get());
