聊聊高并发（二十九）解析java.util.concurrent各个组件（十一） 再看看ReentrantReadWriteLock可重入读-写锁

上一篇聊聊高并发（二十八）解析java.util.concurrent各个组件（十） 理解ReentrantReadWriteLock可重入读-写锁 讲了可重入读写锁的基本情况和主要的方法，显示了如何实现的锁降级。但是下面几个问题没说清楚，这篇补充一下

1. 释放锁时的优先级问题，是让写锁先获得还是先让读锁先获得

2. 是否允许读线程插队

3. 是否允许写线程插队，因为读写锁一般用在大量读，少量写的情况，如果写线程没有优先级，那么可能造成写线程的饥饿

关于释放锁后是让写锁先获得还是让读锁先获得，这里有两种情况

1. 释放锁后，请求获取写锁的线程不在AQS队列

2. 释放锁后，请求获取写锁的线程已经AQS队列

如果是第一种情况，那么非公平锁的实现下，获取写锁的线程直接尝试竞争锁也不用管AQS里面先来的线程。获取读锁的线程只判断是否已经有线程获得写锁（既Head节点是独占模式的节点），如果没有，那么就不用管AQS里面先来的准备获取读锁的线程。

static final class NonfairSync extends Sync {
        private static final long serialVersionUID = -8159625535654395037L;
        final boolean writerShouldBlock() {
            return false; // writers can always barge
        }
        final boolean readerShouldBlock() {
            return apparentlyFirstQueuedIsExclusive();
        }
    }

static final class FairSync extends Sync {
        private static final long serialVersionUID = -2274990926593161451L;
        final boolean writerShouldBlock() {
            return hasQueuedPredecessors();
        }
        final boolean readerShouldBlock() {
            return hasQueuedPredecessors();
        }
    }

下面这张图模拟了AQS队列中等待的线程节点的情况

1. Head节点始终是当前获得了锁的线程

2. 非Head节点在竞争锁失败后，acquire方法会不断地轮询，于自旋不同的是，AQS轮询过程中的线程是阻塞等待。

所以要理解AQS的release释放动作并不是让后续节点直接获取锁，而是唤醒后续节点unparkSuccessor()。真正获取锁的地方还是在acquire方法，被release唤醒的线程继续轮询状态，如果它的前驱是head，并且tryAcquire获取资源成功了，那么它就获得锁

public final boolean release(int arg) {
        if (tryRelease(arg)) {
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }

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

3. 图中Head之后有3个准备获取读锁的线程，最后是1个准备获取写锁的线程。

那么如果是AQS队列中的节点获取锁

如果释放失败了，那就把它的状态标记了PROPAGATE，当它释放的时候，会再次取尝试唤醒下一个读锁节点

如果后继节点是写锁，那么就不唤醒

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

 private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head; // Record old head for check below
        setHead(node);
        
        if (propagate > 0 || h == null || h.waitStatus < 0) {
            Node s = node.next;
            if (s == null || s.isShared())
                doReleaseShared();
        }
    }

private void doReleaseShared() {
        for (;;) {
            Node h = head;
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                if (ws == Node.SIGNAL) {
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;            // loop to recheck cases
                    unparkSuccessor(h);
                }
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            if (h == head)                   // loop if head changed
                break;
        }
    }

关于读锁能不能插队的问题，非公平性的Sync提供了插队的可能，但是前提是它在tryAcquire就成功获得了，如果tryAcquire失败了，它就得进入AQS队列排队，也不会出现让写锁饥饿的情况。

关于写锁能不能插队的情况，也是和读锁一样，非公平的Sync提供了插队的可能，如果tryAcquire获取失败，就得进入AQS等待。

最后说说为什么Semaphore和ReentrantLock在tryAcquireXX方法就实现了非公平性和公平性，而ReentrantReadWriteLock却要抽象出readerShouldBlock和writerShouldBlock的方法来单独处理公平性。

abstract boolean readerShouldBlock();
abstract boolean writerShouldBlock();

ReentrantLock只支持独占模式，所以它只需要在NonfairSync和FairSync里面实现tryAcquire方法就能实现公平性和非公平性。

而ReentrantReadWriteLock即要支持共享和独占模式，又要支持公平性和非公平性，所以它在基类的Sync里面用tryAcquire和tryAcquireShared方法来区分独占和共享模式，

在NonfairSync和FairSync的readerShouldBlock和writerShouldBlock里面实现非公平性和公平性。