1. 線程池的創建
線程池的創建使用ThreadPoolExecutor類，有利於編碼時更好的明確線程池運行規則。
//構造函數
/**
* Creates a new {@code ThreadPoolExecutor} with the given initial
* parameters.
*
* @param corePoolSize the number of threads to keep in the pool, even
* if they are idle, unless {@code allowCoreThreadTimeOut} is set
* @param maximumPoolSize the maximum number of threads to allow in the
* pool
* @param keepAliveTime when the number of threads is greater than
* the core, this is the maximum time that excess idle threads
* will wait for new tasks before terminating.
* @param unit the time unit for the {@code keepAliveTime} argument
* @param workQueue the queue to use for holding tasks before they are
* executed. This queue will hold only the {@code Runnable}
* tasks submitted by the {@code execute} method.
* @param threadFactory the factory to use when the executor
* creates a new thread
* @param handler the handler to use when execution is blocked
* because the thread bounds and queue capacities are reached
* @throws IllegalArgumentException if one of the following holds:<br>
* {@code corePoolSize < 0}<br>
* {@code keepAliveTime < 0}<br>
* {@code maximumPoolSize <= 0}<br>
* {@code maximumPoolSize < corePoolSize}
* @throws NullPointerException if {@code workQueue}
* or {@code threadFactory} or {@code handler} is null
*/
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;
}
參數含義
(1) 核心線程數corePoolSize： 保持在池中的線程數
(2) 最大線程數maximumPoolSize
(3) 保活時間keepAliveTime： 線程數大於corePoolSize，閑置線程最大空閑時間
(4) 時間單位unit
(5) 阻塞隊列workQueue
java.util.concurrent.BlockingQueue主要實現類有：
- ArrayBlockingQueue： 數組結構有界阻塞隊列，FIFO排序。其構造函數必須設置隊列長度。
- LinkedBlockingQueue：鏈表結構有界阻塞隊列，FIFO排序。隊列默認最大長度為Integer.MAX_VALUE,故可能會堆積大量請求，導致OOM。
- PriorityBlockingQueue：支持優先級排序的無界阻塞隊列。默認自然順序排列，可以通過比較器comparator指定排序規則。
- DelayQueue：支持延時獲取元素的無界阻塞隊列。隊列使用PriorityQueue實現。
(6) 線程創建接口threadFactory
- 默認使用Executors.defaultThreadFactory()。
- 可以自定義ThreadFactory實現或使用第三方實現，方便指定有意義的線程名稱。
import com.google.common.util.concurrent.ThreadFactoryBuilder;
...
ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("my-pool-%d").build();
public class MyThreadFactory implements ThreadFactory {
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final String namePrefix;
MyThreadFactory(String namePrefix) {
this.namePrefix = namePrefix+"-";
}
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread( r,namePrefix + threadNumber.getAndIncrement());
if (t.isDaemon()) {
t.setDaemon(true);
}
if (t.getPriority() != Thread.NORM_PRIORITY) {
t.setPriority(Thread.NORM_PRIORITY);
}
return t;
}
}
(7) 飽和策略handler
- ThreadPoolExecutor.AbortPolicy()：終止策略(默認) ， 拋出java.util.concurrent.RejectedExecutionException異常。
- ThreadPoolExecutor.CallerRunsPolicy(): 重試添加當前的任務，他會自動重復調用execute()方法。
- ThreadPoolExecutor.DiscardOldestPolicy(): 拋棄下一個即將被執行的任務，然后嘗試重新提交新的任務。最好不和優先級隊列一起使用，因為它會拋棄優先級最高的任務。
- ThreadPoolExecutor.DiscardPolicy(): 拋棄策略， 拋棄當前任務。
2. 線程池的運行規則
execute添加任務到線程池:
一個任務通過execute(Runnable)方法被添加到線程池。任務是一個 Runnable類型的對象，任務的執行方法就是 Runnable類型對象的run()方法。
線程池運行規則:
當一個任務通過execute(Runnable)方法添加到線程池時：
-
如果此時線程池中的數量小於corePoolSize，即使線程池中的線程都處於空閑狀態，也要創建新的線程來處理被添加的任務。
-
如果此時線程池中的數量等於 corePoolSize，但是緩沖隊列 workQueue未滿，那么任務被放入緩沖隊列。
-
如果此時線程池中的數量大於corePoolSize，緩沖隊列workQueue滿，並且線程池中的數量小於maximumPoolSize，建新的線程來處理被添加的任務。
-
如果此時線程池中的數量大於corePoolSize，緩沖隊列workQueue滿，並且線程池中的數量等於maximumPoolSize，那么通過 handler所指定的策略來處理此任務。
-
也就是：處理任務的優先級為：
核心線程corePoolSize - > 任務隊列workQueue - > 最大線程maximumPoolSize
如果三者都滿了，使用handler策略處理該任務。
- 當線程池中的線程數量大於 corePoolSize時，如果某線程空閑時間超過keepAliveTime，線程將被終止。這樣，線程池可以動態的調整池中的線程數。
// execute方法源碼實現(jdk1.8)
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
/*
* 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);
}
3. 線程池的關閉
通過調用線程池的shutdown或shutdownNow方法來關閉線程池。
- shutdown：將線程池的狀態設置成SHUTDOWN狀態，然后interrupt空閑線程。
- shutdownNow：線程池的狀態設置成STOP，然后嘗試interrupt所有線程，包括正在運行的。
關於線程池狀態，源碼中的注釋比較清晰：
再看一下源代碼：
// 在關閉中，之前提交的任務會被執行（包含正在執行的，在阻塞隊列中的），但新任務會被拒絕。
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
// 狀態設置為shutdown
advanceRunState(SHUTDOWN);
// interrupt空閑線程
interruptIdleWorkers();
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
// 嘗試終止線程池
tryTerminate();
}
其中，interruptIdleWorkers()方法往下調用了interruptIdleWorkers(), 這里w.tryLock()比較關鍵。
中斷之前需要先tryLock()獲取worker鎖，正在運行的worker tryLock()失敗(runWorker()方法會先對worker上鎖)，故正在運行的worker不能中斷。
// 嘗試停止所有正在執行的任務，停止對等待任務的處理，並返回正在等待被執行的任務列表
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
// 狀態設置為STOP
advanceRunState(STOP);
// 停止所有線程 interruptWorkers邏輯簡單些，循環對所有worker調用interruptIfStarted().(interrupt所有線程)
interruptWorkers();
tasks = drainQueue();
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}
4. 線程池的使用場合
（1）單個任務處理的時間比較短；
（2）需要處理的任務數量大；
5. 線程池大小的設置
可根據計算任務類型估算線程池設置大小：
cpu密集型：可采用Runtime.avaliableProcesses()+1個線程；
IO密集型：由於阻塞操作多，可使用更多的線程，如2倍cpu核數。
6 實現舉例
場景: ftp服務器收到文件后，觸發相關搬移/處理操作。
public class FtpEventHandler extends DefaultFtplet {
@Override
public FtpletResult onUploadEnd(FtpSession session, FtpRequest request)
throws FtpException, IOException {
// 獲取文件名
String fileName = request.getArgument();
Integer index = fileName.lastIndexOf("/");
String realFileName = fileName.substring(index + 1);
index = realFileName.lastIndexOf("\\");
realFileName = realFileName.substring(index + 1);
// **處理文件**
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(10, 50, 10,
TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(3));
threadPool.execute(new fileSenderThread(realFileName));
return FtpletResult.DEFAULT;
}
}
Spring也提供了ThreadPoolTaskExecutor
<!--spring.xml配置示例-->
<bean id="gkTaskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
<property name="allowCoreThreadTimeOut" value="true"/>
<property name="corePoolSize" value="10"/>
<property name="maxPoolSize" value="50"/>
<property name="queueCapacity" value="3"/>
<property name="keepAliveSeconds" value="10"/>
<property name="rejectedExecutionHandler"
value="#{new java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy()}"/>
<property name="threadNamePrefix" value="gkTaskExecutor"/>
</bean>
//java代碼中注入bean
@Autowired
@Qualifier("gkTaskExecutor")
private ThreadPoolTaskExecutor gkTaskExecutor;
end.