1.首先看类的定义:
abstract static class Sync extends AbstractQueuedSynchronizerSync是抽象类,且继承了AbstractQueuedSynchronizer
而AbstractQueuedSynchronizer继承了AbstractOwnableSynchronizer
其中AbstractOwnableSynchronizer定义了一个独占线程,并提供了GET、SET方法。
private transient Thread exclusiveOwnerThread;
AbstractQueuedSynchronizer抽象类提供了一个基于FIFO队列,可以用于构建锁或者其他相关同步装置的基础框架,具体源码另写博客分享。
2.定义了几个常量和变量
static final int SHARED_SHIFT = 16; AQS的state字段拆成两部分了,高16位表示读锁的次数,低16位表示写锁的次数 static final int SHARED_UNIT = (1 << SHARED_SHIFT);每次线程获取读锁成功就会执行state+=SHARED_UNIT操作,不是+1因为高16位表示获取读锁的次数。 static final int MAX_COUNT = (1 << SHARED_SHIFT) - 1;允许读或写获取锁的最大次数,都是65535 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; }
private transient ThreadLocalHoldCounter readHolds;当前线程拥有的读锁的个数
private transient HoldCounter cachedHoldCounter;成功获取读锁的最后一个线程的HoldCounter
private transient Thread firstReader = null;第一个获取到读锁的线程
private transient int firstReaderHoldCount; firstReader线程的HoldCounter
/** * 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.每个线程都绑定一个HoldCounter */ static final class ThreadLocalHoldCounter extends ThreadLocal<HoldCounter> { public HoldCounter initialValue() { return new HoldCounter(); } }
3.构造函数
Sync() { readHolds = new ThreadLocalHoldCounter();初始化readHolds setState(getState()); // ensures visibility of readHolds }
4.tryAcquire方法
1.如果有读线程和写线程且非当前线程,则失败,2.如果计数饱和,则失败,3.否则,这个线程可以拥有锁,队列策略允许或者可重入的锁,更新状态并设置所有者 protected final boolean tryAcquire(int acquires) { 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;//如果c!=0且w==0,说明有读锁,返回false,如果c!=0且w!=0且非当前线程,则返回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;//如果写阻塞或者CAS失败,则返回false setExclusiveOwnerThread(current);//设置独占线程标识 return true; }
5.tryRelease方法
protected final boolean tryRelease(int releases) { if (!isHeldExclusively()) throw new IllegalMonitorStateException();//如果独占线程非当前线程,抛异常 int nextc = getState() - releases; boolean free = exclusiveCount(nextc) == 0; if (free) setExclusiveOwnerThread(null);//释放锁后设置独占线程为null setState(nextc);//设置状态 return free; }
6.tryAcquireShared方法
protected final int tryAcquireShared(int unused) { /* * Walkthrough: * 1. 如果写锁被另一个线程持有,则返回失败 * 2. 否则,继续判断,如果队列策略允许(readerShouldBlock返回false)获取锁且CAS设置state成功,则设置读锁count的值。这一步并没有检查读锁重入的情况,被延迟到fullTryAcquireShared里了,因为大多数情况下不是重入的; * 3. 如果步骤2失败了,或许是队列策略返回false或许是CAS设置失败了等,则执行fullTryAcquireShared */ Thread current = Thread.currentThread(); int c = getState(); if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current) return -1;//如果有线程持有写锁,且非当前线程,则返回-1 int r = sharedCount(c); if (!readerShouldBlock() && r < MAX_COUNT && compareAndSetState(c, c + SHARED_UNIT)) {//如果队列策略允许且读锁不超过最大值且CAS状态成功 if (r == 0) {//如果读锁个数为0,则设置线程和线程持有的锁个数 firstReader = current; firstReaderHoldCount = 1; } else if (firstReader == current) {//否则,如果读锁非0,且为当前线程,则增加线程持有个数 firstReaderHoldCount++; } else {//否则,读锁非0,且非当前线程 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); }
final int fullTryAcquireShared(Thread current) { HoldCounter rh = null; for (;;) { int c = getState(); if (exclusiveCount(c) != 0) { if (getExclusiveOwnerThread() != current) return -1;//如果写锁个数不为空且非当前线程,则返回-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; } } }
7.tryReleaseShared方法
protected final boolean tryReleaseShared(int unused) { Thread current = Thread.currentThread(); if (firstReader == current) { // assert firstReaderHoldCount > 0; if (firstReaderHoldCount == 1) firstReader = null;//如果当前线程拥有一个读锁,则设置firstReader为null else firstReaderHoldCount--;//否则,锁个数减1 } 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; } }
8.tryWriteLock
final boolean tryWriteLock() { Thread current = Thread.currentThread(); int c = getState(); if (c != 0) {//如果c!=0且w==0则表示读锁不为空,返回false,或者写锁不为空且非当前线程,则返回false 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))//CAS失败,返回false return false; setExclusiveOwnerThread(current);//设置线程 return true; }
9.tryReadLock
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; } } }
10.非公平版本
static final class NonfairSync extends Sync { private static final long serialVersionUID = -8159625535654395037L; final boolean writerShouldBlock() {//写线程可以插队 return false; // writers can always barge } final boolean readerShouldBlock() { /* As a heuristic to avoid indefinite writer starvation, * block if the thread that momentarily appears to be head * of queue, if one exists, is a waiting writer. This is * only a probabilistic effect since a new reader will not * block if there is a waiting writer behind other enabled * readers that have not yet drained from the queue. */ return apparentlyFirstQueuedIsExclusive();//为了防止写线程饥饿,如果AQS等待队里的第一个线程是独占的,则读线程阻塞 } }
11.公平版本
/** * Fair version of Sync */ static final class FairSync extends Sync { private static final long serialVersionUID = -2274990926593161451L; final boolean writerShouldBlock() { return hasQueuedPredecessors();//如果AQS有等待的线程,则当前线程阻塞 } final boolean readerShouldBlock() { return hasQueuedPredecessors(); } }