ReentrantReadWriterLock源码(state设计、读写锁、共享锁、独占锁及锁降级)

ReentrantReadWriterLock

读写锁类图(截图来源https://blog.csdn.net/wangbo199308/article/details/108688148)

state的设计

读写锁将变量state切分成两个部分,高16位表示读,低16位表示写

源码中将4字节(32位)的int数据类型state,通过SHARED_SHIFT(16)划分读和写;

每次读锁增加的单元,SHARED_UNIT = (1 << SHARED_SHIFT) 也即0x00010000,即每次读锁增加从17位开始加1

读写锁最大数量:MAX_COUNT = (1 << SHARED_SHIFT) – 1,16位最大值

写锁的掩码:EXCLUSIVE_MASK = (1 << SHARED_SHIFT) – 1, 即求写锁数量,将state和此掩码做与运算,将高16位抹去

计算读锁数量逻辑:c >>> SHARED_SHIFT,取高16位

计算写锁数量逻辑:c & EXCLUSIVE_MASK,将state和此掩码做与运算,将高16位抹去

public class ReentrantReadWriteLock
implements ReadWriteLock, java.io.Serializable {
abstract static class Sync extends AbstractQueuedSynchronizer {
//16位划分读和写
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;
static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }
static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }
}
}

读锁

读锁上锁的调用链:ReentrantReadWriteLock$ReadLock#lock() –>AbstractQueuedSynchronizer#acquireShared() –>ReentrantReadWriteLock$Sync#tryAcquireShared()

当前写锁数量为0或独占锁持有者就是当前线程才进行读锁逻辑

读锁数量通过CAS加1

之后逻辑是将读锁线程放入ThreadLocal中,记录各自锁数量

public class ReentrantReadWriteLock
implements ReadWriteLock, java.io.Serializable {
public static class ReadLock implements Lock, java.io.Serializable {
public void lock() {
sync.acquireShared(1);
}
}
}
public abstract class AbstractQueuedSynchronizer
extends AbstractOwnableSynchronizer
implements java.io.Serializable {
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}
}
public class ReentrantReadWriteLock
implements ReadWriteLock, java.io.Serializable {
abstract static class Sync extends AbstractQueuedSynchronizer {
protected final int tryAcquireShared(int unused) {
Thread current = Thread.currentThread();
int c = getState();
// 同时满足写锁数量不为0,且独占锁不是当前线程,走doAcquireShared逻辑
if (exclusiveCount(c) != 0 &&
getExclusiveOwnerThread() != current)
return -1;
// 取高16位读锁数量
int r = sharedCount(c);
if (!readerShouldBlock() &&
r < MAX_COUNT &&
compareAndSetState(c, c + SHARED_UNIT)) {
// ThreadLocal存放锁信息
if (r == 0) {
firstReader = current;
firstReaderHoldCount = 1;
} 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);
}
}
}

在读锁获取锁过程,写锁不为0且占有写锁的不是当前线程,返回-1,走同步器doAcquireShared方法,等待写锁释放;

前置节点是head节点时,尝试获取共享锁

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) {
//前置节点是head节点时,尝试获取共享锁
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null; // help GC
if (interrupted)
selfInterrupt();
failed = false;
return;
}
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}

写锁

  1. 读锁不为0,但写锁为0,获取锁失败;读锁不为0,写锁也不为0,但独占锁不是当前线程,获取锁失败
  2. 如果锁数量已到最大,获取失败
  3. 否则获取写锁,更新state
public class ReentrantReadWriteLock
implements ReadWriteLock, java.io.Serializable {
abstract static class Sync extends AbstractQueuedSynchronizer {
protected final boolean tryAcquire(int acquires) {
Thread current = Thread.currentThread();
int c = getState();
int w = exclusiveCount(c);
if (c != 0) {
// (Note: if c != 0 and w == 0 then shared count != 0)
if (w == 0 || current != getExclusiveOwnerThread())
return false;
if (w + exclusiveCount(acquires) > MAX_COUNT)
throw new Error("Maximum lock count exceeded");
// Reentrant acquire
setState(c + acquires);
return true;
}
if (writerShouldBlock() ||
!compareAndSetState(c, c + acquires))
return false;
setExclusiveOwnerThread(current);
return true;
}
}
}

共享锁和独占锁

读锁是共享锁,当线程1获得读锁时,并不会排斥线程2去获取读锁,而是在ThreadLocal中保存每个锁数量

abstract static class Sync extends AbstractQueuedSynchronizer {
static final class HoldCounter {
int count = 0;
// Use id, not reference, to avoid garbage retention
final long tid = getThreadId(Thread.currentThread());
}
static final class ThreadLocalHoldCounter
extends ThreadLocal<HoldCounter> {
public HoldCounter initialValue() {
return new HoldCounter();
}
}
}

写锁是独占锁,会调用同步器AbstractQueuedSynchronizer#acquire()方法,默认加入队列的node模式是独占模式

public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}

锁降级

锁降级就是从写锁降级成为读锁。在当前线程拥有写锁的情况下,再次获取到读锁,随后释放写锁的过程就是锁降级

锁降级示例:

public void processData() {
ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
ReentrantReadWriteLock.ReadLock readLock = lock.readLock();
ReentrantReadWriteLock.WriteLock writeLock = lock.writeLock();
readLock.lock();
if(!update) {
//必须先释放读锁
readLock.unlock();
// 锁降级从写锁获取到开始
writeLock.lock();
try{
if(!update) {
update = true;
}
// 可以获取到读锁,getExclusiveOwnerThread() == current
readLock.lock();
} finally {
writeLock.unlock();
}
//锁降级完成,写锁降级为读锁
}
try{
// 使用数据的流程
} finally {
readLock.unlock();
}
}

可降级的源码仍是在读锁tryAcquireShared方法中,getExclusiveOwnerThread() == current,也即当前独占锁owner就是当前线程,可进行读锁逻辑。

protected final int tryAcquireShared(int unused) {
if (exclusiveCount(c) != 0 &&
getExclusiveOwnerThread() != current)
return -1;
}

参考:《Java并发编程的艺术》

博客园精华区
我还没有学会写个人说明!
下一篇

一、Linux cpuidle framework(1)_概述和软件架构

你也可能喜欢

评论已经被关闭。

插入图片