接下来探讨ReadWriteLock的公平锁实现, 也是分如下场景分析
情景1 三个线程都是读
public static void main(String[] args){
final Printer printer = new Printer();
Thread thread1 = new Thread(){
@Override
public void run() {
try {
printer.read("test1");
} catch (Exception e) {
e.printStackTrace();
}
}
};
Thread thread2 = new Thread(){
@Override
public void run() {
try {
printer.read("test2");
} catch (Exception e) {
e.printStackTrace();
}
}
};
Thread thread3 = new Thread(){
@Override
public void run() {
try {
printer.read("test3");
} catch (Exception e) {
e.printStackTrace();
}
}
};
thread1.start();
thread2.start();
thread3.start();
}
ReadWriteLock lock = new ReentrantReadWriteLock(true);
public void read(String str) throws Exception{
lock.readLock().lock();
try {
System.out.println(str);
Thread.sleep(200);
}finally {
lock.readLock().unlock();
}
}
public void lock() {
sync.acquireShared(1);
}
// AQS
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}
// sync
protected final int tryAcquireShared(int unused) {
// Thread-0
Thread current = Thread.currentThread();
// c = 0
int c = getState();
// 不走此分支
if (exclusiveCount(c) != 0 &&
getExclusiveOwnerThread() != current)
return -1;
// r = 0
int r = sharedCount(c);
//
if (!readerShouldBlock() &&
r < MAX_COUNT &&
compareAndSetState(c, c + SHARED_UNIT)) {
......
}
......
}
static final class FairSync extends Sync {
final boolean readerShouldBlock() {
return hasQueuedPredecessors();
}
}
// 这段代码目的就是看queue里是否还存在想要获取锁的线程
// 此段代码解析在《ReentrantLock之公平锁unlock》这片文章中可看到
public final boolean hasQueuedPredecessors() {
Node t = tail;
Node h = head;
Node s;
// 此时h和t都是null, h == t, 所以返回false
return h != t &&
((s = h.next) == null || s.thread != Thread.currentThread());
}
接下来回到tryAcquireShared方法中
protected final int tryAcquireShared(int unused) {
......
// readerShouldBlock方法返回false,
if (!readerShouldBlock() &&
// r = 0, 确实比MAX_COUNT小
r < MAX_COUNT &&
// 成功设置state为65536
compareAndSetState(c, c + SHARED_UNIT)) {
// 因为r=0, 进入此分支
if (r == 0) {
// firstReader为thread-0
firstReader = current;
firstReaderHoldCount = 1;
} else if (firstReader == current) {
......
} else {
......
}
// 从此返回
return 1;
}
......
}
此时线程1已经获取读锁, 接下来线程2开始执行
public void lock() {
sync.acquireShared(1);
}
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}
protected final int tryAcquireShared(int unused) {
// thread-1
Thread current = Thread.currentThread();
// state = 65536
int c = getState();
// 没加写锁, 跳过此分支
if (exclusiveCount(c) != 0 &&
getExclusiveOwnerThread() != current)
return -1;
// r = 1, 表示此时已经获得一次读锁
int r = sharedCount(c);
// readerShouldBlock方法仍旧返回false
if (!readerShouldBlock() &&
// MAX_COUNT为65535, 也就是获取读锁的最大次数就是65535
// 此时为1, 并没有大于MAX_COUNT
r < MAX_COUNT &&
// 已经设置state为65536*2
compareAndSetState(c, c + SHARED_UNIT)) {
// 不走此分支
if (r == 0) {
......
}
// 由于非重入锁也不走此分支
else if (firstReader == current) {
firstReaderHoldCount++;
}
// 最终走此分支
else {
// rh为null
HoldCounter rh = cachedHoldCounter;
// 走此分支
if (rh == null || rh.tid != getThreadId(current))
// 新建HoldCounter对象
cachedHoldCounter = rh = readHolds.get();
else if (rh.count == 0)
......
// 并将HoldCounter对象count(对应重入次数)+1
rh.count++;
}
return 1;
}
......
}
到此第二个线程也获得读锁成功
接下来第三个线程开始执行
// ReadLock
public void lock() {
sync.acquireShared(1);
}
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}
protected final int tryAcquireShared(int unused) {
// thread-2
Thread current = Thread.currentThread();
// 131072 ==> 65536 * 2
int c = getState();
// 没有读锁, 此分支不走
if (exclusiveCount(c) != 0 &&
getExclusiveOwnerThread() != current)
return -1;
// r = 2
int r = sharedCount(c);
// 都是读操作, 没有阻塞
if (!readerShouldBlock() &&
// 读次数小于65535
r < MAX_COUNT &&
// state = 65536 * 3 ==> 196608
compareAndSetState(c, c + SHARED_UNIT)) {
if (r == 0) {
......
} else if (firstReader == current) {
......
}
// 走此分支
else {
// cachedHoldCounter表示上一个成功获取读锁的线程
// 在此就是线程thread-1
HoldCounter rh = cachedHoldCounter;
// 由于rh.tid保存的是thread-1的线程id
// 与当前线程保存的线程id不相等
// 所以最终还是走此分支
if (rh == null || rh.tid != getThreadId(current))
// 创建一个HoldCounter对象
cachedHoldCounter = rh = readHolds.get();
else if (rh.count == 0)
readHolds.set(rh);
// 将其获得锁次数加一
rh.count++;
}
// 返回
return 1;
}
...
}
这时线程3也获取到读锁
情景2 前两个线程是读, 最后一个线程是写
由于线程1和线程2的执行流程与情景1相同, 所以不再做分析, 接下来就做线程3的分析
public void lock() {
sync.acquire(1);
}
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
protected final boolean tryAcquire(int acquires) {
// thread-2
Thread current = Thread.currentThread();
// c = 65536 * 2
int c = getState();
// w = 0
int w = exclusiveCount(c);
// 进入此分支
if (c != 0) {
// 由于w为0, 进入此分支
if (w == 0 || current != getExclusiveOwnerThread())
return false;
......
}
......
}
由于之前已经有读操作获取锁, 所以写操作会被阻塞住, 接下来是执行写操作的线程如何被阻塞
回到acquire方法
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
// 这块代码就没什么特别的
// ReentrantLock阻塞线程也用的这个
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
网友评论