六、執行緒池（一）
執行緒池
通過建立池可以有效的利用系統資源，節約系統性能。Java 中的執行緒池就是一種非常好的實現，從 JDK1.5 開始 Java 提供了一個執行緒工廠 Executors 用來生成執行緒池，通過 Executors 可以方便的生成不同類型的執行緒池。
執行緒池的優點
- 降低資源消耗。執行緒的開啟和銷毀會消耗資源，通過重複利用已創建的執行緒降低執行緒創建和銷毀造成的消耗。
- 提高響應速度。當任務到達時，任務可以不需要的等到執行緒創建就能立即執行。
- 提高執行緒的可管理性。執行緒是稀缺資源，如果無限制的創建，不僅會消耗系統資源，還會降低系統的穩定性，使用執行緒池可以進行統一的分配，調優和監控。
常見的執行緒池
- CachedThreadPool：可快取的執行緒池，該執行緒池中沒有核心執行緒，非核心執行緒的數量為 Integer.max_value，就是無限大，當有需要時創建執行緒來執行任務，沒有需要時回收執行緒，適用於耗時少，任務量大的情況。
- SecudleThreadPool：周期性執行任務的執行緒池，按照某種特定的計劃執行執行緒中的任務，有核心執行緒，但也有非核心執行緒，非核心執行緒的大小也為無限大。適用於執行周期性的任務。
- SingleThreadPool：只有一條執行緒來執行任務，適用於有順序的任務的應用場景。
- FixedThreadPool：定長的執行緒池，有核心執行緒，核心執行緒的即為最大的執行緒數量，沒有非核心執行緒
- Executors.newFixedThreadPool()、Executors.newSingleThreadExecutor() 和 Executors.newCachedThreadPool() 等方法的底層都是通過 ThreadPoolExecutor 實現的。
ThreadPoolExecutor
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
if (corePoolSize < 0 ||
// maximumPoolSize 必須大於 0，且必須大於 corePoolSize
maximumPoolSize <= 0 ||
maximumPoolSize < corePoolSize ||
keepAliveTime < 0)
throw new IllegalArgumentException();
if (workQueue == null || threadFactory == null || handler == null)
throw new NullPointerException();
this.acc = System.getSecurityManager() == null ?
null :
AccessController.getContext();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
參數介紹：
-
corePoolSize
- 執行緒池的核心執行緒數。在沒有設置 allowCoreThreadTimeOut 為 true 的情況下，核心執行緒會在執行緒池中一直存活，即使處於閑置狀態。
- 如果設置為 0，則表示在沒有任何任務時，銷毀執行緒池；如果大於 0，即使沒有任務時也會保證執行緒池的執行緒數量等於此值。
- 但需要注意，此值如果設置的比較小，則會頻繁的創建和銷毀執行緒，如果設置的比較大，則會浪費系統資源，所以需要根據自己的實際業務來調整此值。
-
maximumPoolSize
- 執行緒池所能容納的最大執行緒數。當活動執行緒(核心執行緒+非核心執行緒)達到這個數值後，後續任務將會根據 RejectedExecutionHandler 來進行拒絕策略處理。
- 官方規定此值必須大於 0，也必須大於等於 corePoolSize，此值只有在任務比較多，且不能存放在任務隊列時，才會用到。
-
keepAliveTime
- 非核心執行緒閑置時的超時時長。超過該時長，非核心執行緒就會被回收。
- 若執行緒池通過 allowCoreThreadTimeOut() 方法設置 allowCoreThreadTimeOut 屬性為 true，則該時長同樣會作用於核心執行緒，AsyncTask 配置的執行緒池就是這樣設置的。
-
unit
- keepAliveTime 時長對應的單位。
-
workQueue
- 表示執行緒池執行的任務隊列，當執行緒池的所有執行緒都在處理任務時，如果來了新任務就會快取到此任務隊列中排隊等待執行。
- 是一個阻塞隊列 BlockingQueue，雖然它是 Queue 的子介面，但是它的主要作用並不是容器，而是作為執行緒同步的工具，他有一個特徵，當生產者試圖向 BlockingQueue 放入(put)元素，如果隊列已滿，則該執行緒被阻塞；當消費者試圖從 BlockingQueue 取出(take)元素，如果隊列已空，則該執行緒被阻塞。
-
ThreadFactory
- 執行緒的創建工廠，功能很簡單，就是為執行緒池提供創建新執行緒的功能。
- 也可以自定義一個執行緒工廠，通過實現 ThreadFactory 介面來完成，這樣就可以自定義執行緒的名稱或執行緒執行的優先順序了。
- 通常在創建執行緒池時不指定此參數，它會使用默認的執行緒創建工廠的方法來創建執行緒，源程式碼如下：
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) { // Executors.defaultThreadFactory() 為默認的執行緒創建工廠 this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), defaultHandler); } public static ThreadFactory defaultThreadFactory() { return new DefaultThreadFactory(); } // 默認的執行緒創建工廠，需要實現 ThreadFactory 介面 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; } } -
RejectedExecutionHandler
- 表示指定執行緒池的拒絕策略，當執行緒池的任務已經在快取隊列 workQueue 中存儲滿了之後，並且不能創建新的執行緒來執行此任務時，就會用到此拒絕策略.
- 它屬於一種限流保護的機制，這裡有四種任務拒絕類型：
- AbortPolicy: 不執行新任務，直接拋出異常，提示執行緒池已滿，涉及到該異常的任務也不會被執行，執行緒池默認的拒絕策略就是該策略。
- DisCardPolicy: 不執行新任務，也不拋出異常，即忽略此任務；
- DisCardOldSetPolicy: 將消息隊列中的第一個任務(即等待時間最久的任務)替換為當前新進來的任務執行，忽略最早的任務（最先加入隊列的任務）；
- CallerRunsPolicy: 把任務交給當前執行緒來執行；
/** * 執行緒池的拒絕策略 */ @Test public void test1() { // 創建執行緒池 核心執行緒為1，最大執行緒為3，任務隊列大小為2 ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(1, 3, 10, TimeUnit.SECONDS, new LinkedBlockingDeque<>(2), new ThreadPoolExecutor.AbortPolicy() // 添加 AbortPolicy 拒絕策略 ); for (int i = 0; i < 6; i++) { poolExecutor.execute(() -> { System.out.println(Thread.currentThread().getName()); }); } }- 自定義執行緒池拒絕策略
/** * 自定義執行緒池的拒絕策略 * 實現介面 RejectedExecutionHandler */ @Test public void test2() { ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 3, 10, TimeUnit.SECONDS, new LinkedBlockingDeque<>(2), new RejectedExecutionHandler() { @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { // 業務處理方法 System.out.println("執行自定義拒絕策略"); } } ); for (int i = 0; i < 6; i++) { executor.execute(() -> { System.out.println(Thread.currentThread().getName()); }); } }
執行緒池工作原理
執行緒池的工作流程要從它的執行方法 execute() 說起，源碼如下：
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
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);
// 如果執行緒池的執行緒數為 0 時（當 corePoolSize 設置為 0 時會發生）
else if (workerCountOf(recheck) == 0)
addWorker(null, false); // 新建執行緒執行任務
}
// 核心執行緒都在忙且隊列都已爆滿，嘗試新啟動一個執行緒執行失敗
else if (!addWorker(command, false))
// 執行拒絕策略
reject(command);
}
execute() VS submit()
- execute() 和 submit() 都是用來執行執行緒池任務的，它們最主要的區別是，submit() 方法可以接收執行緒池執行的返回值，而 execute() 不能接收返回值。
- sumbit 之所以可以接收返回值，是因為參數中可以傳遞：Callable
task，而通過 callable 創建的執行緒任務有返回值並且可以拋出異常。
/**
* execute VS sumbin
* execute 提交任務沒有返回值
* submit 提交任務有返回值
*/
@Test
public void test3() throws ExecutionException, InterruptedException {
ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 10, TimeUnit.SECONDS, new LinkedBlockingDeque<>(20));
// execute
executor.execute(new Runnable() {
@Override
public void run() {
System.out.println("Hello, execute");
}
});
// submit 使用
Future<String> future = executor.submit(new Callable<String>() {
@Override
public String call() throws Exception {
System.out.println("Hello, submit");
return "submit success";
}
});
System.out.println(future.get());
}
- 它們的另一個區別是 execute() 方法屬於 Executor 介面的方法，而 submit() 方法則是屬於 ExecutorService 介面的方法。
-
執行緒池的使用：
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
/**
* @author xiandongxie
*/
public class ThreadPool {
//參數初始化 返回Java虛擬機可用的處理器數量
// private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
private static final int CPU_COUNT = 2;
//核心執行緒數量大小
private static final int corePoolSize = Math.max(2, Math.min(CPU_COUNT - 1, 4));
//執行緒池最大容納執行緒數
private static final int maximumPoolSize = CPU_COUNT * 2 + 1;
//執行緒空閑後的存活時長
private static final int keepAliveTime = 30;
//任務過多後，存儲任務的一個阻塞隊列
BlockingQueue<Runnable> workQueue = new SynchronousQueue<>();
//執行緒的創建工廠
ThreadFactory threadFactory = new ThreadFactory() {
private final AtomicInteger mCount = new AtomicInteger(1);
public Thread newThread(Runnable r) {
return new Thread(r, "AdvacnedAsyncTask #" + mCount.getAndIncrement());
}
};
//執行緒池任務滿載後採取的任務拒絕策略: 不執行新任務，直接拋出異常，提示執行緒池已滿
RejectedExecutionHandler rejectHandler = new ThreadPoolExecutor.AbortPolicy();
//執行緒池對象，創建執行緒
ThreadPoolExecutor mExecute = new ThreadPoolExecutor(
corePoolSize,
maximumPoolSize,
keepAliveTime,
TimeUnit.SECONDS,
workQueue,
threadFactory,
rejectHandler
);
public static void main(String[] args) {
System.out.println("main start ..... \nCPU_COUNT = " + CPU_COUNT + "\tcorePoolSize=" + corePoolSize + "\tmaximumPoolSize=" + maximumPoolSize);
ThreadPool threadPool = new ThreadPool();
ThreadPoolExecutor execute = threadPool.mExecute;
// 預啟動所有核心執行緒
execute.prestartAllCoreThreads();
for (int i = 0; i < 5; i++) {
execute.execute(new Runnable() {
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + "\tstart..." + System.currentTimeMillis());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + "\tend..." + System.currentTimeMillis());
}
});
}
execute.shutdown();
System.out.println("main end .....");
}
}
