Java并发编程之ReentrantReadWriteLock

ReentrantReadWriteLock 定义

Reentrant有可重入之意，从字面上看，ReentrantReadWriteLock 是一个可重入的读写锁。

ReentrantReadWriteLock 的特点

读写锁都可重入, 线程可同时具有读写锁线程同时获取读写锁时, 必须先获取 writeLock, 再获取 readLock (也就是锁的降级), 反过来的直接导致死锁(这个问题下面会重点分析)ReentrantReadWriteLock支持公平与非公平机制, 主要依据是 AQS 类中的Sync Queue 里面是否有节点 或 Sync Queue 里面的 head.next 是否是获取 writeLock 的线程节点; 公平模式就会依据获取的先后顺序在 SyncQueue 里面排队获取读写锁互斥获取 readLock 的过程中, 若此时有线程已获取写锁 或 AQS 的 Sync Queue 里面有 获取 writeLock 的线程, 则一定会等待获取writeLock成功并释放或放弃获取 后才能获取(PS: 这里有个例外, 在死锁时, 已获取 readLock 的线程还是能重复获取 readLock)获取 writeLock 时 一定是在没有线程获取 readLock 或 writeLock 时才获取成功 (PS: 一个典型的死锁场景就是 一个线程先获取readLock, 后又获取writeLock)锁的获取支持线程中断, 且writeLock 中支持 Condition (PS: condition 只支持排他的场景)

ReentrantReadWriteLock的构造函数

ReentrantReadWriteLock 支持公平与非公平模式, 构造函数中可以通过指定的值传递进去

//用 nonfair 来构建 read/WriteLock (这里的 nonfair 指的是当进行获取 lock 时 若 aqs的syn queue 里面是否有 Node 节点而决定所采取的的策略) public KReentrantReadWriteLock(){ this(false); } public KReentrantReadWriteLock(boolean fair){ sync = fair ? new FairSync() : new NonfairSync(); readerLock = new ReadLock(this); writerLock = new WriteLock(this); }

而公不公平主要区分在lock的获取策略上, 而 readLock writeLock主要依据 Sync 类, 下面我们来看一下 Sync 类

ReentrantReadWriteLock 内部类 Sync

Sync 是 ReentrantReadWriteLock 的核心类, 主要做readLock writeLock 的获取, 数据存放操作(PS: 其子类 FairSync NonfairSync 只是实现获取的策略是不是要阻塞)

读写锁计数存放

/** * ReentrantReadWriteLock 这里使用 AQS里面的 state的高低16位来记录 read /write 获取的次数(PS: writeLock 是排他的 exclusive, readLock 是共享的 sahred, ) * 记录的操作都是通过 CAS 操作(有竞争发生) * * 特点: * 1) 同一个线程可以拥有 writeLock 与 readLock (但必须先获取 writeLock 再获取 readLock, 反过来进行获取会导致死锁) * 2) writeLock 与 readLock 是互斥的(就像 Mysql 的 X S 锁) * 3) 在因 先获取 readLock 然后再进行获取 writeLock 而导致 死锁时, 本线程一直卡住在对应获取 writeLock 的代码上(因为 readLock 与 writeLock 是互斥的, 在获取 writeLock 时监测到现在有线程获取 readLock , 锁一会一直在 aqs 的 sync queue 里面进行等待), 而此时 * 其他的线程想获取 writeLock 也会一直 block, 而若获取 readLock 若这个线程以前获取过 readLock, 则还能继续 重入 (reentrant), 而没有获取 readLock 的线程因为 aqs syn queue 里面有获取 writeLock 的 Node 节点存在会存放在 aqs syn queue 队列里面 一直 block */ /** 对 32 位的 int 进行分割 (对半 16) */ static final int SHARED_SHIFT = 16; static final int SHARED_UNIT = (1 << SHARED_SHIFT); // 000000000 00000001 00000000 00000000 static final int MAX_COUNT = (1 << SHARED_SHIFT) - 1; // 000000000 00000000 11111111 11111111 static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1; // 000000000 00000000 11111111 11111111 /** 计算 readLock 的获取次数(包含 reentrant 的次数) */ static int sharedCount(int c) { return c >>> SHARED_SHIFT; } // 将字节向右移动 16位, 只剩下 原来的 高 16 位 /** 计算 writeLock 的获取的次数(包括 reentrant的次数) */ static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; } // 与 EXCLUSIVE_MASK 与一下

读写锁的获取次数存放在 AQS 里面的state上, state的高 16 位存放 readLock 获取的次数, 低16位 存放 writeLock 获取的次数

针对readLock 存储每个线程获取的次数是使用内部类 HoldCounter, 并且存储在 ThreadLocal 里面

/** * 几乎每个获取 readLock 的线程都会含有一个 HoldCounter 用来记录 线程 id 与 获取 readLock 的次数 ( writeLock 的获取是由 state 的低16位 及 aqs中的exclusiveOwnerThread 来进行记录) * 这里有个注意点 第一次获取 readLock 的线程使用 firstReader, firstReaderHoldCount 来进行记录 * (PS: 不对, 我们想一下为什么不 统一用 HoldCounter 来进行记录呢? 原因: 所用的 HoldCounter 都是放在 ThreadLocal 里面, 而很多有些场景中只有一个线程获取 readLock 与 writeLock , 这种情况还用 ThreadLocal 的话那就有点浪费(ThreadLocal.get() 比直接 通过 reference 来获取数据相对来说耗性能)) */ static final class HoldCounter { int count = 0; // 重复获取 readLock/writeLock 的次数 final long tid = getThreadId(Thread.currentThread()); // 线程 id } /** 简单的自定义的 ThreadLocal 来用进行记录 readLock 获取的次数 */ static final class ThreadLocalHoldCounter extends ThreadLocal<HoldCounter>{ @Override protected HoldCounter initialValue() { return new HoldCounter(); } } /** * readLock 获取记录容器 ThreadLocal(ThreadLocal 的使用过程中当 HoldCounter.count == 0 时要进行 remove , 不然很有可能导致 内存的泄露) */ private transient ThreadLocalHoldCounter readHolds; /** * 最后一次获取 readLock 的 HoldCounter 的缓存 * (PS: 还是上面的问题 有了 readHolds 为什么还需要 cachedHoldCounter呢? 大非常大的场景中, 这次进行release readLock的线程就是上次 acquire 的线程, 这样直接通过cachedHoldCounter来进行获取, 节省了通过 readHolds 的 lookup 的过程) */ private transient HoldCounter cachedHoldCounter; /** * 下面两个是用来进行记录 第一次获取 readLock 的线程的信息 * 准确的说是第一次获取 readLock 并且 没有 release 的线程, 一旦线程进行 release readLock, 则 firstReader会被置位 null */ private transient Thread firstReader = null; private transient int firstReaderHoldCount;

针对获取 readLock 的线程的获取次数需要分3种情况

线程的tid 及获取次数 count 存放在 HoldCounter 里面, 最后放在ThreadLocal 中从cachedHoldCounter获取存入的信息, 额, 这里不是有 ThreadLocal,干嘛还需要cachedHoldCounter呢? 原因是这样的, 但多数情况在进行线程 acquire readLock后不久就会进行相应的release, 而从 cachedHoldCounter 获取, 省去了从ThreadLocal 中 lookup 的操作(其实就是节省资源, ThreadLocal 中的查找需要遍历数组)firstReader firstReaderHoldCount 这两个属性是用来记录第一次获取锁的线程,及重入的次数(这里说第一次有点不准确, 因为当线程进行释放 readLock 后, firstReader 会被置空,当再有新的线程获取 readLock 后, firstReader 就会被赋值新的线程)

Sync 的抽象方法

/** * 当线程进行获取 readLock 时的策略(这个策略依赖于 aqs 中 sync queue 里面的Node存在的情况来定), * @return */ abstract boolean readerShouldBlock(); /** * 当线程进行获取 writerLock 时的策略(这个策略依赖于 aqs 中 sync queue 里面的Node存在的情况来定) * @return */ abstract boolean writerShouldBlock();

这两个方法主要用于子类实现lock的获取策略

内部类 Sync 的 tryRelease 方法

这个方法主要用于独占锁释放时操作 AQS里面的state状态

/** - 在进行 release 锁 时, 调用子类的方法 tryRelease(主要是增对 aqs 的 state 的一下赋值操作) (PS: 这个操作只有exclusive的lock才会调用到) - @param releases - @return */ protected final boolean tryRelease(int releases){ if(!isHeldExclusively()){ // 1 监测当前的线程进行释放锁的线程是否是获取独占锁的线程 throw new IllegalMonitorStateException(); } int nextc = getState() - releases; // 2. 进行 state 的释放操作 boolean free = exclusiveCount(nextc) == 0; // 3. 判断 exclusive lock 是否释放完(因为这里支持 lock 的 reentrant) if(free){ // 4. 锁释放掉后 清除掉 独占锁 exclusiveOwnerThread 的标志 setExclusiveOwnerThread(null); } setState(nextc); // 5. 直接修改 state 的值 (PS: 这里没有竞争的出现, 因为调用 tryRelease方法的都是独占锁, 互斥, 所以没有 readLock 的获取, 相反 readLock 对 state 的修改就需要 CAS 操作) return free; }

整个操作流程比较简单, 有两个注意点

这里的 state 操作没用 CAS, 为啥? 主要这是独占的(此刻没有其他的线程获取 readLock, 所以没有竞争的更改state的情况)方法最后返回的是 free, free 指的是writeLock是否完全释放完, 因为这里有 锁重入的情况, 而在完全释放好之后才会有后续的唤醒操作

内部类 Sync 的 tryAcquire 方法

tryAcquire方法是 AQS 中 排他获取锁 模板方法acquire里面的策略方法

/** - AQS 中 排他获取锁 模板方法acquire里面的策略方法 tryAcquire 的实现 - @param acquires - @return */ protected final boolean tryAcquire(int acquires){ Thread current = Thread.currentThread(); int c = getState(); int w = exclusiveCount(c); // 1. 获取现在writeLock 的获取的次数 if(c != 0){ // Note: if c != 0 and w == 0 then shared count != 0 if(w == 0 || current != getExclusiveOwnerThread()){ // 2. 并发的情况来了, 这里有两种情况 (1) c != 0 && w == 0 -> 说明现在只有读锁的存在, 则直接 return, return后一般就是进入 aqs 的 sync queue 里面进行等待获取 (2) c != 0 && w != 0 && current != getExclusiveOwnerThread() 压根就是其他的线程获取 read/writeLock, 读锁是排他的, 所以这里也直接 return -> 进入 aqs 的 sync queue 队列 return false; } if(w + exclusiveCount(acquires) > MAX_COUNT){ // 3. 计算是否获取writeLock的次数 饱和了(saturate) throw new Error("Maximum lock count exceeded"); } // Reentrant acquire setState(c + acquires); // 4. 进行 state值得修改 (这里也不需要 CAS 为什么? 读锁是排他的, 没有其他线程和他竞争修改) return true; } if(writerShouldBlock() || !compareAndSetState(c, c + acquires)){ // 5. 代码运行到这里 (c == 0) 这时可能代码刚刚到这边时, 就有可能其他的线程获取读锁, 所以 c == 0 不一定了, 所以需要再次调用 writerShouldBlock查看, 并且用 CAS 来进行 state 值得更改 return false; } setExclusiveOwnerThread(current); // 6. 设置 exclusiveOwnerThread writeLock 获取成功 return true; } c != 0 代表有线程获取 read/writeLock, 这里有两种情况 (1) c != 0 && w == 0 ->说明现在只有读锁的存在, 则直接 return, return后一般就是进入 aqs 的 sync queue 里面进行等待获取 (2)c != 0 && w != 0 && current != getExclusiveOwnerThread() 压根就是其他的线程获取read/writeLock, 读锁是排他的, 所以这里也直接 return -> 进入 aqs 的 sync queue 队列writerShouldBlock 是判断是否需要进入 AQS 的 Sync Queue 队列

内部类 Sync 的 tryReleaseShared 方法

这个方法是 readLock 进行释放lock时调用的

/** * AQS 里面 releaseShared 的实现 * @param unused * @return */ protected final boolean tryReleaseShared(int unused){ Thread current = Thread.currentThread(); if(firstReader == current){ // 1. 判断现在进行 release 的线程是否是 firstReader // assert firstReaderHoldCount > 0 if(firstReaderHoldCount == 1){ // 2. 只获取一次 readLock 直接置空 firstReader firstReader = null; }else{ firstReaderHoldCount--; // 3. 将 firstReaderHoldCount 减 1 } }else{ HoldCounter rh = cachedHoldCounter; // 4. 先通过 cachedHoldCounter 来取值 if(rh == null || rh.tid != getThreadId(current)){ // 5. cachedHoldCounter 代表的是上次获取 readLock 的线程, 若这次进行 release 的线程不是, 再通过 readHolds 进行 lookup 查找 rh = readHolds.get(); } int count = rh.count; if(count <= 1){ readHolds.remove(); // 6. count <= 1 时要进行 ThreadLocal 的 remove , 不然容易内存泄露 if(count <= 0){ throw unmatchedUnlockException(); // 7. 并发多次释放就有可能出现 } } --rh.count; // 9. HoldCounter.count 减 1 } for(;;){ // 10. 这里是一个 loop CAS 操作, 因为可能其他的线程此刻也在进行 release操作 int c = getState(); int nextc = c - SHARED_UNIT; // 11. 这里是 readLock 的减 1, 也就是 aqs里面state的高 16 上进行 减 1, 所以 减 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; // 12. 返回值是判断 是否还有 readLock 没有释放完, 当释放完了会进行 后继节点的 唤醒( readLock 在进行获取成功时也进行传播式的唤醒后继的 获取 readLock 的节点) } } }

整个过程主要是 firstReader, cachedHoldCounter, readHolds 数据操作, 需要注意的是当 count == 1 时需要进行 readHolds.remove(), 不然会导致内存的泄漏

内部类 Sync 的 tryAcquireShared 方法

AQS 中 acquireShared 的子方法，主要是进行改变 aqs 的state的值进行获取 readLock

/** * AQS 中 acquireShared 的子方法 * 主要是进行改变 aqs 的state的值进行获取 readLock * @param unused * @return */ protected final int tryAcquireShared(int unused){ Thread current = Thread.currentThread(); int c = getState(); // 1. 判断是否有其他的线程获取了 writeLock, 有的话直接返回 -1 进行 aqs的 sync queue 里面 if(exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current){ return -1; } int r = sharedCount(c); // 2. 获取 readLock的获取次数 if(!readerShouldBlock() && r < MAX_COUNT && compareAndSetState(c, c + SHARED_UNIT)){ // 3. if 中的判断主要是 readLock获取的策略, 及 操作 CAS 更改 state 值是否OK if(r == 0){ // 4. r == 0 没有线程获取 readLock 直接对 firstReader firstReaderHoldCount 进行初始化 firstReader = current; firstReaderHoldCount = 1; }else if(firstReader == current){ // 5. 第一个获取 readLock 的是 current 线程, 直接计数器加 1 firstReaderHoldCount++; }else{ HoldCounter rh = cachedHoldCounter; if(rh == null || rh.tid != getThreadId(current)){ // 6. 还是上面的逻辑, 先从 cachedHoldCounter, 数据不对的话, 再从readHolds拿数据 cachedHoldCounter = rh = readHolds.get(); }else if(rh.count == 0){ // 7. 为什么要 count == 0 时进行 ThreadLocal.set? 因为上面 tryReleaseShared方法 中当 count == 0 时, 进行了ThreadLocal.remove readHolds.set(rh); } rh.count++; // 8. 统一的 count++ } return 1; } return fullTryAcquireShared(current); // 9.代码调用 fullTryAcquireShared 大体情况是 aqs 的 sync queue 里面有其他的节点 或 sync queue 的 head.next 是个获取 writeLock 的节点, 或 CAS 操作 state 失败 }

在遇到writeLock被其他的线程占用时, 直接返回 -1; 而整个的操作无非是 firstReader cachedHoldCounter readHolds 的赋值操作;当 遇到 readerShouldBlock == true 或 因竞争导致 “compareAndSetState(c, c + SHARED_UNIT)” 失败时会用兜底的函数 fullTryAcquireShared 来进行解决

内部类 Sync 的 fullTryAcquireShared 方法

fullTryAcquireShared 这个方法其实是 tryAcquireShared 的冗余(redundant)方法, 主要补足 readerShouldBlock 导致的获取等待 和 CAS 修改 AQS 中 state 值失败进行的修补工作

/** - fullTryAcquireShared 这个方法其实是 tryAcquireShared 的冗余(redundant)方法, 主要补足 readerShouldBlock 导致的获取等待 和 CAS 修改 AQS 中 state 值失败进行的修补工作 */ final int fullTryAcquireShared(Thread current){ HoldCounter rh = null; for(;;){ int c= getState(); if(exclusiveCount(c) != 0){ if(getExclusiveOwnerThread() != current) // 1. 若此刻 有其他的线程获取了 writeLock 则直接进行 return 到 aqs 的 sync queue 里面 return -1; // else we hold the exclusive lock; blocking here // would cause deadlock }else if(readerShouldBlock()){ // 2. 判断 获取 readLock 的策略 // Make sure we're not acquiring read lock reentrantly if(firstReader == current){ // 3. 若是 readLock 的 重入获取, 则直接进行下面的 CAS 操作 // 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(); // 4. 若 rh.count == 0 进行 ThreadLocal.remove } } } if(rh.count == 0){ // 5. count != 0 则说明这次是 readLock 获取锁的 重入(reentrant), 所以即使出现死锁, 以前获取过 readLock 的线程还是能继续 获取 readLock return -1; // 6. 进行到这一步只有 当 aqs sync queue 里面有 获取 readLock 的node 或 head.next 是获取 writeLock 的节点 } } } if(sharedCount(c) == MAX_COUNT){ // 7. 是否获取 锁溢出 throw new Error("Maximum lock count exceeded"); } if(compareAndSetState(c, c + SHARED_UNIT)){ // 8. CAS 可能会失败, 但没事, 我们这边外围有个 for loop 来进行保证 操作一定进行 if(sharedCount(c) == 0){ // 9. r == 0 没有线程获取 readLock 直接对 firstReader firstReaderHoldCount 进行初始化 firstReader = current; firstReaderHoldCount = 1; }else if(firstReader == current){ // 10. 第一个获取 readLock 的是 current 线程, 直接计数器加 1 firstReaderHoldCount++; }else{ if(rh == null){ rh = cachedHoldCounter; } if(rh == null || rh.tid != getThreadId(current)){ rh = readHolds.get(); // 11. 还是上面的逻辑, 先从 cachedHoldCounter, 数据不对的话, 再从readHolds拿数据 }else if(rh.count == 0){ readHolds.set(rh); // 12. 为什么要 count == 0 时进行 ThreadLocal.set? 因为上面 tryReleaseShared方法 中当 count == 0 时, 进行了ThreadLocal.remove } rh.count++; cachedHoldCounter = rh; // cache for release // 13. 获取成功 } return 1; } } }

若 有其他的线程获取writeLock, 则直接 return -1, 将 线程放入到 AQS 的 sync queue 里面

代码中的 “firstReader == current” 其实表明无论什么情况, readLock都可以重入的获取(包括死锁的情况)

内部类 Sync 的 tryReadLock 方法

/** * 尝试性的获取一下 readLock */ final boolean tryReadLock(){ Thread current = Thread.currentThread(); for(;;){ int c = getState(); if(exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current){ // 1. 若当前有其他的线程获取 writeLock 直接 return return false; } int r = sharedCount(c); // 2. 获取 readLock 的次数 if(r == MAX_COUNT){ throw new Error("Maximum lock count exceeded"); } if(compareAndSetState(c, c + SHARED_UNIT)){ // 3. CAS 设置 state if(r == 0){ // 4. r == 0 没有线程获取 readLock 直接对 firstReader firstReaderHoldCount 进行初始化 firstReader = current; firstReaderHoldCount = 1; }else if(firstReader == current){ // 5. 第一个获取 readLock 的是 current 线程, 直接计数器加 1 firstReaderHoldCount++; }else{ HoldCounter rh = cachedHoldCounter; if(rh == null || rh.tid != getThreadId(current)){ // 6. 还是上面的逻辑, 先从 cachedHoldCounter, 数据不对的话, 再从readHolds拿数据 cachedHoldCounter = rh = readHolds.get(); }else if(rh.count == 0){ readHolds.set(rh); // 7. 为什么要 count == 0 时进行 ThreadLocal.set? 因为上面 tryReleaseShared方法 中当 count == 0 时, 进行了ThreadLocal.remove } rh.count++; } return true; } } }

尝试性的获取ReadLock, 失败的话就直接返回

内部类 Sync 的 tryWriteLock 方法

/** * 尝试性的获取 writeLock 失败的话也无所谓 * @return */ final boolean tryWriteLock(){ Thread current = Thread.currentThread(); int c = getState(); if(c != 0){ int w = exclusiveCount(c); // 1. 获取现在writeLock 的获取的次数 if(w == 0 || current != getExclusiveOwnerThread()){ // 2. 判断是否是其他的线程获取了 writeLock return false; } if(w == MAX_COUNT){ // 3. 获取锁是否 溢出 throw new Error("Maximum lock count exceeded"); } } if(!compareAndSetState(c, c + 1)){ // 4. 这里有竞争, cas 操作失败也无所谓 return false; } setExclusiveOwnerThread(current); // 5. 设置 当前的 exclusiveOwnerThread return true; }

尝试性的获取 writeLock

内部类 Sync 的 基本方法

/** 判断当前线程是否 是 writeLock 的获取者 */ 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(); } /** 创建一个 condition, condition 只用于 独占场景 */ // Methods relayed to outer class final ConditionObject newCondition(){ return new ConditionObject(); } /** 判断当前线程是否 是 writeLock 的获取者 */ final Thread getOwner(){ // Must read state before owner to ensure memory consistency return ((exclusiveCount(getState()) == 0 )? null : getExclusiveOwnerThread()); } /** 获取 readLock 的获取次数 */ final int getReadLockCount(){ return sharedCount(getState()); } /** 判断 writeLock 是否被获取 */ final boolean isWriteLocked(){ return exclusiveCount(getState()) != 0; } /** 获取 writeLock 的获取次数 */ final int getWriteHoldCount(){ return isHeldExclusively()?exclusiveCount(getState()) : 0; } /** 获取当前线程获取 readLock 的次数 */ 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(ObjectInputStream s) throws Exception{ s.defaultReadObject(); readHolds = new ThreadLocalHoldCounter(); setState(0); // reset to unlocked state } final int getCount(){ return getState(); }

内部类 FairSync NonfairSync

锁的获取释放主要由 AQS, Sync 来实现, 这两个类主要是用来定义锁获取的策略(是否需要抢占)

/** * Nonfair version of Sync * 非公平版本 sync */ static final class NonfairSync extends Sync{ private static final long serialVersionUID = -8159625535654395037L; @Override boolean readerShouldBlock() { /** readLock 的获取主要看 aqs sync queue 队列里面的 head.next 是否是获取 读锁的 */ return apparentlyFirstQueuedIsExclusive(); } @Override boolean writerShouldBlock() { // 获取 writeLock 的话 直接获取 return false; // writers can always barge } } /** * Fair version of Sync * 公平版的 sync */ static final class FairSync extends Sync{ private static final long serialVersionUID = -2274990926593161451L; /** * readerShouldBlock writerShouldBlock 都是看 aqs sync queue 里面是否有节点 */ @Override boolean readerShouldBlock() { return hasQueuedPredecessors(); } @Override boolean writerShouldBlock() { return hasQueuedPredecessors(); } }

apparentlyFirstQueuedIsExclusive 主要是查看 AQS Sync Queue 里面 head.next 节点是否是获取 writeLock 的

hasQueuedPredecessors 是查看当前 AQS Sync Queue 里面是否有前继节点, 这个方法有优秀的并发代码

内部类 ReadLock WriteLock

这两个类主要还是调用Sync 里面的代码, 下面只是对代码做了一些简单的描述

/** * The lock returned by method {@link KReentrantReadWriteLock} * 读锁 */ public static class ReadLock implements Lock, Serializable{ private static final long serialVersionUID = -5992448646407690164L; private final Sync sync; /** * Constructor for use by subclasses * * @param lock the outer lock object * @throws NullPointerException if the lock is null */ protected ReadLock(KReentrantReadWriteLock lock){ sync = lock.sync; } /** * 所得获取都是调用 aqs 中 acquireShared */ public void lock() { sync.acquireShared(1); } /** * 支持中断的获取锁 */ public void lockInterruptibly() throws InterruptedException{ sync.acquireSharedInterruptibly(1); } /** * 尝试获取锁 */ public boolean tryLock(){ return sync.tryReadLock(); } /** * 支持中断与 timeout 的获取 writeLock */ public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException{ return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout)); } /** * 释放 readLock */ public void unlock(){ sync.releaseShared(1); } /** * 创建一个 condition */ public Condition newCondition(){ throw new UnsupportedOperationException(); } public String toString(){ int r = sync.getReadLockCount(); return super.toString() + " [Read locks = " + r + "]"; } } /** * The lock returned by method {@link KReentrantReadWriteLock} */ /** * 写锁 */ public static class WriteLock implements Lock, Serializable{ private static final long serialVersionUID = -4992448646407690164L; private final Sync sync; /** * Constructor for use by subclasses * * @param lock the outer lock object * @throws NullPointerException if the lock is null */ protected WriteLock(KReentrantReadWriteLock lock){ sync = lock.sync; } /** * 调用 aqs 的 acquire 来获取 锁 */ public void lock(){ sync.acquire(1); } /** * 支持中断方式的获取 writeLock */ public void lockInterruptibly() throws InterruptedException{ sync.acquireInterruptibly(1); } /** * 尝试性的获取锁 */ public boolean tryLock(){ return sync.tryWriteLock(); } /** * 支持中断 timeout 方式获取锁 */ public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException{ return sync.tryAcquireNanos(1, unit.toNanos(timeout)); } /** * 释放锁 */ public void unlock(){ sync.release(1); } /** * new 一个 condition */ public Condition newCondition(){ return sync.newCondition(); } public String toString(){ Thread o = sync.getOwner(); return super.toString() + ((o == null) ? "[Unlocked]" : "[Locked by thread " + o.getName() + "]"); } /** * 判断当前的 writeLock 是否被 本线程 占用 */ public boolean isHeldByCurrentThread(){ return sync.isHeldExclusively(); } /** * 获取 writeLock 的获取次数 */ public int getHoldCount(){ return sync.getWriteHoldCount(); } }

ReentrantReadWriteLock 死锁案例

在上面进行分析代码时, 一直提到 ReentrantReadWriteLock 的死锁问题, 主要是 先获取了 readLock, 然后在获取writeLock, 这样就会出现死锁; 先看一下下面的demo

import org.apache.log4j.Logger; import java.util.concurrent.locks.ReentrantReadWriteLock; /** * ReentrantReadWriteLock 死锁 demo * Created by xujiankang on 2017/2/6. */ public class ReentrantReadWriteLockTest { private static final Logger logger = Logger.getLogger(ReentrantReadWriteLockTest.class); public static void main(String[] args) throws Exception{ final ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock(); logger.info("readWriteLock.readLock().lock() begin"); readWriteLock.readLock().lock(); logger.info("readWriteLock.readLock().lock() over"); new Thread(){ @Override public void run() { for(int i = 0; i< 2; i++){ logger.info(" "); logger.info("Thread readWriteLock.readLock().lock() begin i:"+i); readWriteLock.readLock().lock(); // 获取过一次就能再次获取, 但是若其他没有获取的线程因为 syn queue里面 head.next 是获取write的线程, 则到 syn queue 里面进行等待 logger.info("Thread readWriteLock.readLock().lock() over i:" + i); try { Thread.sleep(1000); } catch (InterruptedException e) { } } } }.start(); Thread.sleep(1 * 1000); logger.info("readWriteLock.writeLock().lock() begin"); readWriteLock.writeLock().lock(); logger.info("readWriteLock.writeLock().lock() over"); } }

代码执行结果 :

[2017-02-08 00:29:02,192] INFO main (ReentrantReadWriteLockTest.java:18) - readWriteLock.readLock().lock() begin [2017-02-08 00:29:02,199] INFO main (ReentrantReadWriteLockTest.java:20) - readWriteLock.readLock().lock() over [2017-02-08 00:29:02,201] INFO Thread-0 (ReentrantReadWriteLockTest.java:27) - [2017-02-08 00:29:02,202] INFO Thread-0 (ReentrantReadWriteLockTest.java:28) - Thread readWriteLock.readLock().lock() begin i:0 [2017-02-08 00:29:02,202] INFO Thread-0 (ReentrantReadWriteLockTest.java:30) - Thread readWriteLock.readLock().lock() over i:0 [2017-02-08 00:29:03,204] INFO main (ReentrantReadWriteLockTest.java:41) - readWriteLock.writeLock().lock() begin [2017-02-08 00:29:03,204] INFO Thread-0 (ReentrantReadWriteLockTest.java:27) - [2017-02-08 00:29:03,208] INFO Thread-0 (ReentrantReadWriteLockTest.java:28) - Thread readWriteLock.readLock().lock() begin i:1 [2017-02-08 00:29:03,208] INFO Thread-0 (ReentrantReadWriteLockTest.java:30) - Thread readWriteLock.readLock().lock() over i:1

执行状况

主线程获取 readLock 成功, 但在获取 writeLock 一直卡在这边子线程两次获取 readLock都成功了(PS: 第二次获取 readLock是在 主线程获取 writeLock 之后, 并且还成功了)

原因分析:

主线程获取 readLock 成功子线程在开始第一次获取 readLock 时 AQS 的 Sync Queue 里面没有等待的节点, 没有获取 writeLock 的节点, 所以直接获取成功主线程开始获取 writeLock, 这时发现 readLock 已经 readLock 已经被线程获取, 且writeLock没有被人获取(方法 tryAcquire c != 0 && w == 0), 则直接加入到 AQS 的 Sync Queue 里面, 并且一直等待 readLock 的释放(也就是等待自己的readLock释放 <- 这是不可能的)子线程在睡了1秒后再次获取 readLock 锁, 而此时 获取 writeLock 的节点已经在 AQS 的 Sync queue 里面, 所以 readerShouldBlock == true (见方法 tryAcquireShared), 直接跳到 fullTryAcquireShared 中进行获取, 在fullTryAcquireShared中又因为 “readerShouldBlock == true && rh.count != 0”, 所以可以执行 fullTryAcquireShared下面的代码, 而获取 readLock 成功

我们改变一下代码, 将主线程中的 “Thread.sleep(1 * 1000);” 注释掉, 在子线程获取 readLock 之前进行睡1秒 这时发现, 子线程获取 readLock 不成功了

import org.apache.log4j.Logger; import java.util.concurrent.locks.ReentrantReadWriteLock; /** * ReentrantReadWriteLock 死锁 demo * Created by xujiankang on 2017/2/6. */ public class ReentrantReadWriteLockTest { private static final Logger logger = Logger.getLogger(ReentrantReadWriteLockTest.class); public static void main(String[] args) throws Exception{ final ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock(); logger.info("readWriteLock.readLock().lock() begin"); readWriteLock.readLock().lock(); logger.info("readWriteLock.readLock().lock() over"); new Thread(){ @Override public void run() { for(int i = 0; i< 2; i++){ try { Thread.sleep(1000); } catch (InterruptedException e) { } logger.info(" "); logger.info("Thread readWriteLock.readLock().lock() begin i:"+i); readWriteLock.readLock().lock(); // 获取过一次就能再次获取, 但是若其他没有获取的线程因为 syn queue里面 head.next 是获取write的线程, 则到 syn queue 里面进行等待 logger.info("Thread readWriteLock.readLock().lock() over i:" + i); try { Thread.sleep(1000); } catch (InterruptedException e) { } } } }.start(); // Thread.sleep(1 * 1000); logger.info("readWriteLock.writeLock().lock() begin"); readWriteLock.writeLock().lock(); logger.info("readWriteLock.writeLock().lock() over"); } }

执行结果:

[2017-02-08 00:35:34,216] INFO main (ReentrantReadWriteLockTest.java:18) - readWriteLock.readLock().lock() begin [2017-02-08 00:35:34,228] INFO main (ReentrantReadWriteLockTest.java:20) - readWriteLock.readLock().lock() over [2017-02-08 00:35:34,229] INFO main (ReentrantReadWriteLockTest.java:47) - readWriteLock.writeLock().lock() begin [2017-02-08 00:35:35,231] INFO Thread-0 (ReentrantReadWriteLockTest.java:33) - [2017-02-08 00:35:35,232] INFO Thread-0 (ReentrantReadWriteLockTest.java:34) - Thread readWriteLock.readLock().lock() begin i:0

执行状态:

主线程 readLock 获取成功, writeLock 获取一直阻塞在那边子线程获取 readLock 一直失败

原因分析:

主线程获取 readLock 成功主线程开始获取 writeLock, 这时发现 readLock 已经 readLock 已经被线程获取, 且writeLock没有被人获取(方法 tryAcquire c != 0 && w == 0), 则直接加入到 AQS 的 Sync Queue 里面, 并且一直等待 readLock 的释放(也就是等待自己的readLock释放 <- 这是不可能的)子线程睡过 1 秒后, 开始第一次获取 readLock, 此时 AQS 的 Sync Queue 里面有等待的节点(阻线程获取 writeLock 的节点 head.next), 所以 readerShouldBlock == true (见方法 tryAcquireShared), 则直接跳到 fullTryAcquireShared方法中进行获取, 在方法 fullTryAcquireShared 中 因 “readerShouldBlock == true && rh.count == 0” 所以直接 return -1, 放入 AQS Sync Queue 里面, 进而一直等待

对ReentrantReadWriteLock的总结

ReentrantReadWriteLock允许多个读线程同时访问，但不允许写线程和读线程、写线程和写线程同时访问。相对于排他锁，提高了并发性。在实际应用中，大部分情况下对共享数据（如缓存）的访问都是读操作远多于写操作，这时ReentrantReadWriteLock能够提供比排他锁更好的并发性和吞吐量。读写锁内部维护了两个锁，一个用于读操作，一个用于写操作。所有ReadWriteLock实现都必须保证writeLock操作的内存同步效果也要保持与相关readLock的联系。也就是说，成功获取读锁的线程会看到写入锁之前版本所做的所有更新。

转载至：ReentrantReadWriteLock 源码分析(基于Java 8)