Java并发-ReentrantReadWriteLock

发表于 | 评论数： | 热度： ℃   |   字数统计: 3k字   |   阅读时长: ≈ 12分钟

概述

ReadWriteLock，顾名思义，是读写锁。它维护了一对相关的锁 — — “读取锁”和“写入锁”，一个用于读取操作，另一个用于写入操作。

  • “读取锁”用于只读操作，它是“共享锁”，能同时被多个线程获取。
  • “写入锁”用于写入操作，它是“独占锁”，写入锁只能被一个线程锁获取。
    注意：不能同时存在读取锁和写入锁！
    ReadWriteLock是一个接口。ReentrantReadWriteLock是它的实现类，ReentrantReadWriteLock包括子类ReadLock和WriteLock。
    读写锁底层也是通过AQS框架实现的，通过之前的文章可知，如果使用AQS框架，在这个类内部必定有一个AQS框架的子类，AQS子类无非就是实现如下几点：

    第一点：对共享资源state进行操作
    第二点：如果是独占模式下对资源的获取和释放，那么必须实现tryAcquire()方法和tryRelease()方法。
    第三点：如果是共享模式下对资源的获取和释放，那么必须实现tryAcquireShared()方法和tryReleaseShared()方法。

而读写锁ReentrantReadWriteLock即包含独占模式下资源的获取和释放(写锁)，也包含共享模式下资源的获取和释放(读锁)，所以需要上面的四个方法都需要实现。

用state一个整形变量怎样去维护读和写两种状态

共享资源state是一个整形变量，在读写锁中又必须维护读和写两种状态，所以只能按位把state分割成两个部分，一个整形有32位，高16位代表读，低16位代表写。如图所示：
![img][link1]

1
2
3
4
5
6
7
8
9
10
11
12
//读状态占用位数
static final int SHARED_SHIFT   = 16;
//每一次增加的读状态：(1<<16),因为读状态是state的高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;
//计算读锁的数量
static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }
//计算写锁的数量
static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }

读操作，因为高16位为读操作，要想获取读的数量，只需将state无符号右移16位即可：state>>>16。读最大的数量：65535(二进制表示：11111111110000000000000000)

写操作，因为低16位为写操作，要向获取写数量，只需要把state的高16位抹去就可以了，很容易想到“按位与”操作：state&1111111111111111。写最大的数量：65535(二进制表示：00000000000000001111111111111111)

共享锁

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
public static class ReadLock implements Lock, java.io.Serializable {
    private static final long serialVersionUID = -5992448646407690164L;
    // ReentrantReadWriteLock的AQS对象
    private final Sync sync;

    protected ReadLock(ReentrantReadWriteLock lock) {
        sync = lock.sync;
    }

    // 获取“共享锁”
    public void lock() {
        sync.acquireShared(1);
    }

    // 如果线程是中断状态，则抛出一场，否则尝试获取共享锁。
    public void lockInterruptibly() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }

    // 尝试获取“共享锁”
    public  boolean tryLock() {
        return sync.tryReadLock();
    }

    // 在指定时间内，尝试获取“共享锁”
    public boolean tryLock(long timeout, TimeUnit unit)
            throws InterruptedException {
        return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
    }

    // 释放“共享锁”
    public  void unlock() {
        sync.releaseShared(1);
    }

    // 新建条件
    public Condition newCondition() {
        throw new UnsupportedOperationException();
    }

    public String toString() {
        int r = sync.getReadLockCount();
        return super.toString() +
            "[Read locks = " + r + "]";
    }
}

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

获取共享锁

获取共享锁的思想(即lock函数的步骤)，是先通过tryAcquireShared()尝试获取共享锁。尝试成功的话，则直接返回；尝试失败的话，则通过doAcquireShared()不断的循环并尝试获取锁，若有需要，则阻塞等待。doAcquireShared()在循环中每次尝试获取锁时，都是通过tryAcquireShared()来进行尝试的。下面看看“获取共享锁”的详细流程。

lock()

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

acquireShared()

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

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

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

tryAcquireShared()

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

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

fullTryAcquireShared()

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
52
53
54
55
56
57
58
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;
        }
    }
}

fullTryAcquireShared()会根据“是否需要阻塞等待”，“读取锁的共享计数是否超过限制”等等进行处理。如果不需要阻塞等待，并且锁的共享计数没有超过限制，则通过CAS尝试获取锁，并返回1。

doAcquireShared()

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
private void doAcquireShared(int arg) {
    // addWaiter(Node.SHARED)的作用是，创建“当前线程”对应的结点，并将该线程添加到同步队列中。
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            // 获取“node”的前一结点
            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;
                }
            }
            // 如果“当前线程”不是同步队列的表头，则通过shouldParkAfterFailedAcquire()判断是否需要等待，
            // 需要的话，则通过parkAndCheckInterrupt()进行阻塞等待。若阻塞等待过程中，线程被中断过，则设置interrupted为true。
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

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

释放共享锁

释放共享锁的思想，是先通过tryReleaseShared()尝试释放共享锁。尝试成功的话，则通过doReleaseShared()唤醒“其他等待获取共享锁的线程”，并返回true；否则的话，返回flase。

unlock()

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

该函数实际上调用releaseShared(1)释放共享锁。

releaseShared()

1
2
3
4
5
6
7
public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

releaseShared()的目的是让当前线程释放它所持有的共享锁。
它首先会通过tryReleaseShared()去尝试释放共享锁。尝试成功，则直接返回；尝试失败，则通过doReleaseShared()去释放共享锁。

tryReleaseShared()

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
protected final boolean tryReleaseShared(int unused) {
    // 获取当前线程，即释放共享锁的线程。
    Thread current = Thread.currentThread();
    // 如果想要释放锁的线程(current)是第1个获取锁(firstReader)的线程，
    // 并且“第1个获取锁的线程获取锁的次数”=1，则设置firstReader为null；
    // 否则，将“第1个获取锁的线程的获取次数”-1。
    if (firstReader == current) {
        // assert firstReaderHoldCount > 0;
        if (firstReaderHoldCount == 1)
            firstReader = null;
        else
            firstReaderHoldCount--;
    // 获取rh对象，并更新“当前线程获取锁的信息”。
    } else {
 
        HoldCounter rh = cachedHoldCounter;
        if (rh == null || rh.tid != current.getId())
            rh = readHolds.get();
        int count = rh.count;
        if (count <= 1) {
            readHolds.remove();
            if (count <= 0)
                throw unmatchedUnlockException();
        }
        --rh.count;
    }
    for (;;) {
        // 获取锁的状态
        int c = getState();
        // 将锁的获取次数-1。
        int nextc = c - SHARED_UNIT;
        // 通过CAS更新锁的状态。
        if (compareAndSetState(c, nextc))
            return nextc == 0;
    }
}

tryReleaseShared()的作用是尝试释放共享锁。

doReleaseShared()

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
private void doReleaseShared() {
    for (;;) {
        // 获取CLH队列的头节点
        Node h = head;
        // 如果头节点不为null，并且头节点不等于tail节点。
        if (h != null && h != tail) {
            // 获取头节点对应的线程的状态
            int ws = h.waitStatus;
            // 如果头节点对应的线程是SIGNAL状态，则意味着“头节点的下一个节点所对应的线程”需要被unpark唤醒。
            if (ws == Node.SIGNAL) {
                // 设置“头节点对应的线程状态”为空状态。失败的话，则继续循环。
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;
                // 唤醒“头节点的下一个节点所对应的线程”。
                unparkSuccessor(h);
            }
            // 如果头节点对应的线程是空状态，则设置“文件点对应的线程所拥有的共享锁”为其它线程获取锁的空状态。
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        // 如果头节点发生变化，则继续循环。否则，退出循环。
        if (h == head)                   // loop if head changed
            break;
    }
}

doReleaseShared()会释放“共享锁”。它会从前往后的遍历同步队列，依次“唤醒”然后“执行”队列中每个节点对应的线程；最终的目的是让这些线程释放它们所持有的锁。

公平/非公平共享锁

和互斥锁ReentrantLock一样，ReadLock也分为公平锁和非公平锁。

公平锁和非公平锁的区别，体现在判断是否需要阻塞的函数readerShouldBlock()是不同的。

公平锁的readerShouldBlock()的源码如下：

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

在公平共享锁中，如果在当前线程的前面有其他线程在等待获取共享锁，则返回true；否则，返回false。

非公平锁的readerShouldBlock()的源码如下：

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

在非公平共享锁中，它会无视当前线程的前面是否有其他线程在等待获取共享锁。只要该非公平共享锁对应的线程不为null，则返回true。

本文标题:Java并发-ReentrantReadWriteLock

文章作者:王洪博

发布时间:2019年03月13日 - 15:03

最后更新:2019年09月12日 - 10:09

原始链接:http://whb1990.github.io/posts/b29432a8.html

▄︻┻═┳一如果你喜欢这篇文章，请点击下方"打赏"按钮请我喝杯 ☕
0%