finalvoidlock(){ acquire(1); } // AQS中的方法 队列中当前节点是否有前驱结点 publicfinalbooleanhasQueuedPredecessors(){ // The correctness of this depends on head being initialized // before tail and on head.next being accurate if the current // thread is first in queue. Node t = tail; // Read fields in reverse initialization order Node h = head; Node s; // h != t表示头节点不等于尾结点,如果h==t表示当前队列为空 // h != t && s == null,则说明有一个元素将要作为AQS的第一个节点入队 // h != t && s != null && s.thread != Thread.currentThread() 说明当前队列里的第一个元素不是当前线程 return h != t && ((s = h.next) == null || s.thread != Thread.currentThread()); }
/** * Fair version of tryAcquire. Don't grant access unless * recursive call or no waiters or is first. */ // 会尝试再次通过CAS获取一次锁 protectedfinalbooleantryAcquire(int acquires){ final Thread current = Thread.currentThread(); int c = getState(); // getState 为0,表示没有线程持有锁 if (c == 0) { // 公平策略的体现 // 看当前线程是否有前驱节点(即是否在队列头),如果在队头则尝试更改status是否成功 if (!hasQueuedPredecessors() && compareAndSetState(0, acquires)) { // 设置当前线程为占有线程 setExclusiveOwnerThread(current); returntrue; } } // getState 不为0,表示有线程持有锁,判断是否为当前线程占有 elseif (current == getExclusiveOwnerThread()) { // 将进入次数加一,由于是可重入锁,所以进入一次就加一,出去一次就减一 int nextc = c + acquires; if (nextc < 0) thrownew Error("Maximum lock count exceeded"); setState(nextc); returntrue; } returnfalse; } }
// 获取读锁 protectedfinalinttryAcquireShared(int unused){ Thread current = Thread.currentThread(); int c = getState(); // 写状态不等于0并且锁的持有者不是当前线程 // 已经有其他线程获取到写锁了,此时无法获取读锁 if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current) return -1; // 获取读锁的数量 int r = sharedCount(c); // 尝试获取锁,compareAndSetState多个读线程同时操作时,只有一个线程会成功,如果失败,会执行fullTryAcquireShared方法进行重试 if (!readerShouldBlock() && r < MAX_COUNT && compareAndSetState(c, c + SHARED_UNIT)) { // 读锁数量为0,说明该线程是第一个获取到读锁的 if (r == 0) { firstReader = current; firstReaderHoldCount = 1; // 如果当前线程是第一个获取到读锁的线程,将数量+1 } elseif (firstReader == current) { firstReaderHoldCount++; } else { // cachedHoldCounter记录最后一个获取读锁的线程以及该线程获取读锁的重入数 HoldCounter rh = cachedHoldCounter; if (rh == null || rh.tid != getThreadId(current)) cachedHoldCounter = rh = readHolds.get(); elseif (rh.count == 0) //readHolds记录了当前线程获取读锁的可重入数 readHolds.set(rh); rh.count++; } return1; } return fullTryAcquireShared(current); }
// 释放读锁 protectedfinalbooleantryReleaseShared(int unused){ Thread current = Thread.currentThread(); if (firstReader == current) { // assert firstReaderHoldCount > 0; if (firstReaderHoldCount == 1) firstReader = null; else firstReaderHoldCount--; } else { HoldCounter rh = cachedHoldCounter; if (rh == null || rh.tid != getThreadId(current)) rh = readHolds.get(); int count = rh.count; if (count <= 1) { readHolds.remove(); if (count <= 0) throw unmatchedUnlockException(); } --rh.count; } // 循环读取计数,直到CAS更新成功 for (;;) { int c = getState(); int nextc = c - SHARED_UNIT; if (compareAndSetState(c, nextc)) // Releasing the read lock has no effect on readers, // but it may allow waiting writers to proceed if // both read and write locks are now free. return nextc == 0; } }