读写锁的使用
读写锁：
独占锁：指的该锁一次只能一个线程所持有，对 ReentranLock 和 Synchronized 都是独占锁
共享锁： 指的是该锁可以被多个线程持有
ReentrantReadWriteLock 其读锁是共享锁，写锁是独占锁
作用： 多个线程同时读取一个资源类没有任何的问题，为了满足并发量，读取资源应该是同时进行的，但是如果一个线程想去写共享资源，就不应该有其他的资源可以对其进行读或者写。
使用规则：
加锁解锁格式：
1
2
3
4
5
6
7
r.lock();
try {
// 临界区
} finally {
r.unlock();
}读 - 读能共存、读 - 写不能共存、写 - 写不能共存
读锁不支持条件变量
重入时升级不支持：持有读锁的情况下去获取写锁会导致获取写锁永久等待，需要先释放读，再去获得写
重入时降级支持：持有写锁的情况下去获取读锁，造成只有当前线程会持有读锁，因为写锁会互斥其他的锁
1
2
3
4
5
6
7
8
9
10
11
12
w.lock();
try {
r.lock();// 降级为读锁, 释放写锁, 这样能够让其它线程读取缓存
try {
// ...
} finally{
w.unlock();// 要在写锁释放之前获取读锁
}
} finally{
r.unlock();
}
构造方法：
public ReentrantReadWriteLock()：默认构造方法，非公平锁public ReentrantReadWriteLock(boolean fair)：true 为公平锁
常用 API：
public ReentrantReadWriteLock.ReadLock readLock()：返回读锁public ReentrantReadWriteLock.WriteLock writeLock()：返回写锁public void lock()：加锁public void unlock()：解锁public boolean tryLock()：尝试获取锁
1 |
|
读 - 读 可并发
测试 读锁-读锁 可以并发
1 |
|
输出结果，从这里可以看到 Thread-0 锁定期间，Thread-1 的读操作不受影响
1 |
|
读 - 写 / 写 - 写 互斥
测试 读锁-写锁 相互阻塞
1 |
|
输出结果
1 | JAVA |
写锁-写锁 也是相互阻塞的，
注意事项
- 读锁不支持条件变量，写锁支持
- 重入时不支持升级：即持有读锁的情况下去获取写锁，会导致获取写锁永久等待
1 |
|
- 重入时支持降级：即持有写锁的情况下去获取读锁
1 |
|
读写锁应用：
缓存应用：
缓存更新时候，是先清除缓存还是先更新数据库，
先清理缓存：可能会造成刚清理缓存还没有更新数据库，线程直接查询了数据库更新过期数据到缓存
先更新数据库： 可能造成刚刚更新数据库，还没有更新缓存就有线程从缓存中拿到了旧的数据
补充情况：查询线程 A 查询数据时，恰好缓存数据由于时间到期，或者是第一次查询
先清缓存
先更新数据库
补充一种情况，假设查询线程 A 查询数据时恰好缓存数据由于时间到期失效，或是第一次查询
样例：
1 |
|
- 以上实现体现的是读写锁的应用，保证缓存和数据库的一致性，但有下面的问题没有考虑
- 适合读多写少，如果写操作比较频繁，以上实现性能低
- 没有考虑缓存容量
- 没有考虑缓存过期
- 只适合单机
- 并发性还是低，目前只会用一把锁
- 更新方法太过简单粗暴，清空了所有 key（考虑按类型分区或重新设计 key）
- 乐观锁实现：用 CAS 去更新
实现原理
成员属性
读写锁用的是同一个 Sycn 同步器，因此等待队列、state 等也是同一个，原理与 ReentrantLock 加锁相比没有特殊之处，不同是写锁状态占了 state 的低 16 位，而读锁使用的是 state 的高 16 位
读写锁：
1
2
3
private final ReentrantReadWriteLock.ReadLock readerLock;
private final ReentrantReadWriteLock.WriteLock writerLock;构造方法：默认是非公平锁，可以指定参数创建公平锁
1
2
3
4
5
6
7
8
public ReentrantReadWriteLock(boolean fair) {
// true 为公平锁
sync = fair ? new FairSync() : new NonfairSync();
// 这两个 lock 共享同一个 sync 实例，都是由 ReentrantReadWriteLock 的 sync 提供同步实现
readerLock = new ReadLock(this);
writerLock = new WriteLock(this);
}
Sync 类的属性：
统计变量：
1
2
3
4
5
6
7
8
9
// 用来移位
static final int SHARED_SHIFT = 16;
// 高16位的1
static final int SHARED_UNIT = (1 << SHARED_SHIFT);
// 65535，16个1，代表写锁的最大重入次数
static final int MAX_COUNT = (1 << SHARED_SHIFT) - 1;
// 低16位掩码：0b 1111 1111 1111 1111，用来获取写锁重入的次数
static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;获取读写锁的次数：
1
2
3
4
5
// 获取读写锁的读锁分配的总次数
static int sharedCount(int c) { return c >>> SHARED_SHIFT; }
// 写锁（独占）锁的重入次数
static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }内部类：
1
2
3
4
5
6
7
8
9
10
11
12
13
// 记录读锁线程自己的持有读锁的数量（重入次数），因为 state 高16位记录的是全局范围内所有的读线程获取读锁的总量
static final class HoldCounter {
int count = 0;
// Use id, not reference, to avoid garbage retention
final long tid = getThreadId(Thread.currentThread());
}
// 线程安全的存放线程各自的 HoldCounter 对象
static final class ThreadLocalHoldCounter extends ThreadLocal<HoldCounter> {
public HoldCounter initialValue() {
return new HoldCounter();
}
}内部类实例：
1
2
3
4
5
// 当前线程持有的可重入读锁的数量，计数为 0 时删除
private transient ThreadLocalHoldCounter readHolds;
// 记录最后一个获取【读锁】线程的 HoldCounter 对象
private transient HoldCounter cachedHoldCounter;首次获取锁：
1
2
3
4
5
// 第一个获取读锁的线程
private transient Thread firstReader = null;
// 记录该线程持有的读锁次数（读锁重入次数）
private transient int firstReaderHoldCount;Sync 构造方法：
1
2
3
4
5
6
Sync() {
readHolds = new ThreadLocalHoldCounter();
// 确保其他线程的数据可见性，state 是 volatile 修饰的变量，重写该值会将线程本地缓存数据【同步至主存】
setState(getState());
}
加锁原理
t1 线程：w.lock（写锁），成功上锁 state = 0_1
1
2
3
4
5
6
7
8
9
10
// lock() -> sync.acquire(1);
public void lock() {
sync.acquire(1);
}
public final void acquire(int arg) {
// 尝试获得写锁，获得写锁失败，将当前线程关联到一个 Node 对象上, 模式为独占模式
if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}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 tryAcquire(int acquires) {
Thread current = Thread.currentThread();
int c = getState();
// 获得低 16 位, 代表写锁的 state 计数
int w = exclusiveCount(c);
// 说明有读锁或者写锁
if (c != 0) {
// c != 0 and w == 0 表示有读锁，【读锁不能升级】，直接返回 false
// w != 0 说明有写锁，写锁的拥有者不是自己，获取失败
if (w == 0 || current != getExclusiveOwnerThread())
return false;
// 执行到这里只有一种情况：【写锁重入】，所以下面几行代码不存在并发
if (w + exclusiveCount(acquires) > MAX_COUNT)
throw new Error("Maximum lock count exceeded");
// 写锁重入, 获得锁成功，没有并发，所以不使用 CAS
setState(c + acquires);
return true;
}
// c == 0，说明没有任何锁，判断写锁是否该阻塞，是 false 就尝试获取锁，失败返回 false
if (writerShouldBlock() || !compareAndSetState(c, c + acquires))
return false;
// 获得锁成功，设置锁的持有线程为当前线程
setExclusiveOwnerThread(current);
return true;
}
// 非公平锁 writerShouldBlock 总是返回 false, 无需阻塞
final boolean writerShouldBlock() {
return false;
}
// 公平锁会检查 AQS 队列中是否有前驱节点, 没有(false)才去竞争
final boolean writerShouldBlock() {
return hasQueuedPredecessors();
}t2 r.lock（读锁），进入 tryAcquireShared 流程：
- 返回 -1 表示失败
- 如果返回 0 表示成功
- 返回正数表示还有多少后继节点支持共享模式，读写锁返回 1
1
2
3
4
5
6
7
8
9
public void lock() {
sync.acquireShared(1);
}
public final void acquireShared(int arg) {
// tryAcquireShared 返回负数, 表示获取读锁失败
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}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
// 尝试以共享模式获取
protected final int tryAcquireShared(int unused) {
Thread current = Thread.currentThread();
int c = getState();
// exclusiveCount(c) 代表低 16 位, 写锁的 state，成立说明有线程持有写锁
// 写锁的持有者不是当前线程，则获取读锁失败，【写锁允许降级】
if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current)
return -1;
// 高 16 位，代表读锁的 state，共享锁分配出去的总次数
int r = sharedCount(c);
// 读锁是否应该阻塞
if (!readerShouldBlock() && r < MAX_COUNT &&
compareAndSetState(c, c + SHARED_UNIT)) { // 尝试增加读锁计数
// 加锁成功
// 加锁之前读锁为 0，说明当前线程是第一个读锁线程
if (r == 0) {
firstReader = current;
firstReaderHoldCount = 1;
// 第一个读锁线程是自己就发生了读锁重入
} else if (firstReader == current) {
firstReaderHoldCount++;
} else {
// cachedHoldCounter 设置为当前线程的 holdCounter 对象，即最后一个获取读锁的线程
HoldCounter rh = cachedHoldCounter;
// 说明还没设置 rh
if (rh == null || rh.tid != getThreadId(current))
// 获取当前线程的锁重入的对象，赋值给 cachedHoldCounter
cachedHoldCounter = rh = readHolds.get();
// 还没重入
else if (rh.count == 0)
readHolds.set(rh);
// 重入 + 1
rh.count++;
}
// 读锁加锁成功
return 1;
}
// 逻辑到这 应该阻塞，或者 cas 加锁失败
// 会不断尝试 for (;;) 获取读锁, 执行过程中无阻塞
return fullTryAcquireShared(current);
}
// 非公平锁 readerShouldBlock 偏向写锁一些，看 AQS 阻塞队列中第一个节点是否是写锁，是则阻塞，反之不阻塞
// 防止一直有读锁线程，导致写锁线程饥饿
// true 则该阻塞, false 则不阻塞
final boolean readerShouldBlock() {
return apparentlyFirstQueuedIsExclusive();
}
final boolean readerShouldBlock() {
return hasQueuedPredecessors();
}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();
// 说明有线程持有写锁
if (exclusiveCount(c) != 0) {
// 写锁不是自己则获取锁失败
if (getExclusiveOwnerThread() != current)
return -1;
} else if (readerShouldBlock()) {
// 条件成立说明当前线程是 firstReader，当前锁是读忙碌状态，而且当前线程也是读锁重入
if (firstReader == current) {
// assert firstReaderHoldCount > 0;
} else {
if (rh == null) {
// 最后一个读锁的 HoldCounter
rh = cachedHoldCounter;
// 说明当前线程也不是最后一个读锁
if (rh == null || rh.tid != getThreadId(current)) {
// 获取当前线程的 HoldCounter
rh = readHolds.get();
// 条件成立说明 HoldCounter 对象是上一步代码新建的
// 当前线程不是锁重入，在 readerShouldBlock() 返回 true 时需要去排队
if (rh.count == 0)
// 防止内存泄漏
readHolds.remove();
}
}
if (rh.count == 0)
return -1;
}
}
// 越界判断
if (sharedCount(c) == MAX_COUNT)
throw new Error("Maximum lock count exceeded");
// 读锁加锁，条件内的逻辑与 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; // cache for release
}
return 1;
}
}
}获取读锁失败，进入 sync.doAcquireShared (1) 流程开始阻塞，首先也是调用 addWaiter 添加节点，不同之处在于节点被设置为 Node.SHARED 模式而非 Node.EXCLUSIVE 模式，注意此时 t2 仍处于活跃状态
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
private void doAcquireShared(int arg) {
// 将当前线程关联到一个 Node 对象上, 模式为共享模式
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);
// r >= 0 表示获取成功
if (r >= 0) {
//【这里会设置自己为头节点，唤醒相连的后序的共享节点】
setHeadAndPropagate(node, r);
p.next = null; // help GC
if (interrupted)
selfInterrupt();
failed = false;
return;
}
}
// 是否在获取读锁失败时阻塞 park 当前线程
if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}如果没有成功，在 doAcquireShared 内 for (;;) 循环一次，shouldParkAfterFailedAcquire 内把前驱节点的 waitStatus 改为 -1，再 for (;;) 循环一次尝试 tryAcquireShared，不成功在 parkAndCheckInterrupt () 处 park
这种状态下，假设又有 t3 r.lock，t4 w.lock，这期间 t1 仍然持有锁，就变成了下面的样子
解锁原理
t1 w.unlock， 写锁解锁
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
public void unlock() {
// 释放锁
sync.release(1);
}
public final boolean release(int arg) {
// 尝试释放锁
if (tryRelease(arg)) {
Node h = head;
// 头节点不为空并且不是等待状态不是 0，唤醒后继的非取消节点
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
protected final boolean tryRelease(int releases) {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
int nextc = getState() - releases;
// 因为可重入的原因, 写锁计数为 0, 才算释放成功
boolean free = exclusiveCount(nextc) == 0;
if (free)
setExclusiveOwnerThread(null);
setState(nextc);
return free;
}唤醒流程 sync.unparkSuccessor，这时 t2 在 doAcquireShared 的 parkAndCheckInterrupt () 处恢复运行，继续循环，执行 tryAcquireShared 成功则让读锁计数加一
接下来 t2 调用 setHeadAndPropagate (node, 1)，它原本所在节点被置为头节点；还会检查下一个节点是否是 shared，如果是则调用 doReleaseShared () 将 head 的状态从 -1 改为 0 并唤醒下一个节点，这时 t3 在 doAcquireShared 内 parkAndCheckInterrupt () 处恢复运行，唤醒连续的所有的共享节点
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head;
// 设置自己为 head 节点
setHead(node);
// propagate 表示有共享资源（例如共享读锁或信号量），为 0 就没有资源
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
// 获取下一个节点
Node s = node.next;
// 如果当前是最后一个节点，或者下一个节点是【等待共享读锁的节点】
if (s == null || s.isShared())
// 唤醒后继节点
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
27
private void doReleaseShared() {
// 如果 head.waitStatus == Node.SIGNAL ==> 0 成功, 下一个节点 unpark
// 如果 head.waitStatus == 0 ==> Node.PROPAGATE
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
// SIGNAL 唤醒后继
if (ws == Node.SIGNAL) {
// 因为读锁共享，如果其它线程也在释放读锁，那么需要将 waitStatus 先改为 0
// 防止 unparkSuccessor 被多次执行
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue;
// 唤醒后继节点
unparkSuccessor(h);
}
// 如果已经是 0 了，改为 -3，用来解决传播性
else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue;
}
// 条件不成立说明被唤醒的节点非常积极，直接将自己设置为了新的 head，
// 此时唤醒它的节点（前驱）执行 h == head 不成立，所以不会跳出循环，会继续唤醒新的 head 节点的后继节点
if (h == head)
break;
}
}
下一个节点不是 shared 了，因此不会继续唤醒 t4 所在节点
t2 读锁解锁，进入 sync.releaseShared (1) 中，调用 tryReleaseShared (1) 让计数减一，但计数还不为零，t3 同样让计数减一，计数为零，进入 doReleaseShared () 将头节点从 -1 改为 0 并唤醒下一个节点
1
2
3
4
5
6
7
8
9
10
11JAVA
public void unlock() {
sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}1
2
3
4
5
6
7
8
9
10
11
12
protected final boolean tryReleaseShared(int unused) {
for (;;) {
int c = getState();
int nextc = c - SHARED_UNIT;
// 读锁的计数不会影响其它获取读锁线程, 但会影响其它获取写锁线程，计数为 0 才是真正释放
if (compareAndSetState(c, nextc))
// 返回是否已经完全释放了
return nextc == 0;
}
}t4 在 acquireQueued 中 parkAndCheckInterrupt 处恢复运行，再次 for (;;) 这次自己是头节点的临节点，并且没有其他节点竞争，tryAcquire (1) 成功，修改头结点，流程结束
Stamped
StampedLock：读写锁，该类自 JDK 8 加入，是为了进一步优化读性能
特点：
- 在使用读锁、写锁时都必须配合戳使用
- StampedLock 不支持条件变量
- StampedLock 不支持重入
基本用法
加解读锁：
1
2
3
long stamp = lock.readLock();
lock.unlockRead(stamp); // 类似于 unpark，解指定的锁加解写锁：
1
2
3
long stamp = lock.writeLock();
lock.unlockWrite(stamp);乐观读，StampedLock 支持
tryOptimisticRead()方法，读取完毕后做一次戳校验，如果校验通过，表示这期间没有其他线程的写操作，数据可以安全使用，如果校验没通过，需要重新获取读锁，保证数据一致性1
2
3
4
5
6
long stamp = lock.tryOptimisticRead();
// 验戳
if(!lock.validate(stamp)){
// 锁升级
}
提供一个数据容器类内部分别使用读锁保护数据的 read () 方法，写锁保护数据的 write () 方法：
- 读 - 读可以优化
- 读 - 写优化读，补加读锁
1 |
|
本博客所有文章除特别声明外，均采用 CC BY-NC-SA 4.0 许可协议。转载请注明来自 little_kim！
相关推荐
评论
