詳解鎖原理，synchronized、volatile+cas底層實現

時間 2020-07-22

標籤解鎖原理 synchronized volatile+cas volatile cas 底層實現欄目 Java 简体版

原文原文鏈接

隨着多進程多線程的出現，對共享資源(設備，數據等)的競爭每每會致使資源的使用表現爲隨機無序
例如：一個線程想在控制檯輸出"I am fine"，剛寫到"I am"，就被另外一線程搶佔控制檯輸出"naughty"，致使結果是"I am naughty"；對於資源的被搶佔使用，咱們能怎麼辦呢？固然不是涼拌，可以使用鎖進行同步管理，使得資源在加鎖期間，其餘線程不可搶佔使用

1 鎖的分類

悲觀鎖html
- 悲觀鎖，每次去請求數據的時候，都認爲數據會被搶佔更新(悲觀的想法)；因此每次操做數據時都要先加上鎖，其餘線程修改數據時就要等待獲取鎖。適用於寫多讀少的場景，synchronized就是一種悲觀鎖
樂觀鎖java
- 在請求數據時，以爲無人搶佔修改。等真正更新數據時，才判斷此期間別人有沒有修改過(預先讀出一個版本號或者更新時間戳，更新時判斷是否變化，沒變則期間無人修改)；和悲觀鎖不一樣的是，期間數據容許其餘線程修改
自旋鎖node
- 一句話，魔力轉轉圈。當嘗試給資源加鎖卻被其餘線程先鎖定時，不是阻塞等待而是循環再次加鎖
- 在鎖常被短暫持有的場景下，線程阻塞掛起致使CPU上下文頻繁切換，這可用自旋鎖解決；但自旋期間它佔用CPU空轉，所以不適用長時間持有鎖的場景

2 synchronized底層原理

代碼使用synchronized加鎖，在編譯以後的字節碼是怎樣的呢

public class Test {
    public static void main(String[] args){
        synchronized(Test.class){
            System.out.println("hello");
        }
    }
}

截取部分字節碼，以下git

4: monitorenter
    5: getstatic    #9    // Field java/lang/System.out:Ljava/io/PrintStream; 
    8: ldc           #15   // String hello
    10: invokevirtual #17  // Method java/io/PrintStream.println:(Ljava/lang/String;)V
    13: aload_1
    14: monitorexit

字節碼出現了4: monitorenter和14: monitorexit兩個指令；字面理解就是監視進入，監視退出。能夠理解爲代碼塊執行前的加鎖，和退出同步時的解鎖github

那monitorenter和monitorexit，又揹着咱們幹了啥呢？
執行monitorenter指令時，線程會爲鎖對象關聯一個ObjectMonitor對象

objectMonitor.cpp
  ObjectMonitor() {
    _header       = NULL;
    _count        = 0;   \\用來記錄該線程獲取鎖的次數
    _waiters      = 0,
    _recursions   = 0;    \\鎖的重入次數
    _object       = NULL;
    _owner        = NULL;  \\當前持有ObjectMonitor的線程
    _WaitSet      = NULL;  \\wait()方法調用後的線程等待隊列
    _WaitSetLock  = 0 ;
    _Responsible  = NULL ;
    _succ         = NULL ;
    _cxq          = NULL ; \\阻塞等待隊列
    FreeNext      = NULL ;
    _EntryList    = NULL ; \\synchronized 進來線程的排隊隊列
    _SpinFreq     = 0 ;
    _SpinClock    = 0 ;  \\自旋計算
    OwnerIsThread = 0 ;
  }

每一個線程都有兩個ObjectMonitor對象列表，分別爲free和used列表，若是當前free列表爲空，線程將向全局global list請求分配ObjectMonitor
ObjectMonitor的owner、WaitSet、Cxq、EntryList這幾個屬性比較關鍵。WaitSet、Cxq、EntryList的隊列元素是包裝線程後的對象-ObjectWaiter；而獲取owner的線程，既爲得到鎖的線程
monitorenter對應的執行方法

void ATTR ObjectMonitor::enter(TRAPS)  {
    ...
    //獲取鎖：cmpxchg_ptr原子操做，嘗試將_owner替換爲本身，並返回舊值
    cur = Atomic::cmpxchg_ptr (Self, &_owner, NULL) ;
    ...
    // 重複獲取鎖，次數加1，返回
    if (cur == Self) {
        _recursions ++ ;
        return ;
    }
    //首次獲取鎖狀況處理
    if (Self->is_lock_owned ((address)cur)) {
        assert (_recursions == 0, "internal state error");
        _recursions = 1 ;
        _owner = Self ;
        OwnerIsThread = 1 ;
        return ;
    }
    ...
    //嘗試自旋獲取鎖
    if (Knob_SpinEarly && TrySpin (Self) > 0) {
    ...

monitorexit對應的執行方法void ATTR ObjectMonitor::exit(TRAPS)...代碼太長，就不貼了。主要是recursions減一、count減小1或者若是線程再也不持有owner(非重入加鎖)則設置owner爲null，退鎖的持有狀態，並喚醒Cxq隊列的線程

總結緩存

線程遇到synchronized同步時，先會進入EntryList隊列中，而後嘗試把owner變量設置爲當前線程，同時monitor中的計數器count加1，即得到對象鎖。不然經過嘗試自旋必定次數加鎖，失敗則進入Cxq隊列阻塞等待

線程執行完畢將釋放持有的owner，owner變量恢復爲null，count自減1，以便其餘線程進入獲取鎖

synchronized修飾方法原理也是相似的。只不過沒用monitor指令，而是使用ACC_SYNCHRONIZED標識方法的同步

public synchronized void lock(){
        System.out.println("world");
    }
....
  public synchronized void lock();
    descriptor: ()V
    flags: (0x0029) ACC_PUBLIC, ACC_SYNCHRONIZED
    Code:
      stack=2, locals=0, args_size=0
         0: getstatic     #20                 // Field java/lang/System.out:Ljava/io/PrintStream;
         3: ldc           #26                 // String world
         5: invokevirtual #28                 // Method java/io/PrintStream.println:(Ljava/lang/String;)V

synchronized是可重入，非公平鎖，由於entryList的線程會先自旋嘗試加鎖，而不是加入cxq排隊等待，不公平

3 Object的wait和notify方法原理

wait，notify必須是持有當前對象鎖Monitor的線程才能調用 (對象鎖代指ObjectMonitor/Monitor，鎖對象代指Object)
上面有說到，當在sychronized中鎖對象Object調用wait時會加入waitSet隊列，WaitSet的元素對象就是ObjectWaiter

class ObjectWaiter : public StackObj {
 public:
  enum TStates { TS_UNDEF, TS_READY, TS_RUN, TS_WAIT, TS_ENTER, TS_CXQ } ;
  enum Sorted  { PREPEND, APPEND, SORTED } ;
  ObjectWaiter * volatile _next;
  ObjectWaiter * volatile _prev;
  Thread*       _thread;
  ParkEvent *   _event;
  volatile int  _notified ;
  volatile TStates TState ;
  Sorted        _Sorted ;           // List placement disposition
  bool          _active ;           // Contention monitoring is enabled
 public:
  ObjectWaiter(Thread* thread);
  void wait_reenter_begin(ObjectMonitor *mon);
  void wait_reenter_end(ObjectMonitor *mon);
};

調用對象鎖的wait()方法時，線程會被封裝成ObjectWaiter，最後使用park方法掛起安全

//objectMonitor.cpp
void ObjectMonitor::wait(jlong millis, bool interruptible, TRAPS){
    ...
    //線程封裝成 ObjectWaiter對象
    ObjectWaiter node(Self);
    node.TState = ObjectWaiter::TS_WAIT ;
    ...
    //一系列判斷操做，當線程確實加入WaitSet時，則使用park方法掛起
    if (node._notified == 0) {
        if (millis <= 0) {
            Self->_ParkEvent->park () ;
        } else {
            ret = Self->_ParkEvent->park (millis) ;
        }
    }

而當對象鎖使用notify()時多線程

若是waitSet爲空，則直接返回
waitSet不爲空從waitSet獲取一個ObjectWaiter，而後根據不一樣的Policy加入到EntryList或經過Atomic::cmpxchg_ptr指令自旋操做加入cxq隊列或者直接unpark喚醒

void ObjectMonitor::notify(TRAPS){
    CHECK_OWNER();
    //waitSet爲空，則直接返回
    if (_WaitSet == NULL) {
        TEVENT (Empty-Notify) ;
        return ;
    }
    ...
    //經過DequeueWaiter獲取_WaitSet列表中的第一個ObjectWaiter
    Thread::SpinAcquire (&_WaitSetLock, "WaitSet - notify") ;
    ObjectWaiter * iterator = DequeueWaiter() ;
    if (iterator != NULL) {
    ....
    if (Policy == 2) {      // prepend to cxq
         // prepend to cxq
         if (List == NULL) {
             iterator->_next = iterator->_prev = NULL ;
             _EntryList = iterator ;
         } else {
            iterator->TState = ObjectWaiter::TS_CXQ ;
            for (;;) {
                ObjectWaiter * Front = _cxq ;
                iterator->_next = Front ;
                if (Atomic::cmpxchg_ptr (iterator, &_cxq, Front) == Front) {
                    break ;
                }
            }
         }
     }

Object的notifyAll方法則對應voidObjectMonitor::notifyAll(TRAPS)，流程和notify相似。不過會經過for循環取出WaitSet的ObjectWaiter節點，再依次喚醒全部線程

4 jvm對synchronized的優化

先介紹下32位JVM下JAVA對象頭的結構

偏向鎖併發
- 未加鎖的時候，鎖標誌爲01，包含哈希值、年齡分代和偏向鎖標誌位(0)
- 施加偏向鎖時，哈希值和一部分無用內存會轉化爲鎖主人的線程信息，以及加鎖時的時間戳epoch，此時鎖標誌位沒變，偏向鎖標誌改成1
- 加鎖時先判斷當前線程id是否與MarkWord的線程id是否一致，一致則執行同步代碼；不一致則檢查偏向標誌是否偏向，未偏向則使用CAS加鎖；未偏向CAS加鎖失敗和存在偏向鎖會致使偏向鎖膨脹爲輕量級鎖，或者從新偏向
- 偏向鎖只有遇到其餘線程競爭偏向鎖時，持有偏向鎖的線程纔會釋放鎖，線程不會主動去釋放偏向鎖
輕量級鎖app
- 當發生多個線程競爭時，偏向鎖會變爲輕量級鎖，鎖標誌位爲00
- 得到鎖的線程會先將偏向鎖撤銷(在安全點)，並在棧楨中建立鎖記錄LockRecord，對象的MarkWord被複制到剛建立的LockRecord，而後CAS嘗試將記錄LockRecord的owner指向鎖對象，再將鎖對象的MarkWord指向鎖，加鎖成功
- 若是CAS加鎖失敗，線程會自旋必定次數加鎖，再失敗則升級爲重量級鎖

重量級鎖
- 重量級鎖就是上面介紹到synchronized使用監視器Monitor實現的鎖機制
- 競爭線程激烈，鎖則繼續膨脹，變爲重量級鎖，也是互斥鎖，鎖標誌位爲10，MarkWord其他內容被替換爲一個指向對象鎖Monitor的指針
自旋鎖
- 減小沒必要要的CPU上下文切換；在輕量級鎖升級爲重量級鎖時，就使用了自旋加鎖的方式
鎖粗化
- 屢次加鎖操做在JVM內部也是種消耗，若是多個加鎖能夠合併爲一個鎖，就可減小沒必要要的開銷

Test.class
//編譯器會考慮將兩次加鎖合併
public void test(){
    synchronized(this){
        System.out.println("hello");   
    }
    synchronized(this){
        System.out.println("world");   
    }
}

鎖消除
- 刪除沒必要要的加鎖操做，若是變量是獨屬一個線程的棧變量，加不加鎖都是安全的，編譯器會嘗試消除鎖
- 開啓鎖消除須要在JVM參數上設置-server -XX:+DoEscapeAnalysis -XX:+EliminateLocks

//StringBuffer的append操做會加上synchronized，
//可是變量buf不加鎖也安全的，編譯器會把鎖消除
public void test() {
    StringBuffer buf = new StringBuffer();
    buf.append("hello").append("world");
}

其餘鎖優化方法
- 分段鎖，分段鎖也並不是一種實際的鎖，而是一種思想；ConcurrentHashMap是學習分段鎖的最好實踐。主要是將大對象拆成小對象，而後對大對象的加鎖操做變成對小對象加鎖，增長了並行度

5 CAS的底層原理

在volatile int i = 0; i++中，volatile類型的讀寫是原子同步的，可是i++卻不能保證同步性，咱們該怎麼呢？
可使用synchronized加鎖；還有就是用CAS(比較並交換)，使用樂觀鎖的思想同步，先判斷共享變量是否改變，沒有則更新。下面看看不一樣步版本的CAS

int expectedValue = 1;
public boolean compareAndSet(int newValue) {
    if(expectedValue == 1){
        expectedValue = newValue;
        return ture;
    }
    return false;
}

在jdk是有提供同步版的CAS解決方案，其中使用了UnSafe.java的底層方法

//UnSafe.java
    @HotSpotIntrinsicCandidate
    public final native boolean compareAndSetInt(Object o, long offset, int expected, int x) ..
    @HotSpotIntrinsicCandidate
    public final native int compareAndExchangeInt(Object o, long offset, int expected, int x)...

咱們再來看看本地方法，Unsafe.cpp中的compareAndSwapInt

//unsafe.cpp
UNSAFE_ENTRY(jboolean, Unsafe_CompareAndSwapInt(JNIEnv *env, jobject unsafe, jobject obj, jlong offset, jint e, jint x))
  UnsafeWrapper("Unsafe_CompareAndSwapInt");
  oop p = JNIHandles::resolve(obj);
  jint* addr = (jint *) index_oop_from_field_offset_long(p, offset);
  return (jint)(Atomic::cmpxchg(x, addr, e)) == e;
UNSAFE_END

在Linux的x86，Atomic::cmpxchg方法的實現以下

/**
    1 __asm__表示彙編的開始；
    2 volatile表示禁止編譯器優化；//禁止指令重排
    3 LOCK_IF_MP是個內聯函數，
      根據當前系統是否爲多核處理器，
      決定是否爲cmpxchg指令添加lock前綴 //內存屏障
*/
inline jint Atomic::cmpxchg (jint exchange_value, volatile jint* dest, jint compare_value) {
  int mp = os::is_MP();
  __asm__ volatile (LOCK_IF_MP(%4) "cmpxchgl %1,(%3)"
                    : "=a" (exchange_value)
                    : "r" (exchange_value), "a" (compare_value), "r" (dest), "r" (mp)
                    : "cc", "memory");
  return exchange_value;
}

到這一步，能夠總結到：jdk提供的CAS機制，在彙編層級，會禁止變量兩側的指令優化，而後使用cmpxchg指令比較並更新變量值(原子性)，若是是多核則使用lock鎖定(緩存鎖、MESI)

6 CAS同步操做的問題

ABA問題
- 線程X準備將變量的值從A改成B，然而這期間線程Y將變量的值從A改成C，而後再改成A；最後線程X檢測變量值是A，並置換爲B。但實際上，A已經再也不是原來的A了
- 解決方法，是把變量定爲惟一類型。值能夠加上版本號，或者時間戳。如加上版本號，線程Y的修改變爲A1->B2->A3，此時線程X再更新則能夠判斷出A1不等於A3
只能保證一個共享變量的原子操做
- 只保證一個共享變量的原子操做，對多個共享變量同步時，循環CAS是沒法保證操做的原子

7 基於volatile + CAS 實現同步鎖的原理

CAS只能同步一個變量的修改，咱們又應該如何用它來鎖住代碼塊呢？
先說說實現鎖的要素
- 1 同步代碼塊同一時刻只能有一個線程能執行
- 2 加鎖操做要happens-before同步代碼塊裏的操做，而代碼塊裏的操做要happens-before解鎖操做
- 3 同步代碼塊結束後相對其餘線程其修改的變量是可見的 (內存可見性)
要素1：能夠利用CAS的原子性來實現，任意時刻只有一個線程能成功操做變量
- 先設想CAS操做的共享變量是一個關聯代碼塊的同步狀態變量，同步開始以前先CAS更新狀態變量爲加鎖狀態，同步結束以後，再CAS狀態變量爲無鎖狀態
- 若是期間有第二個線程來加鎖，則會發現狀態變量爲加鎖狀態，則放棄執行同步代碼塊
要素2：使用volatile修飾狀態變量，禁止指令重排
- volatile保證同步代碼裏的操做happens-before解鎖操做，而加鎖操做happens-before代碼塊裏的操做
要素3：仍是用volatile，volatile變量寫指令先後會插入內存屏障
- volatile修飾的狀態變量被CAS爲無鎖狀態前，同步代碼塊的髒數據就會被更新，被各個線程可見

//僞代碼
volatile state = 0 ;   // 0-無鎖 1-加鎖；volatile禁止指令重排，加入內存屏障
...
if(cas(state, 0 , 1)){ // 1 加鎖成功，只有一個線程能成功加鎖
    ...                // 2 同步代碼塊
    cas(state, 1, 0);  // 3 解鎖時2的操做具備可見性
}

8 LockSupport瞭解一下

LockSupport是基於Unsafe類，由JDK提供的線程操做工具類，主要做用就是掛起線程，喚醒線程。Unsafe.park，unpark操做時，會調用當前線程的變量parker代理執行。Parker代碼

JavaThread* thread=JavaThread::thread_from_jni_environment(env);
...
thread->parker()->park(isAbsolute != 0, time);

class PlatformParker : public CHeapObj {
  protected:
    //互斥變量類型
    pthread_mutex_t _mutex [1] ; 
   //條件變量類型
    pthread_cond_t  _cond  [1] ;
    ...
}

class Parker : public os::PlatformParker {  
private:  
  volatile int _counter ;  
  ...  
public:  
  void park(bool isAbsolute, jlong time);  
  void unpark();  
  ...  
}

在Linux系統下，用的POSIX線程庫pthread中的mutex(互斥量)，condition來實現線程的掛起、喚醒
注意點：當park時，counter變量被設置爲0，當unpark時，這個變量被設置爲1
unpark和park執行順序不一樣時，counter和cond的狀態變化以下
- 先park後unpark; park：counter值不變，但會設置一個cond; unpark：counter先加1，檢查cond存在，counter減爲0
- 先unpark後park；park：counter變爲1，但不設置cond；unpark：counter減爲0(線程不會由於park掛起)
- 先屢次unpark；counter也只設置爲爲1

9 LockSupport.park和Object.wait區別

兩種方式都有具備掛起的線程的能力
線程在Object.wait以後必須等到Object.notify才能喚醒
LockSupport能夠先unpark線程，等線程執行LockSupport.park是不會掛起的，能夠繼續執行
須要注意的是就算線程屢次unpark；也只能讓線程第一次park是不會掛起

10 AbstractQueuedSynchronizer(AQS)

AQS其實就是基於volatile+cas實現的鎖模板；若是須要線程阻塞等待，喚醒機制，則使用LockSupport掛起、喚醒線程

//AbstractQueuedSynchronizer.java
public class AbstractQueuedSynchronizer{
    //線程節點
    static final class Node {
        ...
        volatile Node prev;
        volatile Node next;
        volatile Thread thread;
        ...
    }    
    ....
    //head 等待隊列頭尾節點
    private transient volatile Node head;
    private transient volatile Node tail;
    // The synchronization state. 同步狀態
    private volatile int state;  
    ...
    //提供CAS操做，狀態具體的修改由子類實現
    protected final boolean compareAndSetState(int expect, int update) {
        return STATE.compareAndSet(this, expect, update);
    }
}

AQS內部維護一個同步隊列，元素就是包裝了線程的Node
同步隊列中首節點是獲取到鎖的節點，它在釋放鎖的時會喚醒後繼節點，後繼節點獲取到鎖的時候，會把本身設爲首節點

public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
}

線程會先嚐試獲取鎖，失敗則封裝成Node，CAS加入同步隊列的尾部。在加入同步隊列的尾部時，會判斷前驅節點是不是head結點，並嘗試加鎖(可能前驅節點恰好釋放鎖)，不然線程進入阻塞等待

在AQS還存一個ConditionObject的內部類，它的使用機制和Object.wait、notify相似

//AbstractQueuedSynchronizer.java
public class ConditionObject implements Condition, java.io.Serializable {
    //條件隊列;Node 複用了AQS中定義的Node
    private transient Node firstWaiter;
    private transient Node lastWaiter;
    ...

每一個Condition對象內部包含一個Node元素的FIFO條件隊列
當一個線程調用Condition.await()方法，那麼該線程將會釋放鎖、構造Node加入條件隊列並進入等待狀態

//相似Object.wait
public final void await() throws InterruptedException{
    ...
    Node node = addConditionWaiter(); //構造Node,加入條件隊列
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    while (!isOnSyncQueue(node)) {
        //掛起線程
        LockSupport.park(this);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    //notify喚醒線程後，加入同步隊列繼續競爭鎖
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;

調用Condition.signal時，獲取條件隊列的首節點，將其移動到同步隊列而且利用LockSupport喚醒節點中的線程。隨後繼續執行wait掛起前的狀態，調用acquireQueued(node, savedState)競爭同步狀態

//相似Object.notify
    private void doSignal(Node first) {
        do {
            if ( (firstWaiter = first.nextWaiter) == null)
                lastWaiter = null;
            first.nextWaiter = null;
        } while (!transferForSignal(first) &&
                 (first = firstWaiter) != null);
    }

volatile+cas機制保證了代碼的同步性和可見性，而AQS封裝了線程阻塞等待掛起，解鎖喚醒其餘線程的邏輯。AQS子類只需根據狀態變量，判斷是否可獲取鎖，是否釋放鎖成功便可
繼承AQS須要選性重寫如下幾個接口

protected boolean tryAcquire(int arg);//嘗試獨佔性加鎖
protected boolean tryRelease(int arg);//對應tryAcquire釋放鎖
protected int tryAcquireShared(int arg);//嘗試共享性加鎖
protected boolean tryReleaseShared(int arg);//對應tryAcquireShared釋放鎖
protected boolean isHeldExclusively();//該線程是否正在獨佔資源，只有用到condition才須要取實現它

11 ReentrantLock的原理

ReentrantLock實現了Lock接口，並使用內部類Sync(Sync繼承AbstractQueuedSynchronizer)來實現同步操做
ReentrantLock內部類Sync

abstract static class Sync extends AbstractQueuedSynchronizer{
    .... 
    final boolean nonfairTryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {
                //直接CAS狀態加鎖，非公平操做
                if (compareAndSetState(0, acquires)) { 
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
    ...
    //重寫了tryRelease
    protected final boolean tryRelease(int releases) {
        c = state - releases; //改變同步狀態
        ...
        //修改volatile 修飾的狀態變量
        setState(c); 
        return free;
    }
}

Sync的子類NonfairSync和FairSync都重寫了tryAcquire方法
其中NonfairSync的tryAcquire調用父類的nonfairTryAcquire方法, FairSync則本身重寫tryAcquire的邏輯。其中調用hasQueuedPredecessors()判斷是否有排隊Node，存在則返回false（false會致使當前線程排隊等待鎖）

static final class NonfairSync extends Sync {
        protected final boolean tryAcquire(int acquires) {
            return nonfairTryAcquire(acquires);
        }
    }
    ....
    static final class FairSync extends Sync {
        protected final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {
                if (!hasQueuedPredecessors() &&   
                    compareAndSetState(0, acquires)) {
                    setExclusiveOwnerThread(current);
                    return true;
                }
            }
    ....

12 AQS排他鎖的實例demo

public class TwinsLock implements Lock {

    private final Sync sync = new Sync(2);
    @Override
    public void lockInterruptibly() throws InterruptedException {  throw new RuntimeException(""); }
    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {throw new RuntimeException("");}
    @Override
    public Condition newCondition() {  return sync.newCondition(); }
    @Override
    public void lock() {  sync.acquireShared(1); }
    @Override
    public void unlock() {  sync.releaseShared(1); } }
    @Override
    public boolean tryLock() { return sync.tryAcquireShared(1) > -1;  }
}

再來看看Sync的代碼

class Sync extends AbstractQueuedSynchronizer {
        Sync(int count) {
            if (count <= 0) {
                throw new IllegalArgumentException("count must large than zero");
            }
            setState(count);
        }
        @Override
        public int tryAcquireShared(int reduceCount) {
            for (; ; ) {
                int current = getState(); 
                int newCount = current - reduceCount;
                if (newCount < 0 || compareAndSetState(current, newCount)) {
                    return newCount;
                }
            }
        }
        @Override
        public boolean tryReleaseShared(int returnCount) {
            for (; ; ) {
                int current = getState();
                int newCount = current + returnCount;
                if (compareAndSetState(current, newCount)) {
                    return true;
                }
            }
        }
        public Condition newCondition() {
            return new AbstractQueuedSynchronizer.ConditionObject();
        }
    }