共享锁代码分析

参考代码
ReentrantReadWriteLock

ReadLock中的sync是一个Sync对象，Sync继承于AQS类，即Sync就是一个锁。ReentrantReadWriteLock中也有一个Sync对象，而且ReadLock中的sync和ReentrantReadWriteLock中的sync是对应关系。即ReentrantReadWriteLock和ReadLock共享同一个AQS对象，共享同一把锁。

获取共享锁

通过tryAcquireShared()尝试获取共享锁。尝试成功的话，则直接返回；尝试失败的话，则通过doAcquireShared()不断的循环并尝试获取锁，若有需要，则阻塞等待。

lock

ReadLock.java

1
2
3

public void lock() {
    sync.acquireShared(1);
}

acquireShared

Sync继承于AQS，acquireShared()定义在AQS中。源码如下：

public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}

通过tryAcquireShared()来尝试获取锁
成功的话，则不再做任何动作
失败的话，则通过doAcquireShared()来获取锁。doAcquireShared()会获取到锁了才返回

tryAcquireShared()

试获取锁

protected final int tryAcquireShared(int unused) {
    Thread current = Thread.currentThread();
    // 获取“锁”的状态
    int c = getState();
    // 如果“锁”是“互斥锁”，并且获取锁的线程不是current线程；则返回-1。
    if (exclusiveCount(c) != 0 &&
        getExclusiveOwnerThread() != current)
        return -1;
    // 获取“读取锁”的共享计数
    int r = sharedCount(c);
    // 如果“不需要阻塞等待”，并且“读取锁”的共享计数小于MAX_COUNT；
    // 则通过CAS函数更新“锁的状态”，将“读取锁”的共享计数+1。
    if (!readerShouldBlock() &&
        r < MAX_COUNT &&
        compareAndSetState(c, c + SHARED_UNIT)) {
        // 第1次获取“读取锁”。
        if (r == 0) { 
            firstReader = current;
            firstReaderHoldCount = 1;
        // 如果想要获取锁的线程(current)是第1个获取锁(firstReader)的线程
        } else if (firstReader == current) { 
            firstReaderHoldCount++;
        } else {
            // HoldCounter是用来统计该线程获取“读取锁”的次数。
            HoldCounter rh = cachedHoldCounter;
            if (rh == null || rh.tid != current.getId())
                cachedHoldCounter = rh = readHolds.get();
            else if (rh.count == 0)
                readHolds.set(rh);
            // 将该线程获取“读取锁”的次数+1。
            rh.count++;
        }
        return 1;
    }
    return fullTryAcquireShared(current);
}

exclusiveCount,sharedCount 计算锁的数量。

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

如果从二进制来看EXCLUSIVE_MASK的表示，这个值的低16位全是1，而高16位则全是0，所以exclusiveCount是把state的低16位取出来，表示当前这个锁的写锁加锁次数。

再来看sharedCount，取出了state的高16位，用来表示这个锁的读锁加锁次数。
也就是说state的高16位和低16位来分别表示这个锁上的读锁和写锁的加锁次数。
看源代码
compareAndSetState(c, c + SHARED_UNIT))，将“读取锁”的共享计数+1。SHARED_UNIT正好是高16位的加1操作。
HoldCounter是线程变量，用来统计该线程获取“读取锁”的次数。

说明：
tryAcquireShared()的作用是尝试获取“共享锁”。
如果在尝试获取锁时，“不需要阻塞等待”并且“读取锁的共享计数小于MAX_COUNT”，则直接通过CAS函数更新“读取锁的共享计数”，以及将“当前线程获取读取锁的次数+1”。
否则，通过fullTryAcquireShared()获取读取锁。

fullTryAcquireShared

final int fullTryAcquireShared(Thread current) {
    HoldCounter rh = null;
    for (;;) {
        // 获取“锁”的状态
        int c = getState();
        // 如果“锁”是“互斥锁”，并且获取锁的线程不是current线程；则返回-1。
        if (exclusiveCount(c) != 0) {
            if (getExclusiveOwnerThread() != current)
                return -1;
        // 如果“需要阻塞等待”。
        // (01) 当“需要阻塞等待”的线程是第1个获取锁的线程的话，则继续往下执行。
        // (02) 当“需要阻塞等待”的线程获取锁的次数=0时，则返回-1。
        } else if (readerShouldBlock()) {
            // 如果想要获取锁的线程(current)是第1个获取锁(firstReader)的线程
            if (firstReader == current) {
            } else {
                if (rh == null) {
                    rh = cachedHoldCounter;
                    if (rh == null || rh.tid != current.getId()) {
                        rh = readHolds.get();
                        if (rh.count == 0)
                            readHolds.remove();
                    }
                }
                // 如果当前线程获取锁的计数=0,则返回-1。
                if (rh.count == 0)
                    return -1;
            }
        }
        // 如果“不需要阻塞等待”，则获取“读取锁”的共享统计数；
        // 如果共享统计数超过MAX_COUNT，则抛出异常。
        if (sharedCount(c) == MAX_COUNT)
            throw new Error("Maximum lock count exceeded");
        // 将线程获取“读取锁”的次数+1。
        if (compareAndSetState(c, c + SHARED_UNIT)) {
            // 如果是第1次获取“读取锁”，则更新firstReader和firstReaderHoldCount。
            if (sharedCount(c) == 0) {
                firstReader = current;
                firstReaderHoldCount = 1;
            // 如果想要获取锁的线程(current)是第1个获取锁(firstReader)的线程，
            // 则将firstReaderHoldCount+1。
            } else if (firstReader == current) {
                firstReaderHoldCount++;
            } else {
                if (rh == null)
                    rh = cachedHoldCounter;
                if (rh == null || rh.tid != current.getId())
                    rh = readHolds.get();
                else if (rh.count == 0)
                    readHolds.set(rh);
                // 更新线程的获取“读取锁”的共享计数
                rh.count++;
                cachedHoldCounter = rh; // cache for release
            }
            return 1;
        }
    }
}

doAcquireShared()

private void doAcquireShared(int arg) {
    // addWaiter(Node.SHARED)的作用是，创建“当前线程”对应的节点，并将该线程添加到CLH队列中。
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            // 获取“node”的前一节点
            final Node p = node.predecessor();
            // 如果“当前线程”是CLH队列的表头，则尝试获取共享锁。
            if (p == head) {
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    if (interrupted)
                        selfInterrupt();
                    failed = false;
                    return;
                }
            }
            // 如果“当前线程”不是CLH队列的表头，则通过shouldParkAfterFailedAcquire()判断是否需要等待，
            // 需要的话，则通过parkAndCheckInterrupt()进行阻塞等待。若阻塞等待过程中，线程被中断过，则设置interrupted为true。
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

说明：doAcquireShared()的作用是获取共享锁。
它会首先创建线程对应的CLH队列的节点，然后将该节点添加到CLH队列中。CLH队列是管理获取锁的等待线程的队列。
如果“当前线程”是CLH队列的表头，则尝试获取共享锁；否则，则需要通过shouldParkAfterFailedAcquire()判断是否阻塞等待，需要的话，则通过parkAndCheckInterrupt()进行阻塞等待。
doAcquireShared()会通过for循环，不断的进行上面的操作；目的就是获取共享锁。

专注于快乐的事情

共享锁分析