NonfairSync#writerShouldBlock :写线程在抢锁之前永远不会阻塞,非公平性。 NonfairSync#readerShouldBlock:读线程抢锁之前,如果队列head后继(head.next)是独占节点时阻塞。

static final class NonfairSync extends Sync {

private static final long serialVersionUID = -8159625535654395037L;

final boolean writerShouldBlock() {

//写线程在抢锁之前永远不会阻塞-非公平锁

return false; // writers can always barge

}

final boolean readerShouldBlock() {

读线程抢锁的时候,如果队列第一个是实质性节点(head.next)是独占节点时阻塞 返回true是阻塞

*/

return apparentlyFirstQueuedIsExclusive();

}

}

/**

判断qas队列的第一个元素是否是独占线程(写线程) @return

*/

//AbstractQueuedSynchronizer

final boolean apparentlyFirstQueuedIsExclusive() {

Node h, s;

return (h = head) != null &&

(s = h.next) != null &&

!s.isShared() &&

s.thread != null;

}

final boolean isShared() {

return nextWaiter == SHARED;

}

2、FairSync

在FairSync,无论是写线程还是读线程,只要同步队列中有其他节点在等待锁,就阻塞,这就是公平性。

static final class FairSync extends Sync {

private static final long serialVersionUID = -2274990926593161451L;

/**

同步队列中head后继(head.next)不是当前线程时阻塞, 即同步队列中有其他节点在等待锁,此时当前写线程阻塞 @return

*/

final boolean writerShouldBlock() {

return hasQueuedPredecessors();

}

/**

同步队列中排在第一个实质性节点(head.next)不是当前线程时阻塞, 即同步队列中有其他节点在等待锁,此时当前读线程阻塞 @return

*/

final boolean readerShouldBlock() {

return hasQueuedPredecessors();

}

}

//同步队列中head后继(head.next)是不是当前线程

public final boolean hasQueuedPredecessors() {

// The correctness of this depends on head being initialized

// before tail and on head.next being accurate if the current

// thread is first in queue.

Node t = tail; // Read fields in reverse initialization order

Node h = head;

Node s;

return h != t &&

((s = h.next) == null || s.thread != Thread.currentThread());

}

五、读锁的获取与释放

ReadLock和WriteLock分别实现了Lock的lock和unlock等方法,实际上都是调用的ReentrantReadWriteLock.Sync的中已经实现好的模板方法。

1、ReadLock#lock

读锁是共享锁,首先尝试获取锁tryAcquireShared,获取锁失败则进入同步队列操作doAcquireShared。

//ReentrantReadWriteLock.ReadLock#lock

public void lock() {

sync.acquireShared(1);

}

//AbstractQueuedSynchronizer#acquireShared

public final void acquireShared(int arg) {

//tryAcquireShared 返回-1 获取锁失败,1获取锁成功

if (tryAcquireShared(arg) < 0)

//获取锁失败入同步队列

doAcquireShared(arg);

}

(1)tryAcquireShared获取共享锁

tryAcquireShared在ReentrantReadWriteLock#Sync中实现的。如下是获取共享锁的基本流程:

判断state,是否有线程持有写锁,若有且持有锁的不是当前线程,则返回-1,获取锁失败。(读写互斥) 若持有写锁的是当前线程,或者没有线程持有写锁,接下来判断读线程是否应该阻塞readerShouldBlock()。 readerShouldBlock()区分公平性,非公平锁,队列head后继(head.next)是独占节点,则阻塞;公平锁,队列中有其他节点在等待锁,则阻塞。 读线程不阻塞且加锁次数不超过MAX_COUNT且CAS拿读锁成功c + SHARED_UNIT。 若r = sharedCount(c)=0说明没有线程持有读锁,此时设置firstReader为当前线程,第一个读线程重入次数firstReaderHoldCount为1。 若r = sharedCount(c)!=0说明有线程持有读锁,此时当前线程是firstReader,则firstReaderHoldCount+1。 若持有当前读锁的不是firstReader,则HoldCounter来记录各个线程的读锁重入次数。 若因为CAS获取读锁失败,会进行自旋获取读锁fullTryAcquireShared(current)。

//ReentrantReadWriteLock.Sync#tryAcquireShared

protected final int tryAcquireShared(int unused) {

Thread current = Thread.currentThread();

int c = getState();

//exclusiveCount© != 0 写锁被某线程持有,当前持有锁的线程不是当前线程,直接返回-1

if (exclusiveCount© != 0 &&

getExclusiveOwnerThread() != current)

return -1;

//写锁没有线程持有或者当前持有写锁的写线程可以继续拿读锁

int r = sharedCount©;

//nonFairSync 队列第一个是写线程时,读阻塞,!false && 加锁次数不超过max && CAS拿读锁,高16位+1

//FairSync 当队列的第一个不是当前线程时,阻塞。。。

if (!readerShouldBlock() &&

r < MAX_COUNT &&

compareAndSetState(c, c + SHARED_UNIT)) {

if (r == 0) { //r==0说明是第一个拿到读锁的读线程

firstReader = current;

firstReaderHoldCount = 1;

} else if (firstReader == current) { //第一个持有读锁的线程时当前线程,为重入

firstReaderHoldCount++;

} else {

//HoldCounter 记录线程重入锁的次数

//读锁 可以多个读线程持有,所以会记录持有读锁的所有读线程和分别重入次数

HoldCounter rh = cachedHoldCounter;

//rh==null 从readHolds获取

//rh != null rh.tid != 当前线程

if (rh == null || rh.tid != getThreadId(current))

//取出当前线程的cachedHoldCounter

cachedHoldCounter = rh = readHolds.get();

else if (rh.count == 0)

readHolds.set(rh);

rh.count++;

}

return 1;

}

//compareAndSetState(c, c + SHARED_UNIT) 失败后 自旋尝试获取锁

return fullTryAcquireShared(current);

}

(2)fullTryAcquireShared自旋获取共享锁

自旋获取锁的过程与tryAcquireShared类似,获取读锁,记录重入,只不过加了一个循环,循环结束的条件是获取锁成功(1)或者不满足获取锁的条件(-1)。

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© != 0) {

//写锁有线程持有,但是持锁的线程不是当前线程,返回-1,结束自旋。

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)

//删除不持有该读锁的cachedHoldCounter

readHolds.remove();

}

}

if (rh.count == 0)

//当前线程不持有锁,直接返回-1

return -1;

}

}

//读线程不应该阻塞,判断state

if (sharedCount© == MAX_COUNT)

throw new Error(“Maximum lock count exceeded”);

//获取读锁

if (compareAndSetState(c, c + SHARED_UNIT)) {

//下面就是记录重入的机制了

if (sharedCount© == 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;

}

}

}

(3)doAcquireShared进入同步队列操作

tryAcquireShared(arg)返回-1,获取锁失败,则进入同步队列操作:

创建一个共享节点,并拼接到同步队列尾部。 获取新节点的前继节点,若是head,则尝试获取锁。 获取锁成功,唤醒后继共享节点并出队列。 node的前继节点不是head,或者获取锁失败,判断是否应该阻塞(shouldParkAfterFailedAcquire),应该阻塞parkAndCheckInterrupt阻塞当前线程。

private void doAcquireShared(int arg) {

//创建一个读节点,并入队列

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) {

//获取锁成功,node出队列,

//唤醒其后继共享节点的线程

setHeadAndPropagate(node, r);

p.next = null; // help GC

if (interrupted)

selfInterrupt();

failed = false;

return;

}

}

/**

p不是头结点 or 获取锁失败,判断是否应该被阻塞 前继节点的ws = SIGNAL 时应该被阻塞

*/

if (shouldParkAfterFailedAcquire(p, node) &&

parkAndCheckInterrupt())

interrupted = true;

}

} finally {

if (failed)

cancelAcquire(node);

}

}

这里的setHeadAndPropagate()是在获取共享锁成功的情况下调用的,所以propagate>0,(tryAcquireShared在Semaphore中有返回0的情况,返回结果为资源剩余量)。若node的下一个节点是共享节点,则调用doReleaseShared()唤醒后继节点。

(4)setHeadAndPropagate传播唤醒后继共享节点

首先获取锁的node节点赋值给head,成为新head。 在ReetrantReadWriteLock中node获取锁成功只有可能是propagate > 0,所以后面新旧head判断会省略,可以暂时不用考虑。 若node后面没有节点(调用doReleaseShared没多大意义),或者node后面有节点且是共享节点则会调用doReleaseShared()唤醒后继节点。

(共享锁的传播性,详解请移步《AQS源码解读(六)——从PROPAGATE和setHeadAndPropagate()分析共享锁的传播性》。)

private void setHeadAndPropagate(Node node, int propagate) {

Node h = head; // Record old head for check below

setHead(node);

// propagate > 0 获取锁成功

// propagate < 0 获取锁失败,队列不为空,h.waitStatus < 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();

}

}

2、ReadLock#lockInterruptibly

可中断获取锁,顾名思义就是获取锁的过程可响应中断。ReadLock#lockInterruptibly在获取锁的过程中有被中断(Thread.interrupted()),则会抛出异常InterruptedException,终止操作;其直接调用了AQS的模板方法acquireSharedInterruptibly。

(acquireSharedInterruptibly和doAcquireSharedInterruptibly详解请移步《AQS源码解读(五)——从acquireShared探索共享锁实现原理,何为共享?如何共享?》)

//ReentrantReadWriteLock.ReadLock#lockInterruptibly

public void lockInterruptibly() throws InterruptedException {

sync.acquireSharedInterruptibly(1);

}

//AbstractQueuedSynchronizer#acquireSharedInterruptibly

public final void acquireSharedInterruptibly(int arg)

throws InterruptedException {

if (Thread.interrupted())

//被打断 抛出异常

throw new InterruptedException();

if (tryAcquireShared(arg) < 0)

//获取锁失败,进入队列操作

doAcquireSharedInterruptibly(arg);

}

3、ReadLock#tryLock

ReadLock#tryLoc尝试获取锁,调用的是ReentrantReadWriteLock.Sync实现的tryReadLock,获取锁成功返回true,失败返回false,不会进入队列操作,所以也不区分公平性。代码结构上和ReentrantReadWriteLock.Sync#tryAcquireShared相似,所以不多赘述,不同之处是ReadLock#tryLoc本身就是自旋获取锁。

public boolean tryLock() {

return sync.tryReadLock();

}

//ReentrantReadWriteLock.Sync#tryReadLock

final boolean tryReadLock() {

Thread current = Thread.currentThread();

for (; {

int c = getState();

//判断是否有线程持有写锁

if (exclusiveCount© != 0 &&

getExclusiveOwnerThread() != current)

return false;

int r = sharedCount©;

if (r == MAX_COUNT)

throw new Error(“Maximum lock count exceeded”);

//没有线程持有写锁or持有写锁的是当前线程,写锁–>读锁 锁降级

if (compareAndSetState(c, c + SHARED_UNIT)) {

//获取读锁成功

if (r == 0) {

//第一个读锁的线程

firstReader = current;

firstReaderHoldCount = 1;

} else if (firstReader == current) {

//不是第一次获取读锁 但是firstReader是当前线程,重入

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 true;

}

}

}

同样和ReentrantLock一样,ReadLock#tryLock也有一个重载方法,可传入一个超时时长timeout和一个时间单位TimeUnit,超时时长会被转为纳秒级。

public boolean tryLock(long timeout, TimeUnit unit)

throws InterruptedException {

return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));

}

//AbstractQueuedSynchronizer#tryAcquireSharedNanos

public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout)

throws InterruptedException {

if (Thread.interrupted())

throw new InterruptedException();

return tryAcquireShared(arg) >= 0 ||

doAcquireSharedNanos(arg, nanosTimeout);

}

自我介绍一下,小编13年上海交大毕业,曾经在小公司待过,也去过华为、OPPO等大厂,18年进入阿里一直到现在。

深知大多数Java工程师,想要提升技能,往往是自己摸索成长或者是报班学习,但对于培训机构动则几千的学费,着实压力不小。自己不成体系的自学效果低效又漫长,而且极易碰到天花板技术停滞不前!

因此收集整理了一份《2024年Java开发全套学习资料》,初衷也很简单,就是希望能够帮助到想自学提升又不知道该从何学起的朋友,同时减轻大家的负担。

既有适合小白学习的零基础资料,也有适合3年以上经验的小伙伴深入学习提升的进阶课程,基本涵盖了95%以上Java开发知识点,真正体系化!

由于文件比较大,这里只是将部分目录大纲截图出来,每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频,并且后续会持续更新

如果你觉得这些内容对你有帮助,可以添加V获取:vip1024b (备注Java)

最后总结

ActiveMQ+Kafka+RabbitMQ学习笔记PDF

RabbitMQ实战指南

手写RocketMQ笔记

手写“Kafka笔记”

关于分布式,限流+缓存+缓存,这三大技术(包含:ZooKeeper+Nginx+MongoDB+memcached+Redis+ActiveMQ+Kafka+RabbitMQ)等等。这些相关的面试也好,还有手写以及学习的笔记PDF,都是啃透分布式技术必不可少的宝藏。以上的每一个专题每一个小分类都有相关的介绍,并且小编也已经将其整理成PDF啦 .(img-nNe0WsU8-1711655616914)] [外链图片转存中…(img-7wXS2Hfs-1711655616915)] [外链图片转存中…(img-hwPkzguQ-1711655616915)]

既有适合小白学习的零基础资料,也有适合3年以上经验的小伙伴深入学习提升的进阶课程,基本涵盖了95%以上Java开发知识点,真正体系化!

由于文件比较大,这里只是将部分目录大纲截图出来,每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频,并且后续会持续更新

如果你觉得这些内容对你有帮助,可以添加V获取:vip1024b (备注Java) [外链图片转存中…(img-sRCtB6WB-1711655616916)]

最后总结

ActiveMQ+Kafka+RabbitMQ学习笔记PDF

[外链图片转存中…(img-mASrKmXc-1711655616916)]

RabbitMQ实战指南

[外链图片转存中…(img-oMY09sQh-1711655616916)]

手写RocketMQ笔记

[外链图片转存中…(img-CO3xZiOn-1711655616917)]

手写“Kafka笔记”

[外链图片转存中…(img-6outGAXG-1711655616917)]

关于分布式,限流+缓存+缓存,这三大技术(包含:ZooKeeper+Nginx+MongoDB+memcached+Redis+ActiveMQ+Kafka+RabbitMQ)等等。这些相关的面试也好,还有手写以及学习的笔记PDF,都是啃透分布式技术必不可少的宝藏。以上的每一个专题每一个小分类都有相关的介绍,并且小编也已经将其整理成PDF啦

推荐阅读

评论可见,请评论后查看内容,谢谢!!!
 您阅读本篇文章共花了: