設(shè)計(jì)分析
介紹
提供了對(duì)資源占用、釋放，線程的等待、喚醒等接口和具體實(shí)現(xiàn)，可以用在各種需要控制資源爭(zhēng)用的場(chǎng)景中。
| 獨(dú)占資源接口 | 共享資源接口 |
|---|---|
| acquire | acquireShared |
| release | releaseShared |
| tryAcquire | tryAcquireShared |
| tryRelease | tryReleaseShared |
- acquire、acquireShared：定義了資源爭(zhēng)用的邏輯，如果沒(méi)拿到，則等待。
- tryAcquire、tryAcquireShared：實(shí)際執(zhí)行占用資源的操作，如何判定一個(gè)由使用者具體去實(shí)現(xiàn)。
- release、releaseShared：定義了釋放資源的邏輯，釋放之后，通知后續(xù)節(jié)點(diǎn)進(jìn)行爭(zhēng)搶。
- tryRelease、tryReleaseShared：實(shí)際執(zhí)行資源釋放的操作，具體的AQS使用者去實(shí)現(xiàn)。
共享式獲取與獨(dú)占式獲取區(qū)別
共享式獲取與獨(dú)占式獲取最主要的區(qū)別在于同一時(shí)刻能否有多個(gè)線程同時(shí)獲取到同步狀態(tài)。以文件的讀寫(xiě)為例，如果一個(gè)程序在對(duì)文件進(jìn)行讀操作，那么這一時(shí)刻對(duì)于該文件的寫(xiě)操作均被阻塞，而讀操作能夠同時(shí)進(jìn)行。寫(xiě)操作要求對(duì)資源的獨(dú)占式訪問(wèn)，而讀操作可以是共享式訪問(wèn)，兩種不同的訪問(wèn)模式在同一時(shí)刻對(duì)文件或資源的訪問(wèn)情況，左半部分，共享式訪問(wèn)資源時(shí)，其他共享式的訪問(wèn)均被允許，而獨(dú)占式訪問(wèn)被阻塞，右半部分是獨(dú)占式訪問(wèn)資源時(shí)，同一時(shí)刻其他訪問(wèn)均被阻塞。
同步隊(duì)列
同步器依賴內(nèi)部的同步隊(duì)列（一個(gè)FIFO雙向隊(duì)列）來(lái)完成同步狀態(tài)的管理，當(dāng)前線程獲取同步狀態(tài)失敗時(shí)，同步器會(huì)將當(dāng)前線程以及等待狀態(tài)等信息構(gòu)造成為一個(gè)節(jié)點(diǎn)（Node）并將其加入同步隊(duì)列，同時(shí)會(huì)阻塞當(dāng)前線程，當(dāng)同步狀態(tài)釋放時(shí)，會(huì)把首節(jié)點(diǎn)中的線程喚醒，使其再次嘗試獲取同步狀態(tài)。
同步隊(duì)列的基本結(jié)構(gòu)
節(jié)點(diǎn)是構(gòu)成同步隊(duì)列的基礎(chǔ)，同步器擁有首節(jié)點(diǎn)（head）和尾節(jié)點(diǎn)（tail），沒(méi)有成功獲取同步狀態(tài)的線程將會(huì)成為節(jié)點(diǎn)加入該隊(duì)列的尾部。
節(jié)點(diǎn)加入到同步隊(duì)列
同步器包含了兩個(gè)節(jié)點(diǎn)類型的引用，一個(gè)指向頭節(jié)點(diǎn)，而另一個(gè)指向尾節(jié)點(diǎn)。試想一下，當(dāng)一個(gè)線程成功地獲取了同步狀態(tài)（或者鎖），其他線程將無(wú)法獲取到同步狀態(tài)，轉(zhuǎn)而被構(gòu)造成為節(jié)點(diǎn)并加入到同步隊(duì)列中，而這個(gè)加入隊(duì)列的過(guò)程必須要保證線程安全，因此同步器提供了一個(gè)基于CAS的設(shè)置尾節(jié)點(diǎn)的方法：compareAndSetTail(Node expect, Node update)，它需要傳遞當(dāng)前線程“認(rèn)為”的尾節(jié)點(diǎn)和當(dāng)前節(jié)點(diǎn)，只有設(shè)置成功后，當(dāng)前節(jié)點(diǎn)才正式與之前的尾節(jié)點(diǎn)建立關(guān)聯(lián)。
設(shè)置首節(jié)點(diǎn)
設(shè)置首節(jié)點(diǎn)是通過(guò)獲取同步狀態(tài)成功的線程來(lái)完成的，由于只有一個(gè)線程能夠成功獲取到同步狀態(tài)，因此設(shè)置頭節(jié)點(diǎn)的方法并不需要使用CAS來(lái)保證，它只需要將首節(jié)點(diǎn)設(shè)置成為原首節(jié)點(diǎn)的后繼節(jié)點(diǎn)并斷開(kāi)原首節(jié)點(diǎn)的next引用即可。
獨(dú)占式同步狀態(tài)獲取與釋放
前驅(qū)節(jié)點(diǎn)為頭節(jié)點(diǎn)且能夠獲取同步狀態(tài)的判斷條件和線程進(jìn)入等待狀態(tài)是獲取同步狀態(tài)的自旋過(guò)程。當(dāng)同步狀態(tài)獲取成功之后，當(dāng)前線程從acquire(int arg)方法返回，如果對(duì)于鎖這種并發(fā)組件而言，代表著當(dāng)前線程獲取了鎖。
acquire(獨(dú)占式獲取同步狀態(tài))
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
上述代碼主要完成了同步狀態(tài)獲取、節(jié)點(diǎn)構(gòu)造、加入同步隊(duì)列以及在同步隊(duì)列中自旋等待的相關(guān)工作，其主要邏輯是：首先調(diào)用自定義同步器實(shí)現(xiàn)的tryAcquire(int arg)方法，該方法保證線程安全的獲取同步狀態(tài)，如果同步狀態(tài)獲取失敗，則構(gòu)造同步節(jié)點(diǎn)（獨(dú)占式Node. EXCLUSIVE，同一時(shí)刻只能有一個(gè)線程成功獲取同步狀態(tài)）并通過(guò)addWaiter(Node node)方法將該節(jié)點(diǎn)加入到同步隊(duì)列的尾部，最后調(diào)用acquireQueued(Node node, int arg)方法，使得該節(jié)點(diǎn)以“死循環(huán)”的方式獲取同步狀態(tài)。如果獲取不到則阻塞節(jié)點(diǎn)中的線程，而被阻塞線程的喚醒主要依靠前驅(qū)節(jié)點(diǎn)的出隊(duì)或阻塞線程被中斷來(lái)實(shí)現(xiàn)。
private Node addWaiter(Node mode) {
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
Node pred = tail;
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
private Node enq(final Node node) {
for (;;) {
Node t = tail;
if (t == null) { // Must initialize
if (compareAndSetHead(new Node()))
tail = head;
} else {
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
上述代碼通過(guò)使用compareAndSetTail(Node expect, Node update)方法來(lái)確保節(jié)點(diǎn)能夠被線程安全添加。試想一下：如果使用一個(gè)普通的LinkedList來(lái)維護(hù)節(jié)點(diǎn)之間的關(guān)系，那么當(dāng)一個(gè)線程獲取了同步狀態(tài)，而其他多個(gè)線程由于調(diào)用tryAcquire(int arg)方法獲取同步狀態(tài)失敗而并發(fā)地被添加到LinkedList時(shí)，LinkedList將難以保證Node的正確添加，最終的結(jié)果可能是節(jié)點(diǎn)的數(shù)量有偏差，而且順序也是混亂的。
在enq(final Node node)方法中，同步器通過(guò)“死循環(huán)”來(lái)保證節(jié)點(diǎn)的正確添加，在“死循環(huán)”中只有通過(guò)CAS將節(jié)點(diǎn)設(shè)置成為尾節(jié)點(diǎn)之后，當(dāng)前線程才能從該方法返回，否則，當(dāng)前線程不斷地嘗試設(shè)置?？梢钥闯觯琫nq(final Node node)方法將并發(fā)添加節(jié)點(diǎn)的請(qǐng)求通過(guò)CAS變得“串行化”了。
節(jié)點(diǎn)進(jìn)入同步隊(duì)列之后，就進(jìn)入了一個(gè)自旋的過(guò)程，每個(gè)節(jié)點(diǎn)（或者說(shuō)每個(gè)線程）都在自省地觀察，當(dāng)條件滿足，獲取到了同步狀態(tài)，就可以從這個(gè)自旋過(guò)程中退出，否則依舊留在這個(gè)自旋過(guò)程中（并會(huì)阻塞節(jié)點(diǎn)的線程）
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
在acquireQueued (final Node node, int arg)方法中，當(dāng)前線程在“死循環(huán)”中嘗試獲取同步狀態(tài)，而只有前驅(qū)節(jié)點(diǎn)是頭節(jié)點(diǎn)才能夠嘗試獲取同步狀態(tài)，這是為什么？原因有兩個(gè)，如下。
第一，頭節(jié)點(diǎn)是成功獲取到同步狀態(tài)的節(jié)點(diǎn)，而頭節(jié)點(diǎn)的線程釋放了同步狀態(tài)之后，將會(huì)喚醒其后繼節(jié)點(diǎn)，后繼節(jié)點(diǎn)的線程被喚醒后需要檢查自己的前驅(qū)節(jié)點(diǎn)是否是頭節(jié)點(diǎn)。
第二，維護(hù)同步隊(duì)列的FIFO原則。
由于非首節(jié)點(diǎn)線程前驅(qū)節(jié)點(diǎn)出隊(duì)或者被中斷而從等待狀態(tài)返回，隨后檢查自己的前驅(qū)是否是頭節(jié)點(diǎn)，如果是則嘗試獲取同步狀態(tài)?？梢钥吹焦?jié)點(diǎn)和節(jié)點(diǎn)之間在循環(huán)檢查的過(guò)程中基本不相互通信，而是簡(jiǎn)單地判斷自己的前驅(qū)是否為頭節(jié)點(diǎn)，這樣就使得節(jié)點(diǎn)的釋放規(guī)則符合FIFO，并且也便于對(duì)過(guò)早通知的處理（過(guò)早通知是指前驅(qū)節(jié)點(diǎn)不是頭節(jié)點(diǎn)的線程由于中斷而被喚醒）。
release(獨(dú)占式釋放同步狀態(tài))
該方法在釋放了同步狀態(tài)之后，會(huì)喚醒其后繼節(jié)點(diǎn)（進(jìn)而使后繼節(jié)點(diǎn)重新嘗試獲取同步狀態(tài)）。
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
該方法執(zhí)行時(shí)，會(huì)喚醒頭節(jié)點(diǎn)的后繼節(jié)點(diǎn)線程，unparkSuccessor(Node node)方法使用LockSupport（在后面的章節(jié)會(huì)專門(mén)介紹）來(lái)喚醒處于等待狀態(tài)的線程。
分析了獨(dú)占式同步狀態(tài)獲取和釋放過(guò)程后，適當(dāng)做個(gè)總結(jié)：在獲取同步狀態(tài)時(shí)，同步器維護(hù)一個(gè)同步隊(duì)列，獲取狀態(tài)失敗的線程都會(huì)被加入到隊(duì)列中并在隊(duì)列中進(jìn)行自旋；移出隊(duì)列（或停止自旋）的條件是前驅(qū)節(jié)點(diǎn)為頭節(jié)點(diǎn)且成功獲取了同步狀態(tài)。在釋放同步狀態(tài)時(shí)，同步器調(diào)用tryRelease(int arg)方法釋放同步狀態(tài)，然后喚醒頭節(jié)點(diǎn)的后繼節(jié)點(diǎn)。
共享式同步狀態(tài)獲取與釋放
acquireShared(共享式獲取同步狀態(tài))
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}
private void doAcquireShared(int arg) {
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null; // help GC
if (interrupted)
selfInterrupt();
failed = false;
return;
}
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
在acquireShared(int arg)方法中，同步器調(diào)用tryAcquireShared(int arg)方法嘗試獲取同步狀態(tài)，tryAcquireShared(int arg)方法返回值為int類型，當(dāng)返回值大于等于0時(shí)，表示能夠獲取到同步狀態(tài)。因此，在共享式獲取的自旋過(guò)程中，成功獲取到同步狀態(tài)并退出自旋的條件就是tryAcquireShared(int arg)方法返回值大于等于0?？梢钥吹剑赿oAcquireShared(int arg)方法的自旋過(guò)程中，如果當(dāng)前節(jié)點(diǎn)的前驅(qū)為頭節(jié)點(diǎn)時(shí)，嘗試獲取同步狀態(tài)，如果返回值大于等于0，表示該次獲取同步狀態(tài)成功并從自旋過(guò)程中退出。
releaseShared(共享式獲取同步狀態(tài))
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
該方法在釋放同步狀態(tài)之后，將會(huì)喚醒后續(xù)處于等待狀態(tài)的節(jié)點(diǎn)。對(duì)于能夠支持多個(gè)線程同時(shí)訪問(wèn)的并發(fā)組件（比如Semaphore），它和獨(dú)占式主要區(qū)別在于tryReleaseShared(int arg)方法必須確保同步狀態(tài)（或者資源數(shù)）線程安全釋放，一般是通過(guò)循環(huán)和CAS來(lái)保證的，因?yàn)獒尫磐綘顟B(tài)的操作會(huì)同時(shí)來(lái)自多個(gè)線程。
實(shí)現(xiàn)
ReentrantReadWriteLock中的實(shí)現(xiàn)
abstract static class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 6317671515068378041L;
/*
* Read vs write count extraction constants and functions.
* Lock state is logically divided into two unsigned shorts:
* The lower one representing the exclusive (writer) lock hold count,
* and the upper the shared (reader) hold count.
*/
static final int SHARED_SHIFT = 16;
static final int SHARED_UNIT = (1 << SHARED_SHIFT);
static final int MAX_COUNT = (1 << SHARED_SHIFT) - 1;
static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;
/** Returns the number of shared holds represented in count */
static int sharedCount(int c) { return c >>> SHARED_SHIFT; }
/** Returns the number of exclusive holds represented in count */
static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }
/**
* A counter for per-thread read hold counts.
* Maintained as a ThreadLocal; cached in cachedHoldCounter
*/
static final class HoldCounter {
int count = 0;
// Use id, not reference, to avoid garbage retention
final long tid = getThreadId(Thread.currentThread());
}
/**
* ThreadLocal subclass. Easiest to explicitly define for sake
* of deserialization mechanics.
*/
static final class ThreadLocalHoldCounter
extends ThreadLocal<HoldCounter> {
public HoldCounter initialValue() {
return new HoldCounter();
}
}
/**
* The number of reentrant read locks held by current thread.
* Initialized only in constructor and readObject.
* Removed whenever a thread's read hold count drops to 0.
*/
private transient ThreadLocalHoldCounter readHolds;
/**
* The hold count of the last thread to successfully acquire
* readLock. This saves ThreadLocal lookup in the common case
* where the next thread to release is the last one to
* acquire. This is non-volatile since it is just used
* as a heuristic, and would be great for threads to cache.
*
* <p>Can outlive the Thread for which it is caching the read
* hold count, but avoids garbage retention by not retaining a
* reference to the Thread.
*
* <p>Accessed via a benign data race; relies on the memory
* model's final field and out-of-thin-air guarantees.
*/
private transient HoldCounter cachedHoldCounter;
/**
* firstReader is the first thread to have acquired the read lock.
* firstReaderHoldCount is firstReader's hold count.
*
* <p>More precisely, firstReader is the unique thread that last
* changed the shared count from 0 to 1, and has not released the
* read lock since then; null if there is no such thread.
*
* <p>Cannot cause garbage retention unless the thread terminated
* without relinquishing its read locks, since tryReleaseShared
* sets it to null.
*
* <p>Accessed via a benign data race; relies on the memory
* model's out-of-thin-air guarantees for references.
*
* <p>This allows tracking of read holds for uncontended read
* locks to be very cheap.
*/
private transient Thread firstReader = null;
private transient int firstReaderHoldCount;
Sync() {
readHolds = new ThreadLocalHoldCounter();
setState(getState()); // ensures visibility of readHolds
}
/*
* Acquires and releases use the same code for fair and
* nonfair locks, but differ in whether/how they allow barging
* when queues are non-empty.
*/
/**
* Returns true if the current thread, when trying to acquire
* the read lock, and otherwise eligible to do so, should block
* because of policy for overtaking other waiting threads.
*/
abstract boolean readerShouldBlock();
/**
* Returns true if the current thread, when trying to acquire
* the write lock, and otherwise eligible to do so, should block
* because of policy for overtaking other waiting threads.
*/
abstract boolean writerShouldBlock();
/*
* Note that tryRelease and tryAcquire can be called by
* Conditions. So it is possible that their arguments contain
* both read and write holds that are all released during a
* condition wait and re-established in tryAcquire.
*/
protected final boolean tryRelease(int releases) {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
int nextc = getState() - releases;
boolean free = exclusiveCount(nextc) == 0;
if (free)
setExclusiveOwnerThread(null);
setState(nextc);
return free;
}
protected final boolean tryAcquire(int acquires) {
/*
* Walkthrough:
* 1. If read count nonzero or write count nonzero
* and owner is a different thread, fail.
* 2. If count would saturate, fail. (This can only
* happen if count is already nonzero.)
* 3. Otherwise, this thread is eligible for lock if
* it is either a reentrant acquire or
* queue policy allows it. If so, update state
* and set owner.
*/
Thread current = Thread.currentThread();
int c = getState();
int w = exclusiveCount(c);
if (c != 0) {
// (Note: if c != 0 and w == 0 then shared count != 0)
if (w == 0 || current != getExclusiveOwnerThread())
return false;
if (w + exclusiveCount(acquires) > MAX_COUNT)
throw new Error("Maximum lock count exceeded");
// Reentrant acquire
setState(c + acquires);
return true;
}
if (writerShouldBlock() ||
!compareAndSetState(c, c + acquires))
return false;
setExclusiveOwnerThread(current);
return true;
}
protected final boolean tryReleaseShared(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;
}
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;
}
}
private IllegalMonitorStateException unmatchedUnlockException() {
return new IllegalMonitorStateException(
"attempt to unlock read lock, not locked by current thread");
}
protected final int tryAcquireShared(int unused) {
/*
* Walkthrough:
* 1. If write lock held by another thread, fail.
* 2. Otherwise, this thread is eligible for
* lock wrt state, so ask if it should block
* because of queue policy. If not, try
* to grant by CASing state and updating count.
* Note that step does not check for reentrant
* acquires, which is postponed to full version
* to avoid having to check hold count in
* the more typical non-reentrant case.
* 3. If step 2 fails either because thread
* apparently not eligible or CAS fails or count
* saturated, chain to version with full retry loop.
*/
Thread current = Thread.currentThread();
int c = getState();
if (exclusiveCount(c) != 0 &&
getExclusiveOwnerThread() != current)
return -1;
int r = sharedCount(c);
if (!readerShouldBlock() &&
r < MAX_COUNT &&
compareAndSetState(c, c + SHARED_UNIT)) {
if (r == 0) {
firstReader = current;
firstReaderHoldCount = 1;
} else if (firstReader == current) {
firstReaderHoldCount++;
} else {
HoldCounter rh = cachedHoldCounter;
if (rh == null || rh.tid != getThreadId(current))
cachedHoldCounter = rh = readHolds.get();
else if (rh.count == 0)
readHolds.set(rh);
rh.count++;
}
return 1;
}
return fullTryAcquireShared(current);
}
/**
* Full version of acquire for reads, that handles CAS misses
* and reentrant reads not dealt with in tryAcquireShared.
*/
final int fullTryAcquireShared(Thread current) {
/*
* This code is in part redundant with that in
* tryAcquireShared but is simpler overall by not
* complicating tryAcquireShared with interactions between
* retries and lazily reading hold counts.
*/
HoldCounter rh = null;
for (;;) {
int c = getState();
if (exclusiveCount(c) != 0) {
if (getExclusiveOwnerThread() != current)
return -1;
// else we hold the exclusive lock; blocking here
// would cause deadlock.
} else if (readerShouldBlock()) {
// Make sure we're not acquiring read lock reentrantly
if (firstReader == current) {
// assert firstReaderHoldCount > 0;
} else {
if (rh == null) {
rh = cachedHoldCounter;
if (rh == null || rh.tid != getThreadId(current)) {
rh = readHolds.get();
if (rh.count == 0)
readHolds.remove();
}
}
if (rh.count == 0)
return -1;
}
}
if (sharedCount(c) == MAX_COUNT)
throw new Error("Maximum lock count exceeded");
if (compareAndSetState(c, c + SHARED_UNIT)) {
if (sharedCount(c) == 0) {
firstReader = current;
firstReaderHoldCount = 1;
} else if (firstReader == current) {
firstReaderHoldCount++;
} else {
if (rh == null)
rh = cachedHoldCounter;
if (rh == null || rh.tid != getThreadId(current))
rh = readHolds.get();
else if (rh.count == 0)
readHolds.set(rh);
rh.count++;
cachedHoldCounter = rh; // cache for release
}
return 1;
}
}
}
/**
* Performs tryLock for write, enabling barging in both modes.
* This is identical in effect to tryAcquire except for lack
* of calls to writerShouldBlock.
*/
final boolean tryWriteLock() {
Thread current = Thread.currentThread();
int c = getState();
if (c != 0) {
int w = exclusiveCount(c);
if (w == 0 || current != getExclusiveOwnerThread())
return false;
if (w == MAX_COUNT)
throw new Error("Maximum lock count exceeded");
}
if (!compareAndSetState(c, c + 1))
return false;
setExclusiveOwnerThread(current);
return true;
}
/**
* Performs tryLock for read, enabling barging in both modes.
* This is identical in effect to tryAcquireShared except for
* lack of calls to readerShouldBlock.
*/
final boolean tryReadLock() {
Thread current = Thread.currentThread();
for (;;) {
int c = getState();
if (exclusiveCount(c) != 0 &&
getExclusiveOwnerThread() != current)
return false;
int r = sharedCount(c);
if (r == MAX_COUNT)
throw new Error("Maximum lock count exceeded");
if (compareAndSetState(c, c + SHARED_UNIT)) {
if (r == 0) {
firstReader = current;
firstReaderHoldCount = 1;
} else if (firstReader == current) {
firstReaderHoldCount++;
} else {
HoldCounter rh = cachedHoldCounter;
if (rh == null || rh.tid != getThreadId(current))
cachedHoldCounter = rh = readHolds.get();
else if (rh.count == 0)
readHolds.set(rh);
rh.count++;
}
return true;
}
}
}
protected final boolean isHeldExclusively() {
// While we must in general read state before owner,
// we don't need to do so to check if current thread is owner
return getExclusiveOwnerThread() == Thread.currentThread();
}
// Methods relayed to outer class
final ConditionObject newCondition() {
return new ConditionObject();
}
final Thread getOwner() {
// Must read state before owner to ensure memory consistency
return ((exclusiveCount(getState()) == 0) ?
null :
getExclusiveOwnerThread());
}
final int getReadLockCount() {
return sharedCount(getState());
}
final boolean isWriteLocked() {
return exclusiveCount(getState()) != 0;
}
final int getWriteHoldCount() {
return isHeldExclusively() ? exclusiveCount(getState()) : 0;
}
final int getReadHoldCount() {
if (getReadLockCount() == 0)
return 0;
Thread current = Thread.currentThread();
if (firstReader == current)
return firstReaderHoldCount;
HoldCounter rh = cachedHoldCounter;
if (rh != null && rh.tid == getThreadId(current))
return rh.count;
int count = readHolds.get().count;
if (count == 0) readHolds.remove();
return count;
}
/**
* Reconstitutes the instance from a stream (that is, deserializes it).
*/
private void readObject(java.io.ObjectInputStream s)
throws java.io.IOException, ClassNotFoundException {
s.defaultReadObject();
readHolds = new ThreadLocalHoldCounter();
setState(0); // reset to unlocked state
}
final int getCount() { return getState(); }
}
ReentrantLock中的實(shí)現(xiàn)
abstract static class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = -5179523762034025860L;
/**
* Performs {@link Lock#lock}. The main reason for subclassing
* is to allow fast path for nonfair version.
*/
abstract void lock();
/**
* Performs non-fair tryLock. tryAcquire is implemented in
* subclasses, but both need nonfair try for trylock method.
*/
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
protected final boolean tryRelease(int releases) {
int c = getState() - releases;
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
if (c == 0) {
free = true;
setExclusiveOwnerThread(null);
}
setState(c);
return free;
}
protected final boolean isHeldExclusively() {
// While we must in general read state before owner,
// we don't need to do so to check if current thread is owner
return getExclusiveOwnerThread() == Thread.currentThread();
}
final ConditionObject newCondition() {
return new ConditionObject();
}
// Methods relayed from outer class
final Thread getOwner() {
return getState() == 0 ? null : getExclusiveOwnerThread();
}
final int getHoldCount() {
return isHeldExclusively() ? getState() : 0;
}
final boolean isLocked() {
return getState() != 0;
}
/**
* Reconstitutes the instance from a stream (that is, deserializes it).
*/
private void readObject(java.io.ObjectInputStream s)
throws java.io.IOException, ClassNotFoundException {
s.defaultReadObject();
setState(0); // reset to unlocked state
}
}
CountDownLatch中的實(shí)現(xiàn)
private static final class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 4982264981922014374L;
Sync(int count) {
setState(count);
}
int getCount() {
return getState();
}
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
protected boolean tryReleaseShared(int releases) {
// Decrement count; signal when transition to zero
for (;;) {
int c = getState();
if (c == 0)
return false;
int nextc = c-1;
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
}
Semaphore中的實(shí)現(xiàn)
abstract static class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 1192457210091910933L;
Sync(int permits) {
setState(permits);
}
final int getPermits() {
return getState();
}
final int nonfairTryAcquireShared(int acquires) {
for (;;) {
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
protected final boolean tryReleaseShared(int releases) {
for (;;) {
int current = getState();
int next = current + releases;
if (next < current) // overflow
throw new Error("Maximum permit count exceeded");
if (compareAndSetState(current, next))
return true;
}
}
final void reducePermits(int reductions) {
for (;;) {
int current = getState();
int next = current - reductions;
if (next > current) // underflow
throw new Error("Permit count underflow");
if (compareAndSetState(current, next))
return;
}
}
final int drainPermits() {
for (;;) {
int current = getState();
if (current == 0 || compareAndSetState(current, 0))
return current;
}
}
}
圖示
字段圖
exclusiveOwnerThread：當(dāng)前擁有獨(dú)占訪問(wèn)權(quán)限的線程
方法圖
-
管理同步狀態(tài)
-
getState()
獲取當(dāng)前同步狀態(tài)
-
setState(int newState)
設(shè)置當(dāng)前同步狀態(tài)
-
compareAndSetState(int expect, int update)
使用CAS設(shè)置當(dāng)前狀態(tài)，該方法能夠保證設(shè)置的原子性。
-
全圖
