執行緒池原理分析&鎖的深度化
執行緒池
什麼是執行緒池
Java中的線程池是運用場景最多的並發框架,幾乎所有需要非同步或並發執行任務的程式都可以使用線程池。在開發過程中,合理地使用線程池能夠帶來3個好處。第一:降低資源消耗。通過重複利用已創建的線程降低線程創建和銷毀造成的消耗。第二:提高響應速度。當任務到達時,任務可以不需要等到線程創建就能立即執行。第三:提高線程的可管理性。線程是稀缺資源,如果無限制地創建,不僅會消耗系統資源,還會降低系統的穩定性,使用線程池可以進行統一分配、調優和監控。但是,要做到合理利用線程池,必須對其實現原理了如指掌。
執行緒池作用
執行緒池是為突然大量爆發的執行緒設計的,通過有限的幾個固定執行緒為大量的操作服務,減少了建立和銷燬執行緒所需的時間,從而提高效率。
如果一個執行緒的時間非常長,就沒必要用執行緒池了(不是不能作長時間操作,而是不宜。),況且我們還不能控制執行緒池中執行緒的開始、掛起、和中止。
執行緒池的分類
ThreadPoolExecutor
Java是天生就支援併發的語言,支援併發意味著多執行緒,執行緒的頻繁建立在高併發及大資料量是非常消耗資源的,因為java提供了執行緒池。在jdk1.5以前的版本中,執行緒池的使用是及其簡陋的,但是在JDK1.5後,有了很大的改善。JDK1.5之後加入了java.util.concurrent包,java.util.concurrent包的加入給予開發人員開發併發程式以及解決併發問題很大的幫助。這篇文章主要介紹下併發包下的Executor介面,Executor介面雖然作為一個非常舊的介面(JDK1.5 2004年釋出),但是很多程式設計師對於其中的一些原理還是不熟悉,因此寫這篇文章來介紹下Executor介面,同時鞏固下自己的知識。如果文章中有出現錯誤,歡迎大家指出。
Executor框架的最頂層實現是ThreadPoolExecutor類,Executors工廠類中提供的newScheduledThreadPool、newFixedThreadPool、newCachedThreadPool方法其實也只是ThreadPoolExecutor的建構函式引數不同而已。通過傳入不同的引數,就可以構造出適用於不同應用場景下的執行緒池,那麼它的底層原理是怎樣實現的呢,這篇就來介紹下ThreadPoolExecutor執行緒池的執行過程。
corePoolSize: 核心池的大小。 當有任務來之後,就會建立一個執行緒去執行任務,當執行緒池中的執行緒數目達到corePoolSize後,就會把到達的任務放到快取隊列當中
maximumPoolSize: 執行緒池最大執行緒數,它表示線上程池中最多能建立多少個執行緒;keepAliveTime: 表示執行緒沒有任務執行時最多保持多久時間會終止。unit: 引數keepAliveTime的時間單位,有7種取值,在TimeUnit類中有7種靜態屬性:
執行緒池四種建立方式
Java通過Executors(jdk1.5併發包)提供四種執行緒池,分別為:newCachedThreadPool建立一個可快取執行緒池,如果執行緒池長度超過處理需要,可靈活回收空閒執行緒,若無可回收,則新建執行緒。
案例演示:
newFixedThreadPool 建立一個定長執行緒池,可控制執行緒最大併發數,超出的執行緒會在佇列中等待。newScheduledThreadPool 建立一個定長執行緒池,支援定時及週期性任務執行。newSingleThreadExecutor 建立一個單執行緒化的執行緒池,它只會用唯一的工作執行緒來執行任務,保證所有任務按照指定順序(FIFO, LIFO, 優先順序)執行。
newCachedThreadPool
建立一個可快取執行緒池,如果執行緒池長度超過處理需要,可靈活回收空閒執行緒,若無可回收,則新建執行緒。示例程式碼如下:
// 無限大小執行緒池 jvm自動回收
ExecutorService newCachedThreadPool = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
final int temp = i;
newCachedThreadPool.execute(new Runnable() {
@Override
public void run() {
try {
Thread.sleep(100);
} catch (Exception e) {
// TODO: handle exception
}
System.out.println(Thread.currentThread().getName() + ",i:" + temp);
}
});
}
總結: 執行緒池為無限大,當執行第二個任務時第一個任務已經完成,會複用執行第一個任務的執行緒,而不用每次新建執行緒。
newFixedThreadPool
建立一個定長執行緒池,可控制執行緒最大併發數,超出的執行緒會在佇列中等待。示例程式碼如下:
ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(5);
for (int i = 0; i < 10; i++) {
final int temp = i;
newFixedThreadPool.execute(new Runnable() {
@Override
public void run() {
System.out.println(Thread.currentThread().getId() + ",i:" + temp);
}
});
}
總結:因為執行緒池大小為3,每個任務輸出index後sleep 2秒,所以每兩秒列印3個數字。
定長執行緒池的大小最好根據系統資源進行設定。如Runtime.getRuntime().availableProcessors()
newScheduledThreadPool
建立一個定長執行緒池,支援定時及週期性任務執行。延遲執行示例程式碼如下:
ScheduledExecutorService newScheduledThreadPool = Executors.newScheduledThreadPool(5);
for (int i = 0; i < 10; i++) {
final int temp = i;
newScheduledThreadPool.schedule(new Runnable() {
public void run() {
System.out.println("i:" + temp);
}
}, 3, TimeUnit.SECONDS);
}
表示延遲3秒執行。
newSingleThreadExecutor
建立一個單執行緒化的執行緒池,它只會用唯一的工作執行緒來執行任務,保證所有任務按照指定順序(FIFO, LIFO, 優先順序)執行。示例程式碼如下:
ExecutorService newSingleThreadExecutor = Executors.newSingleThreadExecutor();
for (int i = 0; i < 10; i++) {
final int index = i;
newSingleThreadExecutor.execute(new Runnable() {
@Override
public void run() {
System.out.println("index:" + index);
try {
Thread.sleep(200);
} catch (Exception e) {
// TODO: handle exception
}
}
});
}
注意: 結果依次輸出,相當於順序執行各個任務。
執行緒池原理剖析
提交一個任務到執行緒池中,執行緒池的處理流程如下:
1、判斷執行緒池裡的核心執行緒是否都在執行任務,如果不是(核心執行緒空閒或者還有核心執行緒沒有被建立)則建立一個新的工作執行緒來執行任務。如果核心執行緒都在執行任務,則進入下個流程。
2、執行緒池判斷工作佇列是否已滿,如果工作佇列沒有滿,則將新提交的任務儲存在這個工作佇列裡。如果工作佇列滿了,則進入下個流程。
3、判斷執行緒池裡的執行緒是否都處於工作狀態,如果沒有,則建立一個新的工作執行緒來執行任務。如果已經滿了,則交給飽和策略來處理這個任務。
合理配置執行緒池
要想合理的配置執行緒池,就必須首先分析任務特性,可以從以下幾個角度來進行分析:
任務的性質:CPU密集型任務,IO密集型任務和混合型任務。
任務的優先順序:高,中和低。
任務的執行時間:長,中和短。
任務的依賴性:是否依賴其他系統資源,如資料庫連線。
任務性質不同的任務可以用不同規模的執行緒池分開處理。CPU密集型任務配置儘可能少的執行緒數量,如配置Ncpu+1個執行緒的執行緒池。IO密集型任務則由於需要等待IO操作,執行緒並不是一直在執行任務,則配置儘可能多的執行緒,如2*Ncpu。混合型的任務,如果可以拆分,則將其拆分成一個CPU密集型任務和一個IO密集型任務,只要這兩個任務執行的時間相差不是太大,那麼分解後執行的吞吐率要高於序列執行的吞吐率,如果這兩個任務執行時間相差太大,則沒必要進行分解。我們可以通過Runtime.getRuntime().availableProcessors()方法獲得當前裝置的CPU個數。
優先順序不同的任務可以使用優先順序佇列PriorityBlockingQueue來處理。它可以讓優先順序高的任務先得到執行,需要注意的是如果一直有優先順序高的任務提交到佇列裡,那麼優先順序低的任務可能永遠不能執行。
執行時間不同的任務可以交給不同規模的執行緒池來處理,或者也可以使用優先順序佇列,讓執行時間短的任務先執行。
依賴資料庫連線池的任務,因為執行緒提交SQL後需要等待資料庫返回結果,如果等待的時間越長CPU空閒時間就越長,那麼執行緒數應該設定越大,這樣才能更好的利用CPU。
一般總結哦,有其他更好的方式,希望各位留言,謝謝。
CPU密集型時,任務可以少配置執行緒數,大概和機器的cpu核數相當,這樣可以使得每個執行緒都在執行任務
IO密集型時,大部分執行緒都阻塞,故需要多配置執行緒數,2*cpu核數
作業系統之名稱解釋:
某些程序花費了絕大多數時間在計算上,而其他則在等待I/O上花費了大多是時間,
前者稱為計算密集型(CPU密集型)computer-bound,後者稱為I/O密集型,I/O-bound。
Java鎖的深度化
悲觀鎖、樂觀鎖、排他鎖
場景
當多個請求同時操作資料庫時,首先將訂單狀態改為已支付,在金額加上200,在同時併發場景查詢條件下,會造成重複通知。
SQL:
Update
悲觀鎖與樂觀鎖
悲觀鎖:悲觀鎖悲觀的認為每一次操作都會造成更新丟失問題,在每次查詢時加上排他鎖。
每次去拿資料的時候都認為別人會修改,所以每次在拿資料的時候都會上鎖,這樣別人想拿這個資料就會block直到它拿到鎖。傳統的關係型資料庫裡邊就用到了很多這種鎖機制,比如行鎖,表鎖等,讀鎖,寫鎖等,都是在做操作之前先上鎖。
Select * from xxx for update;
樂觀鎖:樂觀鎖會樂觀的認為每次查詢都不會造成更新丟失,利用版本欄位控制
重入鎖
鎖作為併發共享資料,保證一致性的工具,在JAVA平臺有多種實現(如 synchronized 和 ReentrantLock等等 ) 。這些已經寫好提供的鎖為我們開發提供了便利。
重入鎖,也叫做遞迴鎖,指的是同一執行緒 外層函式獲得鎖之後 ,內層遞迴函式仍然有獲取該鎖的程式碼,但不受影響。在JAVA環境下 ReentrantLock 和synchronized 都是 可重入鎖
public class Test implements Runnable {
public synchronized void get() {
System.out.println("name:" + Thread.currentThread().getName() + " get();");
set();
}
public synchronized void set() {
System.out.println("name:" + Thread.currentThread().getName() + " set();");
}
@Override
public void run() {
get();
}
public static void main(String[] args) {
Test ss = new Test();
new Thread(ss).start();
new Thread(ss).start();
new Thread(ss).start();
new Thread(ss).start();
}
}
public class Test02 extends Thread {
ReentrantLock lock = new ReentrantLock();
public void get() {
lock.lock();
System.out.println(Thread.currentThread().getId());
set();
lock.unlock();
}
public void set() {
lock.lock();
System.out.println(Thread.currentThread().getId());
lock.unlock();
}
@Override
public void run() {
get();
}
public static void main(String[] args) {
Test ss = new Test();
new Thread(ss).start();
new Thread(ss).start();
new Thread(ss).start();
}
}
讀寫鎖
相比Java中的鎖(Locks in Java)裡Lock實現,讀寫鎖更復雜一些。假設你的程式中涉及到對一些共享資源的讀和寫操作,且寫操作沒有讀操作那麼頻繁。在沒有寫操作的時候,兩個執行緒同時讀一個資源沒有任何問題,所以應該允許多個執行緒能在同時讀取共享資源。但是如果有一個執行緒想去寫這些共享資源,就不應該再有其它執行緒對該資源進行讀或寫(譯者注:也就是說:讀-讀能共存,讀-寫不能共存,寫-寫不能共存)。這就需要一個讀/寫鎖來解決這個問題。Java5在java.util.concurrent包中已經包含了讀寫鎖。儘管如此,我們還是應該瞭解其實現背後的原理。
public class Cache {
static Map<String, Object> map = new HashMap<String, Object>();
static ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
static Lock r = rwl.readLock();
static Lock w = rwl.writeLock();
// 獲取一個key對應的value
public static final Object get(String key) {
r.lock();
try {
System.out.println("正在做讀的操作,key:" + key + " 開始");
Thread.sleep(100);
Object object = map.get(key);
System.out.println("正在做讀的操作,key:" + key + " 結束");
System.out.println();
return object;
} catch (InterruptedException e) {
} finally {
r.unlock();
}
return key;
}
// 設定key對應的value,並返回舊有的value
public static final Object put(String key, Object value) {
w.lock();
try {
System.out.println("正在做寫的操作,key:" + key + ",value:" + value + "開始.");
Thread.sleep(100);
Object object = map.put(key, value);
System.out.println("正在做寫的操作,key:" + key + ",value:" + value + "結束.");
System.out.println();
return object;
} catch (InterruptedException e) {
} finally {
w.unlock();
}
return value;
}
// 清空所有的內容
public static final void clear() {
w.lock();
try {
map.clear();
} finally {
w.unlock();
}
}
public static void main(String[] args) {
new Thread(new Runnable() {
@Override
public void run() {
for (int i = 0; i < 10; i++) {
Cache.put(i + "", i + "");
}
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
for (int i = 0; i < 10; i++) {
Cache.get(i + "");
}
}
}).start();
}
}
CAS無鎖機制
(1)與鎖相比,使用比較交換(下文簡稱CAS)會使程式看起來更加複雜一些。但由於其非阻塞性,它對死鎖問題天生免疫,並且,執行緒間的相互影響也遠遠比基於鎖的方式要小。更為重要的是,使用無鎖的方式完全沒有鎖競爭帶來的系統開銷,也沒有執行緒間頻繁排程帶來的開銷,因此,它要比基於鎖的方式擁有更優越的效能。
(2)無鎖的好處:
第一,在高併發的情況下,它比有鎖的程式擁有更好的效能;
第二,它天生就是死鎖免疫的。
就憑藉這兩個優勢,就值得我們冒險嘗試使用無鎖的併發。
(3)CAS演算法的過程是這樣:它包含三個引數CAS(V,E,N): V表示要更新的變數,E表示預期值,N表示新值。僅當V值等於E值時,才會將V的值設為N,如果V值和E值不同,則說明已經有其他執行緒做了更新,則當前執行緒什麼都不做。最後,CAS返回當前V的真實值。
(4)CAS操作是抱著樂觀的態度進行的,它總是認為自己可以成功完成操作。當多個執行緒同時使用CAS操作一個變數時,只有一個會勝出,併成功更新,其餘均會失敗。失敗的執行緒不會被掛起,僅是被告知失敗,並且允許再次嘗試,當然也允許失敗的執行緒放棄操作。基於這樣的原理,CAS操作即使沒有鎖,也可以發現其他執行緒對當前執行緒的干擾,並進行恰當的處理。
(5)簡單地說,CAS需要你額外給出一個期望值,也就是你認為這個變數現在應該是什麼樣子的。如果變數不是你想象的那樣,那說明它已經被別人修改過了。你就重新讀取,再次嘗試修改就好了。
(6)在硬體層面,大部分的現代處理器都已經支援原子化的CAS指令。在JDK 5.0以後,虛擬機器便可以使用這個指令來實現併發操作和併發資料結構,並且,這種操作在虛擬機器中可以說是無處不在。
/**
* Atomically increments by one the current value.
*
* @return the updated value
*/
public final int incrementAndGet() {
for (;;) {
//獲取當前值
int current = get();
//設定期望值
int next = current + 1;
//呼叫Native方法compareAndSet,執行CAS操作
if (compareAndSet(current, next))
//成功後才會返回期望值,否則無線迴圈
return next;
}
}
自旋鎖
自旋鎖是採用讓當前執行緒不停地的在迴圈體內執行實現的,當迴圈的條件被其他執行緒改變時 才能進入臨界區。如下
private AtomicReference<Thread> sign =new AtomicReference<>();
public void lock() {
Thread current = Thread.currentThread();
while (!sign.compareAndSet(null, current)) {
}
}
public void unlock() {
Thread current = Thread.currentThread();
sign.compareAndSet(current, null);
}
public class Test implements Runnable {
static int sum;
private SpinLock lock;
public Test(SpinLock lock) {
this.lock = lock;
}
/**
* @param args
* @throws InterruptedException
*/
public static void main(String[] args) throws InterruptedException {
SpinLock lock = new SpinLock();
for (int i = 0; i < 100; i++) {
Test test = new Test(lock);
Thread t = new Thread(test);
t.start();
}
Thread.currentThread().sleep(1000);
System.out.println(sum);
}
@Override
public void run() {
this.lock.lock();
this.lock.lock();
sum++;
this.lock.unlock();
this.lock.unlock();
}
}
當一個執行緒 呼叫這個不可重入的自旋鎖去加鎖的時候沒問題,當再次呼叫lock()的時候,因為自旋鎖的持有引用已經不為空了,該執行緒物件會誤認為是別人的執行緒持有了自旋鎖
使用了CAS原子操作,lock函式將owner設定為當前執行緒,並且預測原來的值為空。unlock函式將owner設定為null,並且預測值為當前執行緒。
當有第二個執行緒呼叫lock操作時由於owner值不為空,導致迴圈一直被執行,直至第一個執行緒呼叫unlock函式將owner設定為null,第二個執行緒才能進入臨界區。
由於自旋鎖只是將當前執行緒不停地執行迴圈體,不進行執行緒狀態的改變,所以響應速度更快。但當執行緒數不停增加時,效能下降明顯,因為每個執行緒都需要執行,佔用CPU時間。如果執行緒競爭不激烈,並且保持鎖的時間段。適合使用自旋鎖。
分散式鎖
如果想在不同的jvm中保證資料同步,使用分散式鎖技術。
有資料庫實現、快取實現、Zookeeper分散式鎖