java多執行緒
- 執行緒和行程
- 執行緒的生命周期
- 新建New
- 就緒&運行 Runable&Runnging
- 阻塞Blocked
- 等待 waiting
- 計時等待Time waiting
- 銷毀Terminated
- 執行緒池概念和多執行緒使用場景
- 執行緒池的引數決議
- 執行緒池阻塞佇列BlockingQueue
- 執行緒池工廠ThreadFactory
- 執行緒池拒絕策略RejectedExecutionHandler
- JDK Executors執行緒池的幾種實作方式
- 創建型別簡單決議
- newCachedThreadPool
- newFixedThreadPool（附加原始碼決議）
- 執行緒池的引數設定方案
- 動態設定引數
- 總結&反思
- Spring boot使用執行緒池
- 參考資料
執行緒和行程
行程是系統進行資源調度和分配的基本單元，是作業系統的基礎，執行緒是系統調度的最小單元，是行程的運算單元，一個行程可能包含一個或者多個執行緒，
執行緒的生命周期
執行緒的生命周期分別六個：新建、就緒&運行、阻塞、等待、計時等待、銷毀
新建New
執行緒的創建有幾種方式：Thread類創建、實作Runnable介面、Callable和Future創建
# 1、thread
new Thread() {
@Override
public void run() {
}
}.start();
# runnable
public class RunnableThread implements Runnable {
@Override
public void run() {
System.out.println("runnable thread");
}
public static void main(String[] args){
Thread t = new Thread(new RunnableThread());
t.start();
}
}
# Callable&Future
public class CallableThread implements Callable<Integer> {
@Override
public Integer call() throws Exception {
System.out.println("Callable Thread return value");
return 0;
}
public static void main(String[] args) throws ExecutionException, InterruptedException {
FutureTask<Integer> future = new FutureTask<Integer>(new CallableThread());
new Thread(future).start();
System.out.println(future.get());
}
}
其實，比較細心看的話，最終都是Runnable一種方式實作，下面我們一起來解讀Thread的部分原始碼：
1、為啥執行緒有如上介紹的六種狀態呢，這是執行緒 Thread物件定義的java.lang.Thread.State列舉屬性
每種狀態的意義和實作方式英文中都描述很清楚，其實個人以前初學執行緒時，還有一些疑問，為啥要用執行緒以及執行執行緒的start和run方法區別在哪，接下來個人感官解讀一下原始碼流程：
執行緒初始化方法：
/**
* Initializes a Thread.
*
* @param g 執行緒組，是維護執行緒樹的物件，所有執行緒必須具備的屬性要素，這里可以判斷執行緒是否具有相應的權限，以及是否合法，執行緒狀態，是否守護執行緒等；目標是維護一組執行緒和執行緒組，同時我們要注意的執行緒之前的通訊是局限于執行緒組，是一組執行緒中維護的執行緒**
* @param target 運行執行緒的物件，執行緒執行時拿到的run或者call方法的目標物件
* @param name 當前執行緒名稱
* @param stackSize 新建執行緒時堆疊大小，當為0時可忽略
*
* @param acc 背景關系權限控制
*/
private void init(ThreadGroup g, Runnable target, String name,
long stackSize, AccessControlContext acc) {
// ………… 省略部分代碼
/*獲取安全管理策略，主要用來檢查權限相關因素，若權限不滿足時，拋出例外SecurityException，啟動時是通過jvm引數設定[java.security.manager]，具體可查看 [java API](https://docs.oracle.com/en/java/javase/15/docs/api/java.base/java/lang/SecurityManager.html)*/
SecurityManager security = System.getSecurityManager();
if (g == null) { // 當java.security.manager不設定時，這里為空
// 若需要安全管理策略，直接取得執行緒組
if (security != null) {
g = security.getThreadGroup();
}
// 不存在父級樹尋找
if (g == null) {
g = parent.getThreadGroup();
}
}
// 檢查權限
g.checkAccess();
/*
* 檢測是否能被實力構造和重寫
*/
if (security != null) {
if (isCCLOverridden(getClass())) {
security.checkPermission(SUBCLASS_IMPLEMENTATION_PERMISSION);
}
}
// 以便垃圾回收，增加未啟動執行緒數
g.addUnstarted();
// 設定是否守護執行緒，執行緒優先級，安全控制，執行目標，堆疊長度以及執行緒id等
………… 省略部分代碼
}
接下來講解一下直接執行run和start方法區別，執行run是居于現有jvm的當前執行緒執行方法體，執行start是居于jvm所在的行程分配資源，創建堆疊幀空間出來創建新的執行單元，分配問堆疊幀空間之類，在當前堆疊幀空間呼叫Thread的run方法，進而run呼叫傳入的target的run方法(有興趣的可以解讀open jdk的start0方法)，
# Thread#run
@Override
public void run() {
if (target != null) {
target.run(); // runnable
}
}
就緒&運行 Runable&Runnging
當我們新建執行緒完后執行start才進入就緒狀態Runnable，執行緒內部呼叫了run方法時進入運行階段Running,但是直接執行run方法不是啟動執行緒，具體如下驗證，
public class RunStartThread extends Thread {
public RunStartThread(String name) {
super(name);
}
@Override
public void run() {
System.out.println(System.currentTimeMillis());
try {
sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public static void main(String[] args) {
RunStartThread rst = new RunStartThread("runThread");
rst.run(); // 主執行緒運行run方法
rst.start(); // 啟動子執行緒運行run
}
}
運行如上代碼的run方法抓取執行緒dump如下圖，我們觀察執行緒名稱是"main"
當運行start方法時，我們再抓取dump時，發現目前正在運行的執行緒名稱是我自定義的執行緒名稱runThread,此外我們發現start并不是直接呼叫runnable的run，而是呼叫本地方堆疊中的start0,讓jvm來處理執行緒調度，
阻塞Blocked
執行緒進入blocked狀態時一般是自動等待后進入運行狀態或者直接死亡結束，一般導致blocked的是synchronized，如下代碼示例，因此我們開發時主要盡量不要使用synchronized，原因有是鎖匙自動釋放不可控，此外是單執行緒運行，同一物件不能同時運行，如果真需要控制執行緒安全性的編程，盡量用Lock：
public class BlockThreads {
public static void main(String[] args) throws InterruptedException {
TestThread th = new TestThread();
th.runThread(th,"Thread1");
th.runThread(th,"Thread2");
th.runThread(th,"Thread3");
System.out.println("111");
}
private static class TestThread {
public synchronized void sayHello() throws InterruptedException {
System.out.println(System.currentTimeMillis());
Thread.sleep(3000);
}
public void runThread(TestThread th, String threadName) {
new Thread(threadName) {
@Override
public void run() {
try {
th.sayHello();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}.start();
}
}
}
等待 waiting
導致執行緒處于等待的方法有Object#wait(Object#notify或者Object#notifyAll恢復)、Thread#join以及LockSupport#park（LockSupport#unpark）,這里導致處于等待狀態時CPU是釋放出資源的，
計時等待Time waiting
Object#wait(time)、LockSupport#parkNanos、以及LockSupport#parkUntil等待時間，其實Time waiting概念我們使用到很多場景，比如nginx調優，執行緒銷毀時間的調優，請求并發時超時時間的設定，有效的設定超時時間有利于提高系統的吞吐量
銷毀Terminated
執行緒銷毀有自動銷毀和手動銷毀，自動銷毀即執行緒自己執行完run方法后，JVM就會銷毀執行緒，手動銷毀可以使用Thread#stop方法銷毀，但是此方法已經廢棄，因為是暴力手段，可能內部JVM監控資訊也無法監控，Thread#interrrupted方法進行銷毀判斷，若不能銷毀，則會發生InterruptedException例外
執行緒池概念和多執行緒使用場景
執行緒是執行單元，執行緒池即一組執行單元組成的集體，也就是執行緒的一種使用方式，執行緒池用來維護執行緒的啟動、服用、調度機制，在多核CPU和多任務調度時，我們可以用執行緒池來處理多執行緒，提高CPU的使用同時控制CPU高壓，提高性能和避免阻塞，比如短信發送，http請求，定時任務，異步呼叫之類，
執行緒池的引數決議
JDK自帶的創建執行緒物件是ThreadPoolExecutor，物件中有幾個引數分別為：核心執行緒數、最大數執行緒數、執行緒存活時間、執行緒工廠、執行緒拒絕策略，如下原始碼
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler)
具體幾個含義有簡單文字簡介：核心執行緒數corePoolSize是運行的執行緒數，當同一時間執行緒數量大于核心執行緒數時，執行緒進入等待佇列workQueue，超出等待佇列時將會有新建非核心的執行緒（maximumPoolSize-corePoolSize）執行，當執行緒數量大于maximumPoolSize+workQueue#size時，將會發生拒絕策略，具體拒接策略后面再討論，參考美團技術團隊的圖美團技術團隊
執行緒池阻塞佇列BlockingQueue
JDK自帶的常見的等待佇列有LinkedBlockingQueue、ArrayBlockingQueue、SynchronousQueue（而外提及一點佇列長度是可以動態改變的哦，比如LinkedBlockingQueue的capacity設定為volatile）
- LinkedBlockingQueue是鏈表節點存盤，FIFO模型，鏈表型別存盤當然是無界限，一般設定這個時，最大執行緒數基本無效，因為永遠不會超出長度，除非發生OOM例外,在大量等待執行緒的高并發條件下建議采用此阻塞佇列，
- ArrayBlockingQueue 指定長度等待佇列，此有點是比較精確的設定資料佇列來實作等待佇列，
- SynchronousQueue 無快取等待佇列，佇列始終為0，此操作一般是無界操作，充分利用CPU的使用率，比如Executors#newCachedThreadPool就是用次方法實作，
執行緒池工廠ThreadFactory
常用開源專案執行緒池工廠有CustomizableThreadFactory、ThreadFactoryBuilder、BasicThreadFactory,工廠方法主要設定執行緒的優先級和執行緒名之類的執行緒屬性，這里就不細講解了，主要如下簡單實用例子：
ublic class ThreadFactoryTest implements ThreadFactory {
private final AtomicInteger threadCount = new AtomicInteger(0);
public static void main(String[] args) {
ExecutorService executor = new ThreadPoolExecutor(2, 2, 60L, TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>(), new ThreadFactoryTest());
executor.submit(() -> {
System.out.println(String.format("thread-Name = %s,Thread priority = %d",
Thread.currentThread().getName(), Thread.currentThread().getPriority()));
});
executor.submit(() -> {
System.out.println(String.format("thread-Name = %s,Thread priority = %d",
Thread.currentThread().getName(), Thread.currentThread().getPriority()));
});
}
/**
* @param r 傳入的執行緒
**/
@Override
public Thread newThread(Runnable r) {
Thread th = new Thread(r);
th.setPriority(2);
th.setName("設定執行緒名前綴" + this.threadCount.incrementAndGet());
return th;
}
}
執行緒池拒絕策略RejectedExecutionHandler
JDK常見的拒絕策略AbortPolicy中止策略、CallerRunsPolicy超出時執行run、DiscardPolicy超出時丟棄、DiscardOldestPolicy丟棄佇列最后一個，
JDK Executors執行緒池的幾種實作方式
Executors是工具類（一般s后最都是工具類，比如Arrays,Systems,Collections，Objects等），廢話多說，Executors創建執行緒的方法有newFixedThreadPool,newCachedThreadPool（ThreadPoolExecutor主要原始碼決議）,newScheduledThreadPool,newSingleThreadExecutor,newSingleThreadScheduledExecutor，
創建型別簡單決議
newCachedThreadPool
newCachedThreadPool提供兩個構造方法實作，一個無參構造Executors#newCachedThreadPool();，一個有參構造 Executors#newCachedThreadPool(ThreadFactory threadFactory)
優點：核心執行緒數設定為0，意思CPU有空閑時，有執行緒進入則馬上進入執行狀態，等待佇列為無界同步等待佇列SynchronousQueue，即多CPU的情況下充分利用，其實綜合上面兩個因素可知最大執行緒數設定相當于無效，因此充分利用多核CPU的特性，
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
缺點：高并發時導致CPU占用100%,其他執行緒任務無法處理，還有長期CPU高壓會發熱導致高溫，我們系統建議CPU使用不要超過80%，常態不要超過60%，下面來介紹其中兩個執行方法以及附加原始碼決議
newFixedThreadPool（附加原始碼決議）
Executors#newFixedThreadPool固定大小的執行緒池，即設定核心執行緒數和最大執行緒數一樣，等待佇列是無界線性等待佇列，這里有個優點是充分利用執行緒的復用，至于為啥呢，當然是解讀原始碼見證：
AbstractExecutorService#submit方法創建任務FutureTask并呼叫ThreadPoolExecutor#execute：
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null); // 新建future任務
execute(ftask);
return ftask;
}
接下來重點決議ThreadPoolExecutor#executor里的執行方法,分三步走(這里提到一個點，執行緒數量和狀態時通過AtomicInteger的32位的ctl，高三位是狀態保存，低29是執行緒池的最大數量)：
int c = ctl.get();/* 獲取主執行緒狀態控制29位變數 */
if (workerCountOf(c) < corePoolSize) { /* 前29位作為統計執行緒數，判斷worker是否大于核心執行緒數 */
if (addWorker(command, true)) /* 添加worker作業執行緒，若新建成功則執行執行緒，并回傳true */
return;
c = ctl.get(); /* 再次檢測當前執行緒地位29 */
}
if (isRunning(c) && workQueue.offer(command)) { // worker無法獲取和創建，插入等待佇列
int recheck = ctl.get(); // 再次檢測執行緒池worker大小
if (! isRunning(recheck) && remove(command)) // 若執行緒池不可運行狀態，且移除當前執行緒成功，則拒絕策略
reject(command);
else if (workerCountOf(recheck) == 0) // 若當前沒有執行緒worker，即核心執行緒為0，則立即執行佇列
addWorker(null, false);
}
else if (!addWorker(command, false)) // 佇列已經滿了，則直接添加非核心執行緒并運行
reject(command); // 運行或者創建非核心執行緒失敗，則拒絕策略
通過上面分析可知，執行緒池的狀態保存是高三位，低29位保存運行執行緒數量，
接下來重點講worker和task
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
// ……………… 省略部分代碼
for (;;) {
// ……………… 省略部分代碼
// 這里判斷最大worker數，查看是核心執行緒還是最大執行緒，若超出范圍直接回傳創建worker失敗
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
if (compareAndIncrementWorkerCount(c)) // 創建worker前檢測后并增加運行執行緒數
break retry;
}
}
// …………
Worker w = null;
try {
// 新建worker，同時呼叫ThreadFactory的newThread進而執行緒池的引數，比如引數名稱等
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock; // 獲取執行緒池鎖
mainLock.lock();
try {
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(); // 創建worker成功后直接呼叫執行緒的start方法執行執行緒，并且回傳成功，呼叫worker啟動后將會執行run方法，
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
經過代碼分析最終都是worker來調度任務，當執行緒數大于核心執行緒數時，worker傳入的執行緒為空，此時需要執行的任務放在佇列的workQueue中（即new Worker->start worker -> runWork -> getTask - > runTask），真理就在我們如下runWorker方法，這里有一個“提交優先級概念”，先執行核心執行緒task != null再執行getTask()，意思佇列溢位的執行緒其實是優先執行，因為如上代碼execute講解溢位佇列直接直接addworker優先于佇列的添加，
/*
worker啟動時，委托給主執行緒的runWorker
*/
public void run() {
runWorker(this);
}
final void runWorker(Worker w) {
// ………省略部分代碼
// 獲取當前任務，當時非核心時且添加進佇列時為null，需要從佇列中獲取
Runnable task = w.firstTask;
// …………
/* 當任務為空，且佇列也不為空是，不執行 */
while (task != null || (task = getTask()) != null) {
try {
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
afterExecute(task, thrown);
}
} finally {
task = null; //這個地方非常關鍵，執行完后佇列中查找，
w.completedTasks++;
w.unlock();
}
}
執行緒池的引數設定方案
我們一般考慮的兩個因素是CPU還有IO，一般引數設定都是針對CPU密集型和IO密集型，但是還要結合實際業務場景分析，比如業務耗時，tps之類的因素合理分配核心執行緒數和最大執行緒數，最好動態設定引數（JDK支持動態調整核心執行緒數和最大執行緒數以及等待佇列的長度，再結合apollo配置中心動態更新再好不過了，如美團技術團隊）
CPU密集型，核心執行緒數設定為CPU有效個數+1，最大執行緒數設定為因為2×CPU有效個數+1，可能有些假死，
IO密集型即核心執行緒數為2×CPU有效個數，最大執行緒數為25×CPU有效個數，具體如下
動態設定引數
動態設定執行緒池的大小有利于處理高峰問題以及調優執行緒池資料，采取方法是ThreadPoolExecutor#setCorePoolSize動態設定核心執行緒數，interruptIdleWorkers可以清空空閑的worker以便占用資源，如下代碼所示：
public void setCorePoolSize(int corePoolSize) {
if (corePoolSize < 0)
throw new IllegalArgumentException();
int delta = corePoolSize - this.corePoolSize; // 設定的核心執行緒數和原來的差值
this.corePoolSize = corePoolSize;
if (workerCountOf(ctl.get()) > corePoolSize) // 作業worker是否大于設定的核心執行緒數，如果大于則當worker空余時清空，
interruptIdleWorkers(); // 這方法其實很重要，我們可以用來設定回收沒有使用的核心執行緒數，
else if (delta > 0) { // 若設定執行緒數大于原有執行緒數，則看佇列是否有等待執行緒，如果有則直接回圈創建worker并執行task任務，知道worker大于最大執行緒數或者佇列已空
// We don't really know how many new threads are "needed".
// As a heuristic, prestart enough new workers (up to new
// core size) to handle the current number of tasks in
// queue, but stop if queue becomes empty while doing so.
int k = Math.min(delta, workQueue.size());
while (k-- > 0 && addWorker(null, true)) {
if (workQueue.isEmpty())
break;
}
}
}
總結&反思
1、addWorker(Runnable firstTask, boolean core) 新建worker,此方法的引數firstTask為空時，居于預熱的功能，類似于spring的懶加載
2、setCorePoolSize動態設定核心執行緒數
3、setMaximumPoolSize 動態設定最大執行緒數
4、CPU密集型和IO密集型
5、runWorker worker執行任務
6、interruptIdleWorkers銷毀空閑核心執行緒
7、Executors創建執行緒方式newFixedThreadPool、newCachedThreadPool、newScheduledThreadPool、newSingleThreadExecutor
8、execute和submit執行優先級和提交優先級,以及兩者區別是有回傳值之類，
9、創建執行緒的幾種方式Thread、runnable、Callable和future，以及執行緒都有哪些狀態，對于JVM的堆疊除錯的使用
10、常見阻塞執行緒LinkedBlockingQueue、ArrayBlockingQueue、SynchronousQueue
11、常見的執行緒工廠CustomizableThreadFactory、ThreadFactoryBuilder、BasicThreadFactory以及目的
12、RejectedExceptionHandler常見幾個實作類AbortPolicy、CallerRunsPolicy、DiscardPolicy、DiscardOldestPolicy
13、如何動態設定核心執行緒數和最大執行緒數以及阻塞佇列
14、ThreadGroup、securityManager有什么意思
15、后續追加………………
Spring boot使用執行緒池
這里制作簡單實用，詳細使用還得結合實際場景：
@SpringBootTest
@EnableAsync
class PoolApplicationTests {
@Autowired
private PoolService poolService;
@Test
void contextLoads() {
poolService.say1();
poolService.say2();
}
}
@Service
public class PoolService {
@Value("${spring.pool.core.size:5}")
private int coreSize;
@Value("${spring.pool.max.size:10}")
private int maxNumSize;
/**
* 自定義執行緒池
*
* @return
*/
@Bean("executor")
public Executor executor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(coreSize);
executor.setMaxPoolSize(maxNumSize);
executor.setQueueCapacity(20);
executor.initialize();
return executor;
}
/**
* 在目標執行緒池執行任務
*/
@Async(value = "executor")
public void say1() {
System.out.println(Thread.currentThread().getName());
}
@Async(value = "executor")
public void say2() {
System.out.println(Thread.currentThread().getName());
}
}
參考資料
【1】JDK 1.8 原始碼
【2】美團技術團隊(部分圖片參考和知識點)
【3】執行緒和行程百度百科
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/230282.html
標籤：java
上一篇：Java中關于i++的一些問題