摘要：饑餓和公平一個(gè)線程因?yàn)闀r(shí)間全部被其他線程搶走而得不到運(yùn)行時(shí)間，這種狀態(tài)被稱之為饑餓。線程需要同時(shí)持有對(duì)象和對(duì)象的鎖，才能向線程發(fā)信號(hào)?，F(xiàn)在兩個(gè)線程都檢查了這個(gè)條件為，然后它們都會(huì)繼續(xù)進(jìn)入第二個(gè)同步塊中并設(shè)置為。

產(chǎn)生死鎖的四個(gè)必要條件：
（1） 互斥條件：一個(gè)資源每次只能被一個(gè)進(jìn)程使用。
（2） 請(qǐng)求與保持條件：一個(gè)進(jìn)程因請(qǐng)求資源而阻塞時(shí)，對(duì)已獲得的資源保持不放。
（3） 不剝奪條件:進(jìn)程已獲得的資源，在末使用完之前，不能強(qiáng)行剝奪。
（4） 循環(huán)等待條件:若干進(jìn)程之間形成一種頭尾相接的循環(huán)等待資源關(guān)系。

類似于下圖：

甚至?xí)懈鼜?fù)雜的，環(huán)狀死鎖:

Thread 1  locks A, waits for B
Thread 2  locks B, waits for C
Thread 3  locks C, waits for D
Thread 4  locks D, waits for A

public class DeadLock implements Runnable {  
    public int flag = 1;  
    //靜態(tài)對(duì)象是類的所有對(duì)象共享的  
    private static Object o1 = new Object(), o2 = new Object();  
    @Override  
    public void run() {  
        if (flag == 1) {  
            synchronized (o1) {  
                synchronized (o2) {  
                    System.out.println("1");  
                }  
            }  
        }  
        if (flag == 0) {  
            synchronized (o2) {  
                synchronized (o1) {  
                    System.out.println("0");  
                }  
            }  
        }  
    }  
  
    public static void main(String[] args) {  
          
        DeadLock td1 = new DeadLock();  
        DeadLock td2 = new DeadLock();  
        
        td1.flag = 1;  
        td2.flag = 0;  
        
        //td1,td2都處于可執(zhí)行狀態(tài)，但JVM線程調(diào)度先執(zhí)行哪個(gè)線程是不確定的。  
        //td2的run()可能在td1的run()之前運(yùn)行  
        new Thread(td1).start();  
        new Thread(td2).start();  
  
    }  
}  

確保所有的線程都是按照相同的順序獲得鎖，那么死鎖就不會(huì)發(fā)生。

Thread 1:
lock A
lock B

Thread 2:
wait for A
lock C (when A locked)

Thread 3:
wait for A
wait for B
wait for C

獲取鎖的時(shí)候加一個(gè)超時(shí)時(shí)間,若一個(gè)線程沒(méi)有在給定的時(shí)限內(nèi)成功獲得所有需要的鎖，則會(huì)進(jìn)行回退并釋放所有已經(jīng)獲得的鎖，然后等待一段隨機(jī)的時(shí)間再重試。

以下是一個(gè)例子，展示了兩個(gè)線程以不同的順序嘗試獲取相同的兩個(gè)鎖，在發(fā)生超時(shí)后回退并重試的場(chǎng)景：

Thread 1 locks A
Thread 2 locks B
Thread 1 attempts to lock B but is blocked
Thread 2 attempts to lock A but is blocked
Thread 1"s lock attempt on B times out
Thread 1 backs up and releases A as well
Thread 1 waits randomly (e.g. 257 millis) before retrying.
Thread 2"s lock attempt on A times out
Thread 2 backs up and releases B as well
Thread 2 waits randomly (e.g. 43 millis) before retrying.

當(dāng)然，如果有非常多的線程同一時(shí)間去競(jìng)爭(zhēng)同一批資源，就算有超時(shí)和回退機(jī)制，還是可能會(huì)導(dǎo)致這些線程重復(fù)地嘗試但卻始終得不到鎖。

在Java中不能對(duì)synchronized同步塊設(shè)置超時(shí)時(shí)間，需要?jiǎng)?chuàng)建一個(gè)自定義鎖！

每當(dāng)一個(gè)線程請(qǐng)求鎖，或者獲得了鎖，可以在線程和鎖相關(guān)的數(shù)據(jù)結(jié)構(gòu)中（map、graph等等）將其記下。當(dāng)一個(gè)線程請(qǐng)求鎖失敗時(shí)，這個(gè)線程可以遍歷鎖的關(guān)系圖看看是否有死鎖發(fā)生。

死鎖一般要比兩個(gè)線程互相持有對(duì)方的鎖這種情況要復(fù)雜的多,下面是一幅關(guān)于四個(gè)線程（A,B,C和D）之間鎖占有和請(qǐng)求的關(guān)系圖。像這樣的數(shù)據(jù)結(jié)構(gòu)就可以被用來(lái)檢測(cè)死鎖。

那么當(dāng)檢測(cè)出死鎖時(shí),可以按下面方式來(lái)處理：

釋放所有鎖，回退，并且等待一段隨機(jī)的時(shí)間后重試。

給這些線程設(shè)置優(yōu)先級(jí)，讓一個(gè)（或幾個(gè)）線程回退，剩下的線程就像沒(méi)發(fā)生死鎖一樣繼續(xù)保持著它們需要的鎖。

一個(gè)線程因?yàn)镃PU時(shí)間全部被其他線程搶走而得不到CPU運(yùn)行時(shí)間，這種狀態(tài)被稱之為饑餓。

解決饑餓的方案,所有線程均能公平地獲得運(yùn)行機(jī)會(huì)被稱之為公平性

在Java中，下面三個(gè)常見(jiàn)的原因會(huì)導(dǎo)致線程饑餓：

高優(yōu)先級(jí)線程吞噬所有的低優(yōu)先級(jí)線程的CPU時(shí)間。

線程被永久堵塞在一個(gè)等待進(jìn)入同步塊的狀態(tài)
Java的同步代碼區(qū)對(duì)哪個(gè)線程允許進(jìn)入的次序沒(méi)有任何保障。理論上存在一個(gè)試圖進(jìn)入該同步區(qū)的線程處于被永久堵塞的風(fēng)險(xiǎn)，因?yàn)槠渌€程總是能持續(xù)地先于它獲得訪問(wèn)

線程在等待一個(gè)本身(在其上調(diào)用wait())也處于永久等待完成的對(duì)象
如果多個(gè)線程處在wait()方法執(zhí)行上，而對(duì)其調(diào)用notify()不會(huì)保證哪一個(gè)線程會(huì)獲得喚醒，任何線程都有可能處于繼續(xù)等待的狀態(tài)。因此存在這樣一個(gè)風(fēng)險(xiǎn)：一個(gè)等待線程從來(lái)得不到喚醒，因?yàn)槠渌却€程總是能被獲得喚醒。

在java中不可能實(shí)現(xiàn)100%的公平性，為了提高等待線程的公平性，我們使用鎖方式來(lái)替代同步塊。

public class Synchronizer{
    Lock lock = new Lock();
    
    //使用lock，而不是synchronized實(shí)現(xiàn)同步塊
    public void doSynchronized() throws InterruptedException{
        this.lock.lock();
        //critical section, do a lot of work which takes a long time
        this.lock.unlock();
    }
}

Lock的簡(jiǎn)單實(shí)現(xiàn)原理：

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

    public synchronized void lock() throws InterruptedException{

    while(isLocked){
        wait();
    }

    isLocked = true;
    lockingThread = Thread.currentThread();

}

    public synchronized void unlock(){
        if(this.lockingThread != Thread.currentThread()){
            throw new IllegalMonitorStateException(
                "Calling thread has not locked this lock");
            }

        isLocked = false;
        lockingThread = null;
        notify();
    }
}

上面的例子可以看到兩點(diǎn)：

如果多個(gè)線程同時(shí)調(diào)用lock.lock方法的話，線程將阻塞在lock方法處，因?yàn)閘ock方法是一個(gè)同步方法。

如果lock對(duì)象的鎖被一個(gè)線程持有，那么其他線程都將調(diào)用在while循環(huán)中的wait方法而阻塞。

現(xiàn)在在把目光集中在doSynchronized方法中，在lock和unlock之間有一段注釋，寫明了這一段代碼將執(zhí)行很長(zhǎng)一段時(shí)間。我們假設(shè)這段時(shí)間比線程進(jìn)入lock方法內(nèi)部并且由于lock已被鎖定而調(diào)用wait方法等待的時(shí)間長(zhǎng)。這意味著線程大部分時(shí)間都消耗在了wait等待上而不是阻塞在lock方法上。

之前曾提到同步塊無(wú)法保證當(dāng)多個(gè)線程等待進(jìn)入同步塊中時(shí)哪個(gè)線程先進(jìn)入，同樣notify方法也無(wú)法保證在多個(gè)線程調(diào)用wait的情況下哪個(gè)線程先被喚醒。當(dāng)前這個(gè)版本的Lock類在公平性上和之前加了synchronized關(guān)鍵字的doSynchronized方法沒(méi)什么區(qū)別，但是我們可以修改它。

我們注意到，當(dāng)前版本的Lock方法是調(diào)用自己的wait方法。如果每個(gè)線程調(diào)用不同對(duì)象的wait方法，那么Lock類就可以決定哪些對(duì)象調(diào)用notify方法，這樣就可以選擇性的喚醒線程。

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

  public void lock() throws InterruptedException{
    QueueObject queueObject           = new QueueObject();
    boolean     isLockedForThisThread = true;
    synchronized(this){
        waitingThreads.add(queueObject);
    }

    while(isLockedForThisThread){
      synchronized(this){
        isLockedForThisThread =
            isLocked || waitingThreads.get(0) != queueObject;
        if(!isLockedForThisThread){
          isLocked = true;
           waitingThreads.remove(queueObject);
           lockingThread = Thread.currentThread();
           return;
         }
      }
      try{
        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");
    }
    isLocked      = false;
    lockingThread = null;
    if(waitingThreads.size() > 0){
      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;
  }
}

FairLock類會(huì)給每個(gè)調(diào)用lock方法的線程創(chuàng)建一個(gè)QueueObject對(duì)象，當(dāng)線程調(diào)用unlock方法時(shí)隊(duì)列中的第一個(gè)。
QueueObject出列并且調(diào)用doNotify方法激活對(duì)應(yīng)的線程。這種方式可以保證只有一個(gè)線程被喚醒而不是所有等待線程。

注意到FairLock在同步塊中設(shè)置了狀態(tài)檢測(cè)來(lái)避免失控。

QueueObject實(shí)際上就是一個(gè)信號(hào)量(semaphore),QueueObject對(duì)象內(nèi)部保存了一個(gè)信號(hào)isNotified.這樣做是為了防止信號(hào)丟失。queueObject.wait方法是被放在了synchronized(this)塊的外部來(lái)避免嵌套監(jiān)視器閉環(huán)。這樣當(dāng)沒(méi)有線程運(yùn)行l(wèi)ock方法中的synchronized同步塊時(shí)其他線程可以調(diào)用unlock方法。

最后我們注意到lock方法用到了try-catch塊，這樣當(dāng)發(fā)生InterruptedException時(shí)線程將退出lock方法，這個(gè)時(shí)候我們應(yīng)該將對(duì)應(yīng)的QueueObject對(duì)象出列。

效率
FairLock的執(zhí)行效率相比Lock類要低一些。它對(duì)你的應(yīng)用程序的影響取決于FairLock所保證的臨界區(qū)代碼的執(zhí)行時(shí)間，這個(gè)時(shí)間越長(zhǎng)，那么影響就越?。煌瑫r(shí)也取決于這段臨界區(qū)代碼的執(zhí)行頻率。

嵌套管程鎖死與死鎖類似，場(chǎng)景如下所示：

線程1獲得A對(duì)象的鎖。
線程1獲得對(duì)象B的鎖（同時(shí)持有對(duì)象A的鎖）。
線程1決定等待另一個(gè)線程的信號(hào)再繼續(xù)。
線程1調(diào)用B.wait()，從而釋放了B對(duì)象上的鎖，但仍然持有對(duì)象A的鎖。
線程2需要同時(shí)持有對(duì)象A和對(duì)象B的鎖，才能向線程1發(fā)信號(hào)。
線程2無(wú)法獲得對(duì)象A上的鎖，因?yàn)閷?duì)象A上的鎖當(dāng)前正被線程1持有。
線程2一直被阻塞，等待線程1釋放對(duì)象A上的鎖。
線程1一直阻塞，等待線程2的信號(hào)，因此，不會(huì)釋放對(duì)象A上的鎖，
而線程2需要對(duì)象A上的鎖才能給線程1發(fā)信號(hào)……
代碼示例：

//lock implementation with nested monitor lockout problem
public class Lock{
    protected MonitorObject monitorObject = new MonitorObject();
    protected boolean isLocked = false;

    public void lock() throws InterruptedException{
        synchronized(this){
            while(isLocked){
                synchronized(this.monitorObject){
                    this.monitorObject.wait();
                }
            }
            isLocked = true;
        }
    }

    public void unlock(){
        synchronized(this){
            this.isLocked = false;
            synchronized(this.monitorObject){
                this.monitorObject.notify();
            }
        }
    }
}

區(qū)別

在死鎖中我們已經(jīng)對(duì)死鎖有了個(gè)大概的解釋，死鎖通常是因?yàn)閮蓚€(gè)線程獲取鎖的順序不一致造成的，線程1鎖住A，等待獲取B，線程2已經(jīng)獲取了B，再等待獲取A。如死鎖避免中所說(shuō)的，死鎖可以通過(guò)總是以相同的順序獲取鎖來(lái)避免。

但是發(fā)生嵌套管程鎖死時(shí)鎖獲取的順序是一致的。線程1獲得A和B，然后釋放B，等待線程2的信號(hào)。線程2需要同時(shí)獲得A和B，才能向線程1發(fā)送信號(hào)。所以，一個(gè)線程在等待喚醒，另一個(gè)線程在等待想要的鎖被釋放。

死鎖中，二個(gè)線程都在等待對(duì)方釋放鎖。

嵌套管程鎖死中，線程1持有鎖A，同時(shí)等待從線程2發(fā)來(lái)的信號(hào)，線程2需要鎖A來(lái)發(fā)信號(hào)給線程1。

從一個(gè)線程檢查某一特定條件到該線程操作此條件期間，這個(gè)條件已經(jīng)被其它線程改變，導(dǎo)致第一個(gè)線程在該條件上執(zhí)行了錯(cuò)誤的操作。這里有一個(gè)簡(jiǎn)單的例子：

public class Lock {
    private boolean isLocked = true;

    public void lock(){
      synchronized(this){
        while(isLocked){
          try{
            this.wait();
          } catch(InterruptedException e){
            //do nothing, keep waiting
          }
        }
      }

      synchronized(this){
        isLocked = true;
      }
    }

    public synchronized void unlock(){
      isLocked = false;
      this.notify();
    }
}

假如在某個(gè)時(shí)刻isLocked為false，有兩個(gè)線程同時(shí)訪問(wèn)lock方法。如果第一個(gè)線程先進(jìn)入第一個(gè)同步塊，這個(gè)時(shí)候它會(huì)發(fā)現(xiàn)isLocked為false，若此時(shí)允許第二個(gè)線程執(zhí)行，它也進(jìn)入第一個(gè)同步塊，同樣發(fā)現(xiàn)isLocked是false。現(xiàn)在兩個(gè)線程都檢查了這個(gè)條件為false，然后它們都會(huì)繼續(xù)進(jìn)入第二個(gè)同步塊中并設(shè)置isLocked為true。

為避免slipped conditions，條件的檢查與設(shè)置必須是原子的，也就是說(shuō)，在第一個(gè)線程檢查和設(shè)置條件期間，不會(huì)有其它線程檢查這個(gè)條件。

public class Lock {
    private boolean isLocked = true;

    public void lock(){
      synchronized(this){
        while(isLocked){
          try{
            this.wait();
          } catch(InterruptedException e){
            //do nothing, keep waiting
          }
        }
        isLocked = true;
      }
    }

    public synchronized void unlock(){
      isLocked = false;
      this.notify();
    }
}

Semaphore（信號(hào)量） 是一個(gè)線程同步結(jié)構(gòu)，用于在線程間傳遞信號(hào)，以避免出現(xiàn)信號(hào)丟失，或者像鎖一樣用于保護(hù)一個(gè)關(guān)鍵區(qū)域。

public class Semaphore {
    private boolean signal = false;

    public synchronized void take() {
        this.signal = true;
        this.notify();
    }

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

Take 方法發(fā)出一個(gè)被存放在 Semaphore內(nèi)部的信號(hào)，而Release方法則等待一個(gè)信號(hào)，當(dāng)其接收到信號(hào)后，標(biāo)記位 signal 被清空，然后該方法終止。

使用這個(gè) semaphore 可以避免錯(cuò)失某些信號(hào)通知。用 take 方法來(lái)代替 notify，release 方法來(lái)代替 wait。如果某線程在調(diào)用 release 等待之前調(diào)用 take 方法，那么調(diào)用 release 方法的線程仍然知道 take 方法已經(jīng)被某個(gè)線程調(diào)用過(guò)了，因?yàn)樵?Semaphore 內(nèi)部保存了 take 方法發(fā)出的信號(hào)。而 wait 和 notify 方法就沒(méi)有這樣的功能。

阻塞隊(duì)列與普通隊(duì)列的區(qū)別在于

當(dāng)隊(duì)列是空的時(shí)，從隊(duì)列中獲取元素的操作將會(huì)被阻塞。
當(dāng)隊(duì)列是滿時(shí)，往隊(duì)列里添加元素的操作會(huì)被阻塞。

試圖從空的阻塞隊(duì)列中獲取元素的線程將會(huì)被阻塞，直到其他的線程往空的隊(duì)列插入新的元素。同樣，試圖往已滿的阻塞隊(duì)列中添加新元素的線程同樣也會(huì)被阻塞，直到其他的線程使隊(duì)列重新變得空閑起來(lái)。

public class BlockingQueue {

    private List queue = new LinkedList();

    private int  limit = 10;

    public BlockingQueue(int limit){
        this.limit = limit;
    }

    public synchronized void enqueue(Object item) throws InterruptedException  {
        while(this.queue.size() == this.limit) {
            wait();
        }
        
         if(this.queue.size() == 0) {
            notifyAll();
        }

         this.queue.add(item);
}

    public synchronized Object dequeue() throws InterruptedException{

         while(this.queue.size() == 0){
              wait();
        }

        if(this.queue.size() == this.limit){
            notifyAll();
        }   

        return this.queue.remove(0);
    }
}

必須注意到，在 enqueue 和 dequeue 方法內(nèi)部，只有隊(duì)列的大小等于上限（limit）或者下限（0）時(shí)，才調(diào)用notifyAll方法。
如果隊(duì)列的大小既不等于上限，也不等于下限，任何線程調(diào)用 enqueue 或者 dequeue 方法時(shí)，都不會(huì)阻塞，都能夠正常的往隊(duì)列中添加或者移除元素。

線程池（Thread Pool）對(duì)于限制應(yīng)用程序中同一時(shí)刻運(yùn)行的線程數(shù)很有用。因?yàn)槊繂?dòng)一個(gè)新線程都會(huì)有相應(yīng)的性能開(kāi)銷，每個(gè)線程都需要給棧分配一些內(nèi)存等等。

我們可以把并發(fā)執(zhí)行的任務(wù)傳遞給一個(gè)線程池，來(lái)替代為每個(gè)并發(fā)執(zhí)行的任務(wù)都啟動(dòng)一個(gè)新的線程。只要池里有空閑的線程，任務(wù)就會(huì)分配給一個(gè)線程執(zhí)行。在線程池的內(nèi)部，任務(wù)被插入一個(gè)阻塞隊(duì)列（Blocking Queue），線程池里的線程會(huì)去取這個(gè)隊(duì)列里的任務(wù)。當(dāng)一個(gè)新任務(wù)插入隊(duì)列時(shí)，一個(gè)空閑線程就會(huì)成功的從隊(duì)列中取出任務(wù)并且執(zhí)行它。

線程池經(jīng)常應(yīng)用在多線程服務(wù)器上。每個(gè)通過(guò)網(wǎng)絡(luò)到達(dá)服務(wù)器的連接都被包裝成一個(gè)任務(wù)并且傳遞給線程池。線程池的線程會(huì)并發(fā)的處理連接上的請(qǐng)求。

public class PoolThread extends Thread {

  private BlockingQueue taskQueue = null;
  private boolean       isStopped = false;

  public PoolThread(BlockingQueue queue) {
    taskQueue = queue;
  }

  public void run() {
    while (!isStopped()) {
      try {
        Runnable runnable =taskQueue.take();
        runnable.run();
      } catch(Exception e) {
        // 寫日志或者報(bào)告異常,
        // 但保持線程池運(yùn)行.
      }
    }
  }

  public synchronized void toStop() {
    isStopped = true;
    this.interrupt(); // 打斷池中線程的 dequeue() 調(diào)用.
  }

  public synchronized boolean isStopped() {
    return isStopped;
  }
}