ReentrantReadWriteLock死锁问题解析

1、案情代码分析：

private static ReentrantReadWriteLock reentrantReadWriteLock = new ReentrantReadWriteLock(true); public static void main(String[] args){ // thread a new Thread(() -> { System.out.println("a第一次获取读锁"); reentrantReadWriteLock.readLock().lock(); try { TimeUnit.SECONDS.sleep(5); } catch (InterruptedException e) { } System.out.println("a尝试再次获取写锁"); reentrantReadWriteLock.writeLock().lock(); System.out.println("a获取到写锁"); }).start(); // thread b new Thread(() -> { try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("b线程尝试获取到读锁"); reentrantReadWriteLock.readLock().lock(); System.out.println("b线程获取到读锁"); try { TimeUnit.SECONDS.sleep(7); } catch (InterruptedException e) { e.printStackTrace(); } reentrantReadWriteLock.readLock().unlock(); System.out.println("b线程释放到读锁"); }).start(); // thread c new Thread(() -> { try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("c线程尝试获取到读锁"); reentrantReadWriteLock.readLock().lock(); System.out.println("c线程获取到读锁"); try { TimeUnit.SECONDS.sleep(7); } catch (InterruptedException e) { e.printStackTrace(); } reentrantReadWriteLock.readLock().unlock(); System.out.println("c线程释放到读锁"); }).start(); // reentrantReadWriteLock.writeLock().unlock(); // reentrantReadWriteLock.readLock().unlock(); }

2、输出结果

a第一次获取读锁 b线程尝试获取到读锁 c线程尝试获取到读锁 b线程获取到读锁 c线程获取到读锁 a尝试再次获取写锁 b线程释放到读锁 c线程释放到读锁 。。。。程序并没有停止

3、案情分析

线程b和c创建后休眠2s，确保线程a能够获取读锁，随后b和c线程唤醒，获取读锁；随后b和c继续休眠7s，a线程从5s后醒来，尝试获取写锁；b和c从7s的休眠后醒来，可以释放掉读锁，但是程序并没有停止。

4、使用jstack分析应用的栈信息

"DestroyJavaVM" #14 prio=5 os_prio=0 tid=0x0000000003735800 nid=0x4620 waiting on condition [0x0000000000000000] java.lang.Thread.State: RUNNABLE Locked ownable synchronizers: - None "Thread-0" #11 prio=5 os_prio=0 tid=0x000000001a31a000 nid=0x3940 waiting on condition [0x000000001abbe000] java.lang.Thread.State: WAITING (parking) at sun.misc.Unsafe.park(Native Method) - parking to wait for <0x00000000d6269068> (a java.util.concurrent.locks.ReentrantReadWriteLock$FairSync) at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175) at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836) at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireQueued(AbstractQueuedSynchronizer.java:870) at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(AbstractQueuedSynchronizer.java:1199) at java.util.concurrent.locks.ReentrantReadWriteLock$WriteLock.lock(ReentrantReadWriteLock.java:943) at com.feather.WordCountExample.lambda$main$0(WordCountExample.java:25) at com.feather.WordCountExample$$Lambda$1/1555093762.run(Unknown Source) at java.lang.Thread.run(Thread.java:748) Locked ownable synchronizers: - None

直接去了线程Thread-0的信息，可以看到线程处于writeLock.lock()方法中，一直在等待唤醒。

5、源码分析

首先这一切发生都是在a、b、c线程获取到读锁之后，a再次获取写锁导致死锁发生。所以我们假定目前ReentrantReadWriteLock中只有共享锁，而接着a尝试在有读锁的时候获取写锁，来看写锁的获取过程。

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

写锁会尝试限制性tryAcquire，如果失败的话就会进入队列中，所以先看tryAcquire这个方法

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; 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; }

tryAcquire的方法比较简单，可以看出先判断当前有无线程拿到锁，就是直接判断state是否为0即可，因为前面已经有a、b、c三个线程获取到了资源，所以c不为0；同时因为当前没有写锁被获取过，所以exclusiveCount(c)返回结果应该也是为false，所以在if (w == 0 || current != getExclusiveOwnerThread())，这个条件应该是直接通过并且整个方法返回false。

因为tryAcquire的返回结果为false，所以我们来看acuqireQueued整个方法(addWaiter方法就是创建一个EXCLUSIVE的节点并且入队)

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); } }

因为当前只有a尝试获取写锁，并且因为失败入队，所以当前队列只有一个虚头节点，以及准备获取写锁的线程a（从这里也可以看出尽管是同一个线程，但是写锁和读锁也是分开判断的）。

这里tryAcquire在公平锁和非公平锁有两种不同的实现：

// 非公平锁 protected final boolean tryAcquire(int acquires) { return nonfairTryAcquire(acquires); } 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; }

在非公平锁中，分两种情况：

1）直接尝试CAS，失败则返回false；

2）当前线程已经获取过写锁了，所以直接重入即可。

// 公平锁 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; 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; }

而在公平锁中，先判断是否有其他线程获取过资源，然后再判断是否有线程拿过写锁，很明显这里会在if (w == 0 || current != getExclusiveOwnerThread())进入，并返回结果false。

所以在已有多数线程获取读锁的情况下，其中一个线程重复获取写锁会失败，并在acuqireQueued这个方法中，进入shouldParkFailedAcquire进行判断。

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { int ws = pred.waitStatus; if (ws == Node.SIGNAL) return true; if (ws > 0) { do { node.prev = pred = pred.prev; } while (pred.waitStatus > 0); pred.next = node; } else { compareAndSetWaitStatus(pred, ws, Node.SIGNAL); } return false; }

shouldParkAfterFailedAcquire 这个方法不陌生了，虽然刚开始创建的头节点wai'tStatus=0，但是架不住acuqireQueued外面有一个死循环，所以会在tryAcquire获取失败的时候继续进来，并且返回结果true，并最终进入parkAndCheckInterrupt自行挂起。

问题就是在这，线程a获取写锁导致线程被挂起，同样也挂起了前面a获取的读锁；虽然b、和c释放了读锁，因为a一直没法释放读锁导致a的写锁一直无法迟迟等待a自己释放读锁，最终陷入死锁。下面是读锁尝试释放所的代码

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; // a读锁没有释放，所以nextc仍然不等于0 if (compareAndSetState(c, nextc)) return nextc == 0; } }

6、总结

    不要在获取了读锁的情况下在获取写锁，但是在拿到写锁的时候可以拿读锁(锁降级)。