Java并發(fā)

venmos 發(fā)布于2019-08-14 17:55 / 2570人閱讀

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

1、死鎖

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

類似于下圖：

甚至會有更復雜的，環(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

java代碼示例

public class DeadLock implements Runnable {  
    public int flag = 1;  
    //靜態(tài)對象是類的所有對象共享的  
    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í)行哪個線程是不確定的。  
        //td2的run()可能在td1的run()之前運行  
        new Thread(td1).start();  
        new Thread(td2).start();  
  
    }  
}

加鎖順序

確保所有的線程都是按照相同的順序獲得鎖，那么死鎖就不會發(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

加鎖時限

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

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

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.

當然，如果有非常多的線程同一時間去競爭同一批資源，就算有超時和回退機制，還是可能會導致這些線程重復地嘗試但卻始終得不到鎖。

在Java中不能對synchronized同步塊設置超時時間，需要創(chuàng)建一個自定義鎖！

死鎖檢測

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

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

那么當檢測出死鎖時,可以按下面方式來處理：

釋放所有鎖，回退，并且等待一段隨機的時間后重試。

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

2、饑餓和公平

一個線程因為CPU時間全部被其他線程搶走而得不到CPU運行時間，這種狀態(tài)被稱之為饑餓。

解決饑餓的方案,所有線程均能公平地獲得運行機會被稱之為公平性

饑餓原因

在Java中，下面三個常見的原因會導致線程饑餓：

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

線程被永久堵塞在一個等待進入同步塊的狀態(tài)
Java的同步代碼區(qū)對哪個線程允許進入的次序沒有任何保障。理論上存在一個試圖進入該同步區(qū)的線程處于被永久堵塞的風險，因為其他線程總是能持續(xù)地先于它獲得訪問

線程在等待一個本身(在其上調(diào)用wait())也處于永久等待完成的對象
如果多個線程處在wait()方法執(zhí)行上，而對其調(diào)用notify()不會保證哪一個線程會獲得喚醒，任何線程都有可能處于繼續(xù)等待的狀態(tài)。因此存在這樣一個風險：一個等待線程從來得不到喚醒，因為其他等待線程總是能被獲得喚醒。

公平性

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

public class Synchronizer{
    Lock lock = new Lock();
    
    //使用lock，而不是synchronized實現(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的簡單實現(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ào)用lock.lock方法的話，線程將阻塞在lock方法處，因為lock方法是一個同步方法。

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

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

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

我們注意到，當前版本的Lock方法是調(diào)用自己的wait方法。如果每個線程調(diào)用不同對象的wait方法，那么Lock類就可以決定哪些對象調(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類會給每個調(diào)用lock方法的線程創(chuàng)建一個QueueObject對象，當線程調(diào)用unlock方法時隊列中的第一個。
QueueObject出列并且調(diào)用doNotify方法激活對應的線程。這種方式可以保證只有一個線程被喚醒而不是所有等待線程。

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

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

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

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

3、嵌套管程鎖死

嵌套管程鎖死與死鎖類似，場景如下所示：

線程1獲得A對象的鎖。
線程1獲得對象B的鎖（同時持有對象A的鎖）。
線程1決定等待另一個線程的信號再繼續(xù)。
線程1調(diào)用B.wait()，從而釋放了B對象上的鎖，但仍然持有對象A的鎖。

線程2需要同時持有對象A和對象B的鎖，才能向線程1發(fā)信號。
線程2無法獲得對象A上的鎖，因為對象A上的鎖當前正被線程1持有。
線程2一直被阻塞，等待線程1釋放對象A上的鎖。

線程1一直阻塞，等待線程2的信號，因此，不會釋放對象A上的鎖，
而線程2需要對象A上的鎖才能給線程1發(fā)信號……

代碼示例：

//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)對死鎖有了個大概的解釋，死鎖通常是因為兩個線程獲取鎖的順序不一致造成的，線程1鎖住A，等待獲取B，線程2已經(jīng)獲取了B，再等待獲取A。如死鎖避免中所說的，死鎖可以通過總是以相同的順序獲取鎖來避免。

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

死鎖中，二個線程都在等待對方釋放鎖。

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

4、Slipped Conditions

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

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();
    }
}

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

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

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();
    }
}

5、信號量

Semaphore（信號量）是一個線程同步結(jié)構(gòu)，用于在線程間傳遞信號，以避免出現(xiàn)信號丟失，或者像鎖一樣用于保護一個關(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ā)出一個被存放在 Semaphore內(nèi)部的信號，而Release方法則等待一個信號，當其接收到信號后，標記位 signal 被清空，然后該方法終止。

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

6、阻塞隊列

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

當隊列是空的時，從隊列中獲取元素的操作將會被阻塞。
當隊列是滿時，往隊列里添加元素的操作會被阻塞。

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

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)部，只有隊列的大小等于上限（limit）或者下限（0）時，才調(diào)用notifyAll方法。
如果隊列的大小既不等于上限，也不等于下限，任何線程調(diào)用 enqueue 或者 dequeue 方法時，都不會阻塞，都能夠正常的往隊列中添加或者移除元素。

7、線程池

線程池（Thread Pool）對于限制應用程序中同一時刻運行的線程數(shù)很有用。因為每啟動一個新線程都會有相應的性能開銷，每個線程都需要給棧分配一些內(nèi)存等等。

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

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

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) {
        // 寫日志或者報告異常,
        // 但保持線程池運行.
      }
    }
  }

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

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