線程池的基本使用
package com.shothook.test;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@Slf4j
public class ThreadPoolDemo {
public static void main(String[] args) {
ExecutorService threadPool = new ThreadPoolExecutor(10, 1000, 10, TimeUnit.SECONDS, new LinkedBlockingDeque<>(10));
LocalDateTime start = LocalDateTime.now();
for (int i = 0; i < 1000; i++) {
threadPool.submit(new Task(i));
}
LocalDateTime end = LocalDateTime.now();
Duration duration = Duration.between(start,end);
log.debug("submit task complete, takes time: {}", duration.getNano());
}
@AllArgsConstructor
@Slf4j
public static class Task implements Runnable {
private int i;
@Override
public void run() {
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
log.debug("thread name: {}, to execute task: {}", Thread.currentThread().getName(), i);
}
}
} 線程池的工作模式，簡單來說如下所示：
主線程往緩沖隊列中添加任務，線程池從隊列中取任務并執行，當然，真實情況比這複雜的多，這個後面再詳細分析，先有個概念就好。
建立線程池的各個參數的意義
線程池的構造函數：
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.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：核心線程數
maximumPoolSize：最大線程數，當往緩沖隊列中增加任務，緩沖隊列滿時，會在corePoolSize的基礎上繼續建立線程，知道觸發最大線程數
keepAliveTime：超過核心線程數的線程的存活時間，對超過存活時間的，線程池結束逾時的空閑線程
unit：超過核心線程數的線程的存活時間對應的時間機關
workQueue：線程池的任務緩存隊列
threadFactory：建立線程的工廠
handler：當到最大線程數，并且緩存隊列滿時，繼續添加任務時的處理政策
線程池的狀态
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
private static final int RUNNING = -1 << COUNT_BITS;
private static final int SHUTDOWN = 0 << COUNT_BITS;
private static final int STOP = 1 << COUNT_BITS;
private static final int TIDYING = 2 << COUNT_BITS;
private static final int TERMINATED = 3 << COUNT_BITS;
private static int runStateOf(int c) { return c & ~CAPACITY; }//擷取線程池的狀态
private static int workerCountOf(int c) { return c & CAPACITY; }//擷取線程池的線程數
private static int ctlOf(int rs, int wc) { return rs | wc; }//參數rs表示runState，參數wc表示workerCount，即根據runState和workerCount打包合并成ctl ctl : 高三位儲存線程池狀态，低29位儲存任務數。
線程池的建立
/**
* Executes the given task sometime in the future. The task
* may execute in a new thread or in an existing pooled thread.
* 在未來的某個時刻執行給定的任務。這個任務用一個新線程執行，或者用一個線程池中已經存在的線程執行
*
* If the task cannot be submitted for execution, either because this
* executor has been shutdown or because its capacity has been reached,
* the task is handled by the current {@code RejectedExecutionHandler}.
* 如果任務無法被送出執行，要麼是因為這個Executor已經被shutdown關閉，要麼是已經達到其容量上限，任務會被目前的RejectedExecutionHandler處理
*
* @param command the task to execute
* @throws RejectedExecutionException at discretion of
* {@code RejectedExecutionHandler}, if the task
* cannot be accepted for execution RejectedExecutionException是一個RuntimeException
* @throws NullPointerException if {@code command} is null
*/
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.
* 如果運作的線程少于corePoolSize，嘗試開啟一個新線程去運作command，command作為這個線程的第一個任務
*
* 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.
* 如果任務成功放入隊列，我們仍需要一個雙重校驗去确認是否應該建立一個線程（因為可能存在有些線程在我們上次檢查後死了） 或者 從我們進入這個方法後，pool被關閉了
* 是以我們需要再次檢查state，如果線程池停止了需要復原入隊列，如果池中沒有線程了，新開啟 一個線程
*
* 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.
* 如果無法将任務入隊列（可能隊列滿了），需要新開區一個線程（自己：往maxPoolSize發展）
* 如果失敗了，說明線程池shutdown 或者 飽和了，是以我們拒絕任務
*/
int c = ctl.get();
/**
* 1、如果目前線程數少于corePoolSize（可能是由于addWorker()操作已經包含對線程池狀态的判斷，如此處沒加，而入workQueue前加了）
*/
if (workerCountOf(c) < corePoolSize) {
//addWorker()成功，傳回
if (addWorker(command, true))
return;
/**
* 沒有成功addWorker()，再次擷取c（凡是需要再次用ctl做判斷時，都會再次調用ctl.get()）
* 失敗的原因可能是：
* 1、線程池已經shutdown，shutdown的線程池不再接收新任務
* 2、workerCountOf(c) < corePoolSize 判斷後，由于并發，别的線程先建立了worker線程，導緻workerCount>=corePoolSize
*/
c = ctl.get();
}
/**
* 2、如果線程池RUNNING狀态，且入隊列成功
*/
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();//再次校驗位
/**
* 再次校驗放入workerQueue中的任務是否能被執行
* 1、如果線程池不是運作狀态了，應該拒絕添加新任務，從workQueue中删除任務
* 2、如果線程池是運作狀态，或者從workQueue中删除任務失敗（剛好有一個線程執行完畢，并消耗了這個任務），確定還有線程執行任務（隻要有一個就夠了）
*/
//如果再次校驗過程中，線程池不是RUNNING狀态，并且remove(command)--workQueue.remove()成功，拒絕目前command
if (! isRunning(recheck) && remove(command))
reject(command);
//如果目前worker數量為0，通過addWorker(null, false)建立一個線程，其任務為null
//為什麼隻檢查運作的worker數量是不是0呢？？ 為什麼不和corePoolSize比較呢？？
//隻保證有一個worker線程可以從queue中擷取任務執行就行了？？
//因為隻要還有活動的worker線程，就可以消費workerQueue中的任務
else if (workerCountOf(recheck) == 0)
addWorker(null, false); //第一個參數為null，說明隻為建立一個worker線程，沒有指定firstTask
//第二個參數為true代表占用corePoolSize，false占用maxPoolSize
}
/**
* 3、如果線程池不是running狀态 或者 無法入隊列
* 嘗試開啟新線程，擴容至maxPoolSize，如果addWork(command, false)失敗了，拒絕目前command
*/
else if (!addWorker(command, false))
reject(command);
} 建立線程
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);//線程池狀态
// Check if queue empty only if necessary.
// 線程池為SHUTDOWN狀态，目前建立新線程的任務為空，并且線程池的隊列不為空
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c);
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
} 鈎子
protected void beforeExecute(Thread t, Runnable r) { }
protected void afterExecute(Runnable r, Throwable t) { }