<rt id="bn8ez"></rt>
<label id="bn8ez"></label>

  • <span id="bn8ez"></span>

    <label id="bn8ez"><meter id="bn8ez"></meter></label>

    上善若水
    In general the OO style is to use a lot of little objects with a lot of little methods that give us a lot of plug points for overriding and variation. To do is to be -Nietzsche, To bei is to do -Kant, Do be do be do -Sinatra
    posts - 146,comments - 147,trackbacks - 0
    轉自:http://blog.csdn.net/chen77716/article/details/6641477

    前文(深入JVM鎖機制-synchronized)分析了JVM中的synchronized實現,本文繼續分析JVM中的另一種鎖Lock的實現。與synchronized不同的是,Lock完全用Java寫成,在java這個層面是無關JVM實現的。

    在java.util.concurrent.locks包中有很多Lock的實現類,常用的有ReentrantLock、ReadWriteLock(實現類ReentrantReadWriteLock),其實現都依賴java.util.concurrent.AbstractQueuedSynchronizer類,實現思路都大同小異,因此我們以ReentrantLock作為講解切入點。

    1. ReentrantLock的調用過程

    經過觀察ReentrantLock把所有Lock接口的操作都委派到一個Sync類上,該類繼承了AbstractQueuedSynchronizer:

    1. static abstract class Sync extends AbstractQueuedSynchronizer  

    Sync又有兩個子類:

    1. final static class NonfairSync extends Sync  
    1. final static class FairSync extends Sync  

    顯然是為了支持公平鎖和非公平鎖而定義,默認情況下為非公平鎖。

    先理一下Reentrant.lock()方法的調用過程(默認非公平鎖):

    這些討厭的Template模式導致很難直觀的看到整個調用過程,其實通過上面調用過程及AbstractQueuedSynchronizer的注釋可以發現,AbstractQueuedSynchronizer中抽象了絕大多數Lock的功能,而只把tryAcquire方法延遲到子類中實現。tryAcquire方法的語義在于用具體子類判斷請求線程是否可以獲得鎖,無論成功與否AbstractQueuedSynchronizer都將處理后面的流程。

    2. 鎖實現(加鎖)

    簡單說來,AbstractQueuedSynchronizer會把所有的請求線程構成一個CLH隊列,當一個線程執行完畢(lock.unlock())時會激活自己的后繼節點,但正在執行的線程并不在隊列中,而那些等待執行的線程全部處于阻塞狀態,經過調查線程的顯式阻塞是通過調用LockSupport.park()完成,而LockSupport.park()則調用sun.misc.Unsafe.park()本地方法,再進一步,HotSpot在Linux中中通過調用pthread_mutex_lock函數把線程交給系統內核進行阻塞。

    該隊列如圖:

    與synchronized相同的是,這也是一個虛擬隊列,不存在隊列實例,僅存在節點之間的前后關系。令人疑惑的是為什么采用CLH隊列呢?原生的CLH隊列是用于自旋鎖,但Doug Lea把其改造為阻塞鎖。

    當有線程競爭鎖時,該線程會首先嘗試獲得鎖,這對于那些已經在隊列中排隊的線程來說顯得不公平,這也是非公平鎖的由來,與synchronized實現類似,這樣會極大提高吞吐量。

    如果已經存在Running線程,則新的競爭線程會被追加到隊尾,具體是采用基于CAS的Lock-Free算法,因為線程并發對Tail調用CAS可能會導致其他線程CAS失敗,解決辦法是循環CAS直至成功。AbstractQueuedSynchronizer的實現非常精巧,令人嘆為觀止,不入細節難以完全領會其精髓,下面詳細說明實現過程:

    2.1 Sync.nonfairTryAcquire

    nonfairTryAcquire方法將是lock方法間接調用的第一個方法,每次請求鎖時都會首先調用該方法。

    1. final boolean nonfairTryAcquire(int acquires) {  
    2.     final Thread current = Thread.currentThread();  
    3.     int c = getState();  
    4.     if (c == 0) {  
    5.         if (compareAndSetState(0, acquires)) {  
    6.             setExclusiveOwnerThread(current);  
    7.             return true;  
    8.         }  
    9.     }  
    10.     else if (current == getExclusiveOwnerThread()) {  
    11.         int nextc = c + acquires;  
    12.         if (nextc < 0) // overflow  
    13.             throw new Error("Maximum lock count exceeded");  
    14.         setState(nextc);  
    15.         return true;  
    16.     }  
    17.     return false;  
    18. }  

    該方法會首先判斷當前狀態,如果c==0說明沒有線程正在競爭該鎖,如果不c !=0 說明有線程正擁有了該鎖。

    如果發現c==0,則通過CAS設置該狀態值為acquires,acquires的初始調用值為1,每次線程重入該鎖都會+1,每次unlock都會-1,但為0時釋放鎖。如果CAS設置成功,則可以預計其他任何線程調用CAS都不會再成功,也就認為當前線程得到了該鎖,也作為Running線程,很顯然這個Running線程并未進入等待隊列。

    如果c !=0 但發現自己已經擁有鎖,只是簡單地++acquires,并修改status值,但因為沒有競爭,所以通過setStatus修改,而非CAS,也就是說這段代碼實現了偏向鎖的功能,并且實現的非常漂亮。

    2.2 AbstractQueuedSynchronizer.addWaiter

    addWaiter方法負責把當前無法獲得鎖的線程包裝為一個Node添加到隊尾:

    1. private Node addWaiter(Node mode) {  
    2.     Node node = new Node(Thread.currentThread(), mode);  
    3.     // Try the fast path of enq; backup to full enq on failure  
    4.     Node pred = tail;  
    5.     if (pred != null) {  
    6.         node.prev = pred;  
    7.         if (compareAndSetTail(pred, node)) {  
    8.             pred.next = node;  
    9.             return node;  
    10.         }  
    11.     }  
    12.     enq(node);  
    13.     return node;  
    14. }  

    其中參數mode是獨占鎖還是共享鎖,默認為null,獨占鎖。追加到隊尾的動作分兩步:

    1. 如果當前隊尾已經存在(tail!=null),則使用CAS把當前線程更新為Tail
    2. 如果當前Tail為null或則線程調用CAS設置隊尾失敗,則通過enq方法繼續設置Tail

    下面是enq方法:

    1. private Node enq(final Node node) {  
    2.     for (;;) {  
    3.         Node t = tail;  
    4.         if (t == null) { // Must initialize  
    5.             Node h = new Node(); // Dummy header  
    6.             h.next = node;  
    7.             node.prev = h;  
    8.             if (compareAndSetHead(h)) {  
    9.                 tail = node;  
    10.                 return h;  
    11.             }  
    12.         }  
    13.         else {  
    14.             node.prev = t;  
    15.             if (compareAndSetTail(t, node)) {  
    16.                 t.next = node;  
    17.                 return t;  
    18.             }  
    19.         }  
    20.     }  
    21. }  


    該方法就是循環調用CAS,即使有高并發的場景,無限循環將會最終成功把當前線程追加到隊尾(或設置隊頭)。總而言之,addWaiter的目的就是通過CAS把當前現在追加到隊尾,并返回包裝后的Node實例。

    把線程要包裝為Node對象的主要原因,除了用Node構造供虛擬隊列外,還用Node包裝了各種線程狀態,這些狀態被精心設計為一些數字值:

    • SIGNAL(-1) :線程的后繼線程正/已被阻塞,當該線程release或cancel時要重新這個后繼線程(unpark)
    • CANCELLED(1):因為超時或中斷,該線程已經被取消
    • CONDITION(-2):表明該線程被處于條件隊列,就是因為調用了Condition.await而被阻塞
    • PROPAGATE(-3):傳播共享鎖
    • 0:0代表無狀態

    2.3 AbstractQueuedSynchronizer.acquireQueued

    acquireQueued的主要作用是把已經追加到隊列的線程節點(addWaiter方法返回值)進行阻塞,但阻塞前又通過tryAccquire重試是否能獲得鎖,如果重試成功能則無需阻塞,直接返回

    1. final boolean acquireQueued(final Node node, int arg) {  
    2.     try {  
    3.         boolean interrupted = false;  
    4.         for (;;) {  
    5.             final Node p = node.predecessor();  
    6.             if (p == head && tryAcquire(arg)) {  
    7.                 setHead(node);  
    8.                 p.next = null; // help GC  
    9.                 return interrupted;  
    10.             }  
    11.             if (shouldParkAfterFailedAcquire(p, node) &&  
    12.                 parkAndCheckInterrupt())  
    13.                 interrupted = true;  
    14.         }  
    15.     } catch (RuntimeException ex) {  
    16.         cancelAcquire(node);  
    17.         throw ex;  
    18.     }  
    19. }  


    仔細看看這個方法是個無限循環,感覺如果p == head && tryAcquire(arg)條件不滿足循環將永遠無法結束,當然不會出現死循環,奧秘在于第12行的parkAndCheckInterrupt會把當前線程掛起,從而阻塞住線程的調用棧。

    1. private final boolean parkAndCheckInterrupt() {  
    2.     LockSupport.park(this);  
    3.     return Thread.interrupted();  
    4. }  

    如前面所述,LockSupport.park最終把線程交給系統(Linux)內核進行阻塞。當然也不是馬上把請求不到鎖的線程進行阻塞,還要檢查該線程的狀態,比如如果該線程處于Cancel狀態則沒有必要,具體的檢查在shouldParkAfterFailedAcquire中:

    1.   private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {  
    2.       int ws = pred.waitStatus;  
    3.       if (ws == Node.SIGNAL)  
    4.           /* 
    5.            * This node has already set status asking a release 
    6.            * to signal it, so it can safely park 
    7.            */  
    8.           return true;  
    9.       if (ws > 0) {  
    10.           /* 
    11.            * Predecessor was cancelled. Skip over predecessors and 
    12.            * indicate retry. 
    13.            */  
    14.    do {  
    15. node.prev = pred = pred.prev;  
    16.    } while (pred.waitStatus > 0);  
    17.    pred.next = node;  
    18.       } else {  
    19.           /* 
    20.            * waitStatus must be 0 or PROPAGATE. Indicate that we 
    21.            * need a signal, but don't park yet. Caller will need to 
    22.            * retry to make sure it cannot acquire before parking.  
    23.            */  
    24.           compareAndSetWaitStatus(pred, ws, Node.SIGNAL);  
    25.       }   
    26.       return false;  
    27.   }  

    檢查原則在于:

    • 規則1:如果前繼的節點狀態為SIGNAL,表明當前節點需要unpark,則返回成功,此時acquireQueued方法的第12行(parkAndCheckInterrupt)將導致線程阻塞
    • 規則2:如果前繼節點狀態為CANCELLED(ws>0),說明前置節點已經被放棄,則回溯到一個非取消的前繼節點,返回false,acquireQueued方法的無限循環將遞歸調用該方法,直至規則1返回true,導致線程阻塞
    • 規則3:如果前繼節點狀態為非SIGNAL、非CANCELLED,則設置前繼的狀態為SIGNAL,返回false后進入acquireQueued的無限循環,與規則2同

    總體看來,shouldParkAfterFailedAcquire就是靠前繼節點判斷當前線程是否應該被阻塞,如果前繼節點處于CANCELLED狀態,則順便刪除這些節點重新構造隊列。

    至此,鎖住線程的邏輯已經完成,下面討論解鎖的過程。

    3. 解鎖

    請求鎖不成功的線程會被掛起在acquireQueued方法的第12行,12行以后的代碼必須等線程被解鎖鎖才能執行,假如被阻塞的線程得到解鎖,則執行第13行,即設置interrupted = true,之后又進入無限循環。

    從無限循環的代碼可以看出,并不是得到解鎖的線程一定能獲得鎖,必須在第6行中調用tryAccquire重新競爭,因為鎖是非公平的,有可能被新加入的線程獲得,從而導致剛被喚醒的線程再次被阻塞,這個細節充分體現了“非公平”的精髓。通過之后將要介紹的解鎖機制會看到,第一個被解鎖的線程就是Head,因此p == head的判斷基本都會成功。

    至此可以看到,把tryAcquire方法延遲到子類中實現的做法非常精妙并具有極強的可擴展性,令人嘆為觀止!當然精妙的不是這個Templae設計模式,而是Doug Lea對鎖結構的精心布局。

    解鎖代碼相對簡單,主要體現在AbstractQueuedSynchronizer.release和Sync.tryRelease方法中:

    class AbstractQueuedSynchronizer

    1. public final boolean release(int arg) {  
    2.     if (tryRelease(arg)) {  
    3.         Node h = head;  
    4.         if (h != null && h.waitStatus != 0)  
    5.             unparkSuccessor(h);  
    6.         return true;  
    7.     }  
    8.     return false;  
    9. }  

    class Sync

    1. protected final boolean tryRelease(int releases) {  
    2.     int c = getState() - releases;  
    3.     if (Thread.currentThread() != getExclusiveOwnerThread())  
    4.         throw new IllegalMonitorStateException();  
    5.     boolean free = false;  
    6.     if (c == 0) {  
    7.         free = true;  
    8.         setExclusiveOwnerThread(null);  
    9.     }  
    10.     setState(c);  
    11.     return free;  
    12. }  


    tryRelease與tryAcquire語義相同,把如何釋放的邏輯延遲到子類中。tryRelease語義很明確:如果線程多次鎖定,則進行多次釋放,直至status==0則真正釋放鎖,所謂釋放鎖即設置status為0,因為無競爭所以沒有使用CAS。

    release的語義在于:如果可以釋放鎖,則喚醒隊列第一個線程(Head),具體喚醒代碼如下:

    1. private void unparkSuccessor(Node node) {  
    2.     /* 
    3.      * If status is negative (i.e., possibly needing signal) try 
    4.      * to clear in anticipation of signalling. It is OK if this 
    5.      * fails or if status is changed by waiting thread. 
    6.      */  
    7.     int ws = node.waitStatus;  
    8.     if (ws < 0)  
    9.         compareAndSetWaitStatus(node, ws, 0);   
    10.   
    11.     /* 
    12.      * Thread to unpark is held in successor, which is normally 
    13.      * just the next node.  But if cancelled or apparently null, 
    14.      * traverse backwards from tail to find the actual 
    15.      * non-cancelled successor. 
    16.      */  
    17.     Node s = node.next;  
    18.     if (s == null || s.waitStatus > 0) {  
    19.         s = null;  
    20.         for (Node t = tail; t != null && t != node; t = t.prev)  
    21.             if (t.waitStatus <= 0)  
    22.                 s = t;  
    23.     }  
    24.     if (s != null)  
    25.         LockSupport.unpark(s.thread);  
    26. }  


    這段代碼的意思在于找出第一個可以unpark的線程,一般說來head.next == head,Head就是第一個線程,但Head.next可能被取消或被置為null,因此比較穩妥的辦法是從后往前找第一個可用線程。貌似回溯會導致性能降低,其實這個發生的幾率很小,所以不會有性能影響。之后便是通知系統內核繼續該線程,在Linux下是通過pthread_mutex_unlock完成。之后,被解鎖的線程進入上面所說的重新競爭狀態。

    4. Lock VS Synchronized

    AbstractQueuedSynchronizer通過構造一個基于阻塞的CLH隊列容納所有的阻塞線程,而對該隊列的操作均通過Lock-Free(CAS)操作,但對已經獲得鎖的線程而言,ReentrantLock實現了偏向鎖的功能。

    synchronized的底層也是一個基于CAS操作的等待隊列,但JVM實現的更精細,把等待隊列分為ContentionList和EntryList,目的是為了降低線程的出列速度;當然也實現了偏向鎖,從數據結構來說二者設計沒有本質區別。但synchronized還實現了自旋鎖,并針對不同的系統和硬件體系進行了優化,而Lock則完全依靠系統阻塞掛起等待線程。

    當然Lock比synchronized更適合在應用層擴展,可以繼承AbstractQueuedSynchronizer定義各種實現,比如實現讀寫鎖(ReadWriteLock),公平或不公平鎖;同時,Lock對應的Condition也比wait/notify要方便的多、靈活的多。

    posted on 2014-07-22 21:46 DLevin 閱讀(441) 評論(0)  編輯  收藏 所屬分類: MultiThreading
    主站蜘蛛池模板: 黄页网址大全免费观看12网站| 亚洲一区二区观看播放| 成年免费a级毛片| 亚洲国产成人a精品不卡在线| 在线91精品亚洲网站精品成人| 日韩高清在线免费观看| 中文无码亚洲精品字幕| 日韩一区二区三区免费体验| 亚洲精品国产日韩| 国产精品成人无码免费| 羞羞视频在线观看免费| 国产亚洲综合色就色| 免费国产黄网站在线观看视频| 亚洲人成电影在线天堂| 99国产精品永久免费视频| 亚洲中文字幕无码中文| 好大好深好猛好爽视频免费| 国产亚洲美女精品久久| 亚洲人成网站观看在线播放| 国产精品九九久久免费视频| 亚洲精品国产精品乱码视色| www视频在线观看免费| 亚洲一区AV无码少妇电影| 五月婷婷亚洲综合| 两个人看的www高清免费观看| 亚洲AV永久精品爱情岛论坛| 在线观看www日本免费网站| 亚洲日韩中文字幕无码一区| 亚洲精品网站在线观看不卡无广告| 免费人成在线观看视频高潮| 亚洲成a人片在线观看中文!!!| 日本人的色道www免费一区| 国产日韩在线视频免费播放| 亚洲黄色在线播放| 国产又大又黑又粗免费视频 | 国产精品免费无遮挡无码永久视频| 亚洲视频中文字幕在线| 国产人妖ts在线观看免费视频| 任你躁在线精品免费| 亚洲一区二区三区在线观看蜜桃| 浮力影院第一页小视频国产在线观看免费 |