Java技术之ReentrantReadWriteLock源码解析

Java技术 ReentrantReadWriteLock 评论

1. 阅读须知

  • JDK版本：1.8

  • 文章中使用/**/注释的方法会做深入分析

2. 正文

ReentrantReadWriteLock，从字面上理解为可重入读写锁，基于AQS（AbstractQueuedSynchronizer，不了解AQS的读者可以去看笔者关于AQS源码解析的文章进行学习）实现，根据读写锁的特性，我们可以猜测，读锁应该是基于AQS的共享锁实现，而写锁应该是基于AQS的独占锁实现，我们来验证这个猜想，首先看一下ReentrantReadWriteLock的构造方法：

ReentrantReadWriteLock：

1
2
3
4
5
6
public ReentrantReadWriteLock(boolean fair) {
    //根据传入的boolean变量fair来确定使用公平锁或非公平锁
    sync = fair ? new FairSync() : new NonfairSync();
    readerLock = new ReadLock(this);
    writerLock = new WriteLock(this);
}

ReentrantReadWriteLock默认的无参构造方法使用的是非公平锁。我们来介绍一下ReentrantReadWriteLock中的同步器Sync中的一些变量：

ReentrantReadWriteLock.Sync：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
//读锁占用高16位表示持有读锁的线程的数量
static final int SHARED_SHIFT   = 16;
//根据SHARED_SHIFT变量的含义，每增加一个持有读锁的线程，state变量就需要累加这个值，也就是1左移16位
static final int SHARED_UNIT    = (1 << SHARED_SHIFT);
//持有读锁的线程的最大数量（65535）
static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1;
//用于计算写锁的重入计数
static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;
//当前线程持有的读锁的重入数量。只在构造函数和readObject方法中初始化。线程的锁重入计数降至0时删除
private transient ThreadLocalHoldCounter readHolds;
//最后一个成功获取readLock的线程的持有锁计数
private transient HoldCounter cachedHoldCounter;
//firstReader是获取读锁的第一个线程。
//更确切地说，firstReader是最后一次将共享计数从0更改为1的唯一线程，
//并且自那以后未释放读锁; 如果没有这样的线程，则返回null。
private transient Thread firstReader = null;
//firstReaderHoldCount是firstReader的锁重入计数
private transient int firstReaderHoldCount;

ReentrantReadWriteLock使用AQS的state的高16位表示持有读锁的线程的数量，低16位表示写锁被同一个线程申请的次数，也就是锁重入的次数。接下来我们来看加锁实现，我们首先来看读锁部分：

ReentrantReadWriteLock.ReadLock：

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

acquireShared方法我们在AQS（AbstractQueuedSynchronizer）源码解析（共享锁部分）这篇文章中进行过详细分析，方法的开始会调用有子类实现的tryAcquireShared方法尝试以共享模式获得锁，我们来看ReentrantReadWriteLock对tryAcquireShared方法的实现：

ReentrantReadWriteLock.Sync：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
protected final int tryAcquireShared(int unused) {
    Thread current = Thread.currentThread();
    int c = getState();
    //如果独占锁的重入计数不为0（说明有线程持有独占锁）并且持有独占锁的线程不是当前线程返回-1代表获取共享锁失败
    if (exclusiveCount(c) != 0 &&
        getExclusiveOwnerThread() != current)
        return -1;
    int r = sharedCount(c); //共享锁的持有线程数量
    /*判断当前获取读锁的线程是否需要阻塞*/
    //共享锁的持有线程的数量是否超过了最大值
    //CAS增加共享锁的持有线程的数量是否成功
    if (!readerShouldBlock() &&
        r < MAX_COUNT &&
        compareAndSetState(c, c + SHARED_UNIT)) {
        if (r == 0) {
            //满足条件说明当前没有任何任何线程持有共享锁，则将当前线程设置为获取共享锁的第一个线程
            firstReader = current;
            //锁重入数量初始化为1
            firstReaderHoldCount = 1;
        } else if (firstReader == current) {
            //如果当前获取共享锁的线程是获取共享锁的第一个线程，则递增锁重入数量
            firstReaderHoldCount++;
        } else {
            HoldCounter rh = cachedHoldCounter;
            if (rh == null || rh.tid != getThreadId(current))
                //如果最后一个成功获取readLock的线程的锁重入计数对象还未初始化或者对象内部维护的线程id不是当前线程id
                //则将cachedHoldCounter赋值为当前线程的锁重入计数对象
                cachedHoldCounter = rh = readHolds.get();
            else if (rh.count == 0)
                readHolds.set(rh);
            rh.count++; //递增当前线程的锁重入计数
        }
        return 1;
    }
    /*tryAcquireShared方法的完整版*/
    return fullTryAcquireShared(current);
}

这里说明一下方法的执行过程：

    1. 如果另一个线程持有写锁，则获取共享锁失败。
    1. 否则，此线程符合锁定状态，判断是否应该因为队列策略而阻塞。如果没有，尝试通过CAS增加共享锁的持有线程的数量。请注意，这步不会检查重入获取，它会被推迟到fullTryAcquireShared方法执行，以避免在更典型的非重入情况下检查锁重入计数。
    1. 如果步骤2因线程需要阻塞或CAS失败或计数饱和而失败，则调用fullTryAcquireShared方法。 关于readerShouldBlock（判断当前获取读锁的线程是否需要阻塞）方法，有公平和非公平两种实现，我们首先来看非公平的实现：

ReentrantReadWriteLock.NonfairSync：

1
2
3
final boolean readerShouldBlock() {
    return apparentlyFirstQueuedIsExclusive();
}

apparentlyFirstQueuedIsExclusive方法来自AQS，主要用于判断等待队列的头结点的下一个节点也就是第一个排队的线程是否以独占模式等待。这里我们要结合调用readerShouldBlock方法之前的if判断进行分析，如果这个if判断不满足，说明有两种情况可能发生：

    1. 当前没有线程占用写锁，这种情况readerShouldBlock方法会返回false。
    1. 当前有线程占用写锁，并且占用写锁的线程就是当前线程（当前线程是head节点），这时就发生了锁降级的情况，也就是当前线程持有写锁，并在申请读锁，这时就要判断head节点的下一个节点是否要申请写锁，如果是则readerShouldBlock方法返回true，说明本次申请读锁的操作需要阻塞。

接下来我们来看readerShouldBlock方法公平锁的实现：

ReentrantReadWriteLock.FairSync：

1
2
3
final boolean readerShouldBlock() {
    return hasQueuedPredecessors();
}

hasQueuedPredecessors方法我们在ReentrantLock源码解析这篇文章中分析过，它的主要作用是确认当前线程是否是下一个能够优先获得锁的线程，公平性也就是通过这个判断来保证的。公平锁我们很好理解，就是根据等待队列中节点的顺序来保证获取锁的顺序。

ReentrantReadWriteLock.Sync：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
final int fullTryAcquireShared(Thread current) {
    HoldCounter rh = null;
    for (;;) {
        int c = getState();
        //同样的判断是否有非当前线程持有独占锁
        if (exclusiveCount(c) != 0) {
            if (getExclusiveOwnerThread() != current)
                return -1;
        //同样的判断当前获取读锁操作是否需要阻塞
        } else if (readerShouldBlock()) {
            //确保没有重复获取读锁
            if (firstReader == current) {
            } else {
                if (rh == null) {
                    rh = cachedHoldCounter;
                    if (rh == null || rh.tid != getThreadId(current)) {
                        rh = readHolds.get();
                        //这里如果当前线程持有的共享锁重入计数为0，则移除锁重入计数对象
                        if (rh.count == 0)
                            readHolds.remove();
                    }
                }
                if (rh.count == 0)
                    return -1; //锁重入计数为0时，返回-1代表获取共享锁失败，要进行排队
            }
        }
        if (sharedCount(c) == MAX_COUNT)
            //超过最大持有读锁线程数量抛出Error
            throw new Error("Maximum lock count exceeded");
        //再次尝试获取共享锁，判断CAS增加共享锁的持有线程的数量是否成功
        //整体共享锁获取成功的处理逻辑与tryAcquireShared方法基本一致
        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;
            }
            return 1;
        }
    }
}

接下来我们来看读锁释放的实现：

ReentrantReadWriteLock.ReadLock：

1
2
3
public void unlock() {
    sync.releaseShared(1);
}

这里的releaseShared释放共享锁方法同样来自于AQS，方法中首先会调用由子类覆盖的tryReleaseShared方法，通过尝试设置state变量来释放共享锁，我们来看ReentrantReadWriteLock对于tryReleaseShared方法的实现：

ReentrantReadWriteLock.Sync：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
protected final boolean tryReleaseShared(int unused) {
    Thread current = Thread.currentThread();
    //判断当前线程是否是获取读锁的第一个线程
    if (firstReader == current) {
        //如果锁重入计数为1，直接将获取读锁的第一个线程置为null，释放资源
        if (firstReaderHoldCount == 1)
            firstReader = null;
        else
            //如果锁重入计数不为1（大于1），则在释放时递减锁重入计数
            firstReaderHoldCount--;
    } else {
        HoldCounter rh = cachedHoldCounter;
        //这里的判断上文分析过
        if (rh == null || rh.tid != getThreadId(current))
            rh = readHolds.get();
        int count = rh.count; //当前线程的锁重入计数
        if (count <= 1) {
            //小于等于1时说明这是最后一个重入锁，则移除锁重入计数对象
            readHolds.remove();
            if (count <= 0)
                //锁重入计数小于等于0说明本次解锁操作没有对应的加锁操作，抛出异常
                throw unmatchedUnlockException();
        }
        --rh.count; //递减锁重入计数
    }
    //下面的自旋操作为递减共享锁持有的线程数量，与加锁时的递增操作正好相反
    for (;;) {
        int c = getState();
        int nextc = c - SHARED_UNIT;
        if (compareAndSetState(c, nextc))
            return nextc == 0;
    }
}

到这里，读锁的加锁和解锁操作就分析完了，下面我们来分析写锁的加锁和解锁操作，首先来看加锁：

ReentrantReadWriteLock.WriteLock：

1
2
3
public void lock() {
    sync.acquire(1);
}

果然，写锁是基于AQS的独占锁实现，这里的acquire方法我们在AQS（AbstractQueuedSynchronizer）源码解析（独占锁部分）这篇文章中已经详细分析过，方法的第一步就是调用由子类实现的tryAcquire方法通过操作state变量尝试以独占模式获取锁，我们来看ReentrantReadWriteLock对tryAcquire方法的实现：

ReentrantReadWriteLock.Sync：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
protected final boolean tryAcquire(int acquires) {
    Thread current = Thread.currentThread();
    int c = getState();
    int w = exclusiveCount(c);
    //如果AQS的state变量不为0，说明当前读锁或写锁有被占用
    if (c != 0) {
        //这里的判断如果成立说明读锁被占用写锁未被占用
        //或者写锁被占用但占用的线程不是当前线程，这是返回false代表获取写锁失败
        if (w == 0 || current != getExclusiveOwnerThread())
            return false;
        //写锁最大重入数量的判断
        if (w + exclusiveCount(acquires) > MAX_COUNT)
            throw new Error("Maximum lock count exceeded");
        //走到这里说明是写锁重入，则递增写锁重入计数
        setState(c + acquires);
        return true;
    }
    //到这里说明当前读锁和写锁都未被任何线程占用
    /*判断获取写锁的线程是否需要阻塞*/
    //判断CAS递增写锁重入计数是否失败
    if (writerShouldBlock() ||
        !compareAndSetState(c, c + acquires))
        return false;
    //设置写锁的拥有者线程为当前线程
    setExclusiveOwnerThread(current);
    return true;
}

这里的writerShouldBlock方法同样区分公平和非公平两个版本的实现，我们先来看非公平版本的实现：

ReentrantReadWriteLock.NonfairSync：

1
2
3
final boolean writerShouldBlock() {
    return false;
}

我们发现方法直接返回false，也就是说每个想要获取非公平写锁的线程都可以直接参与竞争。而writerShouldBlock方法公平锁的版本与读锁的readerShouldBlock方法的公平版本是一样的，都是需要确认当前线程是否是下一个能够优先获得锁的线程，以此来保证公平性。最后我们来看写锁的解锁操作：

ReentrantReadWriteLock.WriteLock：

1
2
3
public void unlock() {
    sync.release(1);
}

这里的release方法我们在AQS独占锁源码解析的文章中同样进行过详细的分析，AQS的release方法首先会尝试调用由子类实现的tryRelease方法来尝试设置state变量来释放独占锁，锁完全释放后，会对后继节点进行唤醒操作，这个流程我们已经分析过，不再赘述。我们来看ReentrantReadWriteLock的对tryRelease方法的实现：

ReentrantReadWriteLock.WriteLock：

1
2
3
4
5
6
7
8
9
10
11
12
13
protected final boolean tryRelease(int releases) {
    //加锁和解锁的线程必须是同一个，不然抛出异常
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    //递减写锁的重入计数
    int nextc = getState() - releases;
    boolean free = exclusiveCount(nextc) == 0;
    //如果递减后的锁重入计数为0，说明锁已经被完全释放，这时将锁的拥有者线程置为null
    if (free)
        setExclusiveOwnerThread(null);
    setState(nextc); //设置最新的锁重入计数
    return free;
}

这样解锁的流程就分析完成了。

ReentrantReadWriteLock的写锁还支持Condition，与ReentrantLock一样完全基于AQS的ConditionObject实现，我们已经分析过ConditionObject源码，不明白的同学可以前往进行查阅学习。到这里，ReentrantReadWriteLock的源码分析就完成了。

本文出自

波哥Java Developer & Designer

个人简介。