Concurrency(十一: 飢餓與公平)

併發 Java · 發表 2019-04-08 07:47:02

摘要：如果一個執行緒因為其他執行緒佔滿了而無法獲取CPU執行時間,這種情況我們稱之為“飢餓現象”.執行緒將一直飢餓下去,因為其他執行緒總能替代它獲取CPU執行時間.解決這種情況的措施我們稱之為“公平措施”.即讓所有執行緒都能獲得一次執行的機會. Java中導致飢餓現象的原因在Java中,以...

如果一個執行緒因為其他執行緒佔滿了而無法獲取CPU執行時間,這種情況我們稱之為“飢餓現象”.執行緒將一直飢餓下去,因為其他執行緒總能替代它獲取CPU執行時間.解決這種情況的措施我們稱之為“公平措施”.即讓所有執行緒都能獲得一次執行的機會.

Java中導致飢餓現象的原因

在Java中,以下三種情況能夠產生飢餓現象:

優先順序高的執行緒蠶食了優先順序低的執行緒的所有CPU執行時間.
執行緒無限期的阻塞進入同步程式碼塊因為其他執行緒總是能夠在它之前進入.
執行緒無限期的等待(呼叫wait())喚醒因為其他執行緒總是能夠在它之間被喚醒(接收到notify()訊號).

優先順序高的執行緒蠶食了優先順序低的執行緒的所有CPU執行時間

你可以分別對每一個執行緒設定優秀級.高優先順序的執行緒能夠獲取到更多的CPU執行時間.你可以為執行緒設定1~10的優先順序,但這完全依賴於應用執行在哪個作業系統之上.對於大多數應用採用預設優秀級就好.

執行緒無限期的阻塞進入同步程式碼塊

Java的同步程式碼塊是引起飢餓現象的另一個原因.Java同步程式碼塊沒辦法保證在等待中的執行緒能夠按照某種序列進入同步程式碼塊.這意味著理論上會有一個執行緒無限期的等待進入同步程式碼塊,因為其他執行緒總是能夠在它之前進入同步程式碼塊.這種問題也稱之為"飢餓現象",該執行緒將會一直飢餓下去,因為其他執行緒總能替代它獲取CPU執行時間.

執行緒無限期的等待喚醒

在多個執行緒同時呼叫同一個物件的wait() 方法時,notify() 方法無法保證喚醒哪個執行緒.這會導致某些執行緒一直在等待.這會產生一個執行緒一直在等待喚醒的風險,因為其他執行緒總能在它之前被喚醒.

Java實現公平措施

雖然在Java中不可能實現百分百的公平措施,但我們仍然可以實現自己的同步器結構來增加執行緒間的公平性.

我們先來學習一個簡單的同步器程式碼塊:

public class Synchronizer{
public synchronized void doSynchronized(){
// 花費相當長的時間來執行工作
}
}
複製程式碼

如果有多於一個的執行緒呼叫doSynchronized方法,那麼其他執行緒都需要等待第一個進入同步程式碼塊的執行緒退出方法.只要有多於一個執行緒在等待狀態,就不能保證哪個執行緒先被允許進入同步程式碼塊.

使用鎖來代替同步程式碼塊

為了增加等待執行緒的公平性,首先我們需要使用鎖來代替同步程式碼塊來實現同步機制.

public class Synchronizer{
Lock lock = new Lock();

public void doSynchronized() throws InterruptedException{
this.lock.lock();
// 臨界區程式碼, 花費相當長的時間來執行工作
this.lock.unLock();
}
}
複製程式碼

我們注意到doSynchronized不再使用synchronized 來宣告.取而代之的是將臨界區程式碼放置在lock.lock()和lock.unLock()方法呼叫之間.

一個簡單Lock類實現:

public class Lock{
public class Lock {
private boolean isLocked = false;
private Thread lockingThread;

public synchronized void lock() throws InterruptedException {
while (isLocked){
wait();
}
isLocked = true;
lockingThread = Thread.currentThread();
}

public synchronized void unLock(){
if(lockingThread != null && lockingThread == Thread.currentThread()){
throw new IllegalMonitorStateException("Calling thread is not locked this lock");
}
isLocked = false;
lockingThread = null;
notify();
}
}
複製程式碼

如果你看了上文的Synchronizer和現在的Lock實現,你會發現如果有多於一個執行緒同時訪問lock()方法時,會發生阻塞.其次,如果當前的鎖為鎖住狀態的話,執行緒會在lock方法中的while迴圈內部呼叫wait()方法從而進入等待狀態.記住,一個執行緒一旦呼叫wait()完畢,則會釋放當前物件鎖,讓其他執行緒可以進入lock()方法.最後的結果是多個執行緒進入lock()方法的while迴圈中呼叫wait()方法進入等待狀態.

如果你回頭看doSynchronized()方法,你會發現lock()和unlock()狀態切換中間的註釋,這將花費相當長的時間來執行兩個方法呼叫間的程式碼.需要我們確認的是執行這些程式碼需要花費相當長的時間來比較進入lock()方法和在鎖被鎖住的情況下呼叫wait()方法.這意味著大部分時間都用來等待鎖的鎖住和在lock()方法內部呼叫的wait()方法完畢後等待退出wait()方法.而不是等待進入lock()方法.

在之前的同步程式碼塊的狀態下,如果有多個執行緒等待進入同步程式碼塊,無法保證哪個執行緒先進入同步程式碼塊.同樣的在呼叫wait()方法進入等待狀態後,無法保證呼叫notify()後哪個執行緒會先被喚醒.所以當前Lock的實現版本並不比之前的synchronized版本的doSynchronized()公平到哪去.但我們可以稍作更改.

當前Lock版本呼叫的是它自己的wait()方法.如果將每個執行緒呼叫的wait()方法替換成不同物件的.即每個執行緒對應呼叫一個物件的wait()方法,即Lock能夠決定在往後的時間裡到底呼叫哪個物件的notify()方法,這樣就能具體有效的選擇喚醒哪個執行緒.

一個公平鎖的實現

以下內容是將上文提及的Lock.class轉變為一個公平鎖即FairLock.class.你會發現與之前版本對比,這個實現僅是調整了同步程式碼塊和wait()/notify()的呼叫方式而已.

實際上在得到當前這個版本的公平鎖之前遇到了許許多多的問題,而每解決這其中的一個問題都需要長篇概論來闡述,解決這些問題的每一個步驟都會在往後的主題中提及.這包含巢狀監控器鎖死, 滑動條件和訊號丟失問題. 現在重點要知道的是執行緒以佇列的方式來呼叫lock()方法中的wait()方法,且每次在公平鎖未鎖住時僅能讓佇列頭部的執行緒來獲取和鎖住公平鎖例項.其他執行緒則處在等待狀態直到進入佇列頭部.

public class FairLock {
private boolean isLocked = false;
private Thread lockingThread;
private List<QueueObject> waitingThreads = new ArrayList<>();

public void lock() throws InterruptedException {
// 1. 為每個執行緒建立一個QueueObject
QueueObject queueObject = new QueueObject();
boolean isLockedForThisThread = true;
synchronized (this) {
// 2. 添加當前執行緒的QueueObject到佇列中
waitingThreads.add(queueObject);
}

while (isLockedForThisThread) {
synchronized (this) {
isLockedForThisThread = isLocked || waitingThreads.get(0) != queueObject;
if (!isLockedForThisThread) {
// 3. 鎖住當前公平鎖
isLocked = true;
waitingThreads.remove(queueObject);
lockingThread = Thread.currentThread();
return;
}
}
try {
// 4. 呼叫該執行緒對應QueueObject的wait()方法進入等待狀態
queueObject.doWait();
} catch (InterruptedException e) {
synchronized (this) {
waitingThreads.remove(queueObject);
}
throw e;
}
}
}

public synchronized void unlock() {
if (this.lockingThread != Thread.currentThread()) {
throw new IllegalMonitorStateException("Calling thread has not locked this lock");
}
// 1. 釋放公平鎖
isLocked = false;
lockingThread = null;
if (waitingThreads.size() > 0) {
// 2. 呼叫佇列頭部執行緒一一對應的QueueObject喚醒執行緒
waitingThreads.get(0).doNotify();
}
}
}
複製程式碼

public class QueueObject {
private boolean isNotified = false;

public synchronized void doWait() throws InterruptedException {
while (!isNotified) {
this.wait();
}
this.isNotified = false;
}

public synchronized void doNotify() {
this.isNotified = true;
this.notify();
}

public boolean equals(Object o) {
return this == o;
}
}
複製程式碼

首先你會注意到lock不再宣告為synchronized .取而代之的是將需要做同步限制的程式碼塊巢狀到synchronized 程式碼塊中.

每個執行緒呼叫lock()方法後都會建立與之對應的QueueObject 例項,並進入到佇列中.執行緒呼叫unlock()方法後會從佇列的頭部取得QueueObject 物件並呼叫它的doNotify()方法來喚醒與之對應的執行緒.對於所有等待的執行緒來說,這種方式每次僅會喚醒一個執行緒.這部分就是FairLock用來確保公平的程式碼.

我們注意到lock的鎖住狀態會在同一個程式碼塊中不停的檢查和設定來解決滑動條件帶來的問題.

同時我們注意到QueueObject 就是一個Semaphore .doWait()和doNotify()呼叫所產生的狀態會儲存在QueueObject 內部.這用來解決訊號丟失問題,即一個執行緒在呼叫queueObject().doWait()時,被另一個執行緒搶先機會呼叫了unlock()中的queueObject.doNotify(). queueObject.doWait()呼叫被放置在synchronized(this) 同步程式碼塊之外,用於解決巢狀監控器鎖死問題.這樣當沒有執行緒在lock()方法的synchronized(this) 程式碼塊中執行時,其他執行緒可以正常呼叫unLock()方法.

最後需要注意的是,為什麼需要將queueObject.doWait()呼叫放置在try-catch中.當執行緒通過丟擲InterruptedException 來終止lock()方法呼叫時,我們需要將執行緒與之對應的QueueObject 踢出佇列.