ThreadPoolExecutor機制
一、概述
1、ThreadPoolExecutor作為java.util.concurrent包對外提供基礎實作，以内部線程池的形式對外提供管理任務執行，線程排程，線程池管理等等服務；
2、Executors方法提供的線程服務，都是通過參數設定來實作不同的線程池機制。
3、先來了解其線程池管理的機制，有助于正确使用，避免錯誤使用導緻嚴重故障。同時可以根據自己的需求實作自己的線程池
二、核心構造方法講解
下面是ThreadPoolExecutor最核心的構造方法
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;
}
構造方法參數講解
參數名
作用
corePoolSize
核心線程池大小
maximumPoolSize
最大線程池大小
keepAliveTime
線程池中超過corePoolSize數目的空閑線程最大存活時間；可以allowCoreThreadTimeOut(true)使得核心線程有效時間
TimeUnit
keepAliveTime時間機關
workQueue
阻塞任務隊列
threadFactory
建立線程工廠
RejectedExecutionHandler
當送出任務數超過maxmumPoolSize+workQueue之和時，任務會交給RejectedExecutionHandler來處理
重點講解：
其中比較容易讓人誤解的是：corePoolSize，maximumPoolSize，workQueue之間關系。
1.當線程池小于corePoolSize時，新送出任務将建立一個新線程執行任務，即使此時線程池中存在空閑線程。
2.當線程池達到corePoolSize時，新送出任務将被放入workQueue中，等待線程池中任務排程執行
3.當workQueue已滿，且maximumPoolSize>corePoolSize時，新送出任務會建立新線程執行任務
4.當送出任務數超過maximumPoolSize時，新送出任務由RejectedExecutionHandler處理
5.當線程池中超過corePoolSize線程，空閑時間達到keepAliveTime時，關閉空閑線程
6.當設定allowCoreThreadTimeOut(true)時，線程池中corePoolSize線程空閑時間達到keepAliveTime也将關閉
線程管理機制圖示：
三、Executors提供的線程池配置方案
1、構造一個固定線程數目的線程池，配置的corePoolSize與maximumPoolSize大小相同，同時使用了一個無界LinkedBlockingQueue存放阻塞任務，是以多餘的任務将存在再阻塞隊列，不會由RejectedExecutionHandler處理
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
2、構造一個緩沖功能的線程池，配置corePoolSize=0，maximumPoolSize=Integer.MAX_VALUE，keepAliveTime=60s,以及一個無容量的阻塞隊列 SynchronousQueue，是以任務送出之後，将會建立新的線程執行；線程空閑超過60s将會銷毀
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
3、構造一個隻支援一個線程的線程池，配置corePoolSize=maximumPoolSize=1，無界阻塞隊列LinkedBlockingQueue；保證任務由一個線程串行執行
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
4、構造有定時功能的線程池，配置corePoolSize，無界延遲阻塞隊列DelayedWorkQueue；有意思的是：maximumPoolSize=Integer.MAX_VALUE，由于DelayedWorkQueue是無界隊列，是以這個值是沒有意義的
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
public static ScheduledExecutorService newScheduledThreadPool(
int corePoolSize, ThreadFactory threadFactory) {
return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
public ScheduledThreadPoolExecutor(int corePoolSize,
ThreadFactory threadFactory) {
super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
new DelayedWorkQueue(), threadFactory);
四、定制屬于自己的非阻塞線程池
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class CustomThreadPoolExecutor {
private ThreadPoolExecutor pool = null;
/**
* 線程池初始化方法
*
* corePoolSize 核心線程池大小----10
* maximumPoolSize 最大線程池大小----30
* keepAliveTime 線程池中超過corePoolSize數目的空閑線程最大存活時間----30+機關TimeUnit
* TimeUnit keepAliveTime時間機關----TimeUnit.MINUTES
* workQueue 阻塞隊列----new ArrayBlockingQueue<Runnable>(10)====10容量的阻塞隊列
* threadFactory 建立線程工廠----new CustomThreadFactory()====定制的線程工廠
* rejectedExecutionHandler 當送出任務數超過maxmumPoolSize+workQueue之和時,
* 即當送出第41個任務時(前面線程都沒有執行完,此測試方法中用sleep(100)),
* 任務會交給RejectedExecutionHandler來處理
*/
public void init() {
pool = new ThreadPoolExecutor(
10,
30,
TimeUnit.MINUTES,
new ArrayBlockingQueue<Runnable>(10),
new CustomThreadFactory(),
new CustomRejectedExecutionHandler());
public void destory() {
if(pool != null) {
pool.shutdownNow();
}
public ExecutorService getCustomThreadPoolExecutor() {
return this.pool;
private class CustomThreadFactory implements ThreadFactory {
private AtomicInteger count = new AtomicInteger(0);
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r);
String threadName = CustomThreadPoolExecutor.class.getSimpleName() + count.addAndGet(1);
System.out.println(threadName);
t.setName(threadName);
return t;
private class CustomRejectedExecutionHandler implements RejectedExecutionHandler {
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
// 記錄異常
// 報警處理等
System.out.println("error.............");
// 測試構造的線程池
public static void main(String[] args) {
CustomThreadPoolExecutor exec = new CustomThreadPoolExecutor();
// 1.初始化
exec.init();
ExecutorService pool = exec.getCustomThreadPoolExecutor();
for(int i=1; i<100; i++) {
System.out.println("送出第" + i + "個任務!");
pool.execute(new Runnable() {
@Override
public void run() {
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("running=====");
}
});
// 2.銷毀----此處不能銷毀,因為任務沒有送出執行完,如果銷毀線程池,任務也就無法執行了
// exec.destory();
try {
Thread.sleep(10000);
} catch (InterruptedException e) {
e.printStackTrace();
}
方法中建立一個核心線程數為30個，緩沖隊列有10個的線程池。每個線程任務，執行時會先睡眠3秒，保證送出10任務時，線程數目被占用完，再送出30任務時，阻塞隊列被占用完，，這樣送出第41個任務是，會交給CustomRejectedExecutionHandler 異常處理類來處理。
送出任務的代碼如下：
public void execute(Runnable command) {
if (command == null)
/*
* Proceed in 3 steps:
*
* 1. If fewer than corePoolSize threads are running, try to
* start a new thread with the given command as its first
* task. The call to addWorker atomically checks runState and
* workerCount, and so prevents false alarms that would add
* threads when it shouldn't, by returning false.
* 2. If a task can be successfully queued, then we still need
* to double-check whether we should have added a thread
* (because existing ones died since last checking) or that
* the pool shut down since entry into this method. So we
* recheck state and if necessary roll back the enqueuing if
* stopped, or start a new thread if there are none.
* 3. If we cannot queue task, then we try to add a new
* thread. If it fails, we know we are shut down or saturated
* and so reject the task.
*/
int c = ctl.get();
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
else if (!addWorker(command, false))
reject(command);
注意：41以後送出的任務就不能正常處理了，因為，execute中送出到任務隊列是用的offer方法，如上面代碼，這個方法是非阻塞的，是以就會交給CustomRejectedExecutionHandler 來處理，是以對于大資料量的任務來說，這種線程池，如果不設定隊列長度會OOM，設定隊列長度，會有任務得不到處理，接下來我們建構一個阻塞的自定義線程池
五、定制屬于自己的阻塞線程池
package com.tongbanjie.trade.test.commons;
public class CustomThreadPoolExecutor {
private ThreadPoolExecutor pool = null;
/**
* 線程池初始化方法
*
* corePoolSize 核心線程池大小----1
* maximumPoolSize 最大線程池大小----3
* keepAliveTime 線程池中超過corePoolSize數目的空閑線程最大存活時間----30+機關TimeUnit
* TimeUnit keepAliveTime時間機關----TimeUnit.MINUTES
* workQueue 阻塞隊列----new ArrayBlockingQueue<Runnable>(5)====5容量的阻塞隊列
* threadFactory 建立線程工廠----new CustomThreadFactory()====定制的線程工廠
* rejectedExecutionHandler 當送出任務數超過maxmumPoolSize+workQueue之和時,
* 即當送出第41個任務時(前面線程都沒有執行完,此測試方法中用sleep(100)),
* 任務會交給RejectedExecutionHandler來處理
*/
public void init() {
pool = new ThreadPoolExecutor(
1,
3,
30,
TimeUnit.MINUTES,
new ArrayBlockingQueue<Runnable>(5),
new CustomThreadFactory(),
new CustomRejectedExecutionHandler());
}
public void destory() {
if(pool != null) {
pool.shutdownNow();
}
public ExecutorService getCustomThreadPoolExecutor() {
return this.pool;
private class CustomThreadFactory implements ThreadFactory {
private AtomicInteger count = new AtomicInteger(0);
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r);
String threadName = CustomThreadPoolExecutor.class.getSimpleName() + count.addAndGet(1);
System.out.println(threadName);
t.setName(threadName);
return t;
private class CustomRejectedExecutionHandler implements RejectedExecutionHandler {
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
try {
// 核心改造點，由blockingqueue的offer改成put阻塞方法
executor.getQueue().put(r);
} catch (InterruptedException e) {
e.printStackTrace();
}
// 測試構造的線程池
public static void main(String[] args) {
CustomThreadPoolExecutor exec = new CustomThreadPoolExecutor();
// 1.初始化
exec.init();
ExecutorService pool = exec.getCustomThreadPoolExecutor();
for(int i=1; i<100; i++) {
System.out.println("送出第" + i + "個任務!");
pool.execute(new Runnable() {
@Override
public void run() {
try {
System.out.println(">>>task is running=====");
TimeUnit.SECONDS.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
// 2.銷毀----此處不能銷毀,因為任務沒有送出執行完,如果銷毀線程池,任務也就無法執行了
// exec.destory();
try {
Thread.sleep(10000);
} catch (InterruptedException e) {
e.printStackTrace();
}
解釋：當送出任務被拒絕時，進入拒絕機制，我們實作拒絕方法，把任務重新用阻塞送出方法put送出，實作阻塞送出任務功能，防止隊列過大，OOM，送出被拒絕方法在下面
// 進入拒絕機制， 我們把runnable任務拿出來，重新用阻塞操作put，來實作送出阻塞功能
總結：
1、用ThreadPoolExecutor自定義線程池，看線程是的用途，如果任務量不大，可以用無界隊列，如果任務量非常大，要用有界隊列，防止OOM
2、如果任務量很大，還要求每個任務都處理成功，要對送出的任務進行阻塞送出，重寫拒絕機制，改為阻塞送出。保證不抛棄一個任務
3、最大線程數一般設為2N+1最好，N是CPU核數
4、核心線程數，看應用，如果是任務，一天跑一次，設定為0，合适，因為跑完就停掉了，如果是常用線程池，看任務量，是保留一個核心還是幾個核心線程數
5、如果要擷取任務執行結果，用CompletionService，但是注意，擷取任務的結果的要重新開一個線程擷取，如果在主線程擷取，就要等任務都送出後才擷取，就會阻塞大量任務結果，隊列過大OOM，是以最好異步開個線程擷取結果
<b>本文轉自農夫山泉别墅部落格園部落格，原文連結：</b><b>http://www.cnblogs.com/yaowen/p/6071537.html</b><b>，如需轉載請自行聯系原作者</b>