ReentrantReadWriteLock

发表于 分类于
可重入锁、读写锁，更新中...

ReentrantReadWriteLock

读写可重入锁

为了避免脏读

  • 当前线程拿到写锁，允许当前线程拿读锁（锁降级），不允许其他线程获取读锁/写锁
  • 当前线程拿到读锁，允许当前/其他线程拿读锁，不允许任何线程拿写锁

读锁和写锁都支持重入

支持 Nonfair（默认）和 Fair 的锁

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
private final ReentrantReadWriteLock.ReadLock readerLock;

private final ReentrantReadWriteLock.WriteLock writerLock;

final Sync sync;

// 默认非公平
public ReentrantReadWriteLock() {
    this(false);
}
public ReentrantReadWriteLock(boolean fair) {
    sync = fair ? new FairSync() : new NonfairSync();
    readerLock = new ReadLock(this);
    writerLock = new WriteLock(this);
}

获取读锁和写锁的两个方法：

1
2
public ReentrantReadWriteLock.WriteLock writeLock() { return writerLock; }
public ReentrantReadWriteLock.ReadLock  readLock()  { return readerLock; }

除此之外，还有一些用于展示工作状态的：

ReentrantReadWriteLock2.jpg

读写锁的获取次数存放在 AQS 里面的 state 上，state 的 高 16 位 存放 readLock 获取的次数，低 16 位 存放 writeLock 获取的次数

获取读状态 就是将状态值 右移 16 位

获取写状态 就是将状态值的 高 16 位抹去

1
2
3
4
5
6
7
8
9
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; }

每个线程获得读锁的次数用内部类 HoldCounter 保存，并且存储在 ThreadLocal 里面

写锁

获取

先拿到 WriteLock

1
public ReentrantReadWriteLock.WriteLock writeLock() { return writerLock; }

然后一样，调 Sync 的 acquire，因为 Sync 是 AQS 的子类，即是 AQS 的 acquire，然后就是 tryAcquire

1
2
3
public void lock() {
    sync.acquire(1);
}
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
protected final boolean tryAcquire(int acquires) {
    Thread current = Thread.currentThread();
    // 获取同步状态
    int c = getState();
    // 获取写状态，即写线程的数量
    int w = exclusiveCount(c);
    // 同步状态不为 0 即有线程持有锁
    if (c != 0) {
        // (Note: if c != 0 and w == 0 then shared count != 0)
        // c 不为 0 而 w 为 0，说明此时读锁不为 0，无法获取写锁，false
        // 或者是
        // w 不为 0，存在写锁，但是却是其他线程拿到了，false
        
        // 此时有读锁，无法获取写锁
        // 或者，存在写锁，但是却是其他线程拿到了
        if (w == 0 || current != getExclusiveOwnerThread())
            return false;
        // 如果重入数量太大，Error
        if (w + exclusiveCount(acquires) > MAX_COUNT)
            throw new Error("Maximum lock count exceeded");
        
        // Reentrant acquire
        // 重入
        setState(c + acquires);
        return true;
    }
    // 此时同步状态 c 等于 0，即没有读锁也没写锁
    // writerShouldBlock 判断要不要获取锁
    // 如果是非公平锁，直接 CAS 尝试获取锁，失败了才排队
    // 如果是公平锁，如果队列前面有在排队，就放弃，没有排队才拿锁
    if (writerShouldBlock() ||
        !compareAndSetState(c, c + acquires))
        return false;
    // 记录拥有写锁的线程
    setExclusiveOwnerThread(current);
    return true;
}

// 公平锁
final boolean writerShouldBlock() {
    return hasQueuedPredecessors();
}
// 非公平锁
final boolean writerShouldBlock() {
    return false; // writers can always barge
}

读锁

获取

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
protected final int tryAcquireShared(int unused) {

    Thread current = Thread.currentThread();
    // 拿到同步状态 c
    int c = getState();
    // 如果写锁不为 0 且 当前持有锁的线程不是当前线程(考虑锁降级)，退出
    if (exclusiveCount(c) != 0 &&
        getExclusiveOwnerThread() != current)
        return -1;
    // 读锁的线程数
    int r = sharedCount(c);
    // 公平锁看队列前面有没有在排队，有就放弃
    // 没有在排队，且读线程小于最大数量，才 CAS 拿锁
    // 非公平锁直接搞
    if (!readerShouldBlock() &&
        r < MAX_COUNT &&
        compareAndSetState(c, c + SHARED_UNIT)) {
        // CAS 如果失败了，那就
        // 如果没有读线程
        if (r == 0) {
            // holdCount 等于 1
            firstReader = current;
            firstReaderHoldCount = 1;
            // 如果有读线程，且是自己之前拿的锁，那 holdCount 就 ++
        } else if (firstReader == current) {
            // ++
            firstReaderHoldCount++;
            // 如果不是自己之前拿的锁
        } else {
            HoldCounter rh = cachedHoldCounter;
            if (rh == null || rh.tid != getThreadId(current))
                cachedHoldCounter = rh = readHolds.get();
            else if (rh.count == 0)
                readHolds.set(rh);
            rh.count++;
        }
        return 1;
    }
    return fullTryAcquireShared(current);
}

// 公平锁
final boolean readerShouldBlock() {
    return hasQueuedPredecessors();
}
// 非公平锁
final boolean readerShouldBlock() {
    return apparentlyFirstQueuedIsExclusive();
}
final boolean apparentlyFirstQueuedIsExclusive() {
    Node h, s;
    return (h = head) != null &&
        (s = h.next)  != null &&
        !s.isShared()         &&
        s.thread != null;
}
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
final int fullTryAcquireShared(Thread current) {
    /*
     * This code is in part redundant with that in
     * tryAcquireShared but is simpler overall by not
     * complicating tryAcquireShared with interactions between
     * retries and lazily reading hold counts.
     */
    HoldCounter rh = null;
    for (;;) {
        int c = getState();
        if (exclusiveCount(c) != 0) {
            if (getExclusiveOwnerThread() != current)
                return -1;
            // else we hold the exclusive lock; blocking here
            // would cause deadlock.
        } else if (readerShouldBlock()) {
            // Make sure we're not acquiring read lock reentrantly
            if (firstReader == current) {
                // assert firstReaderHoldCount > 0;
            } else {
                if (rh == null) {
                    rh = cachedHoldCounter;
                    if (rh == null || rh.tid != getThreadId(current)) {
                        rh = readHolds.get();
                        if (rh.count == 0)
                            readHolds.remove();
                    }
                }
                if (rh.count == 0)
                    return -1;
            }
        }
        if (sharedCount(c) == MAX_COUNT)
            throw new Error("Maximum lock count exceeded");
        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; // cache for release
            }
            return 1;
        }
    }
}

释放

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
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;
        if (compareAndSetState(c, nextc))
            // Releasing the read lock has no effect on readers,
            // but it may allow waiting writers to proceed if
            // both read and write locks are now free.
            return nextc == 0;
    }
}

锁降级

锁降级是指把持住（当前拥有的）写锁，再获取到读锁，随后释放（先前拥有的）写锁的过程

先释放写锁再获取读锁这种分段的不能称为锁降级

主要是为了保证数据的可见性，如果当前线程不获取读锁而直接释放写锁，假设此刻另一个线程（T）获取了写锁并修改了数据，那么当前线程是无法感知线程 T 的数据更新。如果当前线程获取读锁，即遵循锁降级的步骤，则线程T将会被阻塞，知道当前线程使用数据并释放读锁之后，线程T才能获取写锁进行数据更新。

示例：

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
class CachedData {
    Object data;
    volatile boolean cacheValid;
	final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();

	void processCachedData() {
		rwl.readLock().lock();
		if (!cacheValid) {
			// 必须先释放读锁
			rwl.readLock().unlock();
            // 锁降级从写锁获取到时开始
			rwl.writeLock().lock();
            try {
          		// Recheck state because another thread might have
          		// acquired write lock and changed state before we did.
          		if (!cacheValid) {
            		data = ...
            		cacheValid = true;
          		}
          		// Downgrade by acquiring read lock before releasing write lock
          		rwl.readLock().lock();
        	} finally {
          		rwl.writeLock().unlock(); // Unlock write, still hold read
        	}
     	}

		try {
       		use(data);
     	} finally {
       		rwl.readLock().unlock();
     	}
   	}
}