synchronized上锁机制是通过对象头来实现的,通过锁升级的过程来完成加锁。(https://blog.csdn.net/zhangting19921121/article/details/106002751)
但是synchronized锁升级的过程犹如一个黑盒,我们无法掌控。因此,在实际的工作中ReentrantLock使用相对比较频繁。ReentrantLock显式地获取锁,释放锁,可中断,同时还支持实现公平锁等。常用的写法如下:
ReentrantLock lock = new ReentrantLock(false);//false为非公平锁, true为公平锁
lock.lock() //加锁
// todo 业务代码
lock.unlock() //解锁
假如说有t0,t1,t2三个线程来进行调用,t0获取锁,开始执行业务代码。此时,t1,t2都应该停在lock.lock()方法中,不能向下执行业务代码。那么怎么停在里面呢?可以借鉴synchronized自旋的实现,
while(true) {
if (加锁成功) {
break; // 跳出循环
}
}
但是,一直让线程进行while循环,其实也是在不断消耗CPU。所以,如何让这些线程让出CPU呢?
while(true) {
if (加锁成功) {
break; // 跳出循环
}
Thread.yeild(); // 让出CPU的使用权
}
会有这样一个问题,假如说业务代码执行很慢,需要10几秒。此时只有t1,t2两个线程在等待获取锁还好,如果有100多个线程都在等待呢,那这100多个线程只能互相让来让去。如果改为Thread.sleep(睡眠时间);呢,但是这个睡眠时间选多少合适呢,这个时间不好确认,也不是一个很好的方法。那如果我把线程阻塞了,不让它再循环了,等待调度唤醒,减少线程间互相的让来来去或者是找一个合适的睡眠时间。
while(true) {
if (加锁成功) {
break; // 跳出循环
}
LockSupport.part();//阻塞线程,跳出循环
}
那谁去唤醒这些被阻塞的线程呢?
while(true) {
if (加锁成功) {
break; // 跳出循环
}
LockSupport.part();//阻塞线程,跳出循环
}
LockSupport.unpart(t);// 唤醒线程t
LockSupport.unpart(t)可以唤醒线程,但是我怎么知道唤醒哪个线程呢?可以使用一个对象来进行保存。
// lock.lock()
while(true) {
if (加锁成功) {
break; // 跳出循环
}
HashSet.add(t); // LinkedQueue也可以
LockSupport.part();//阻塞线程,跳出循环
}
// todo 业务逻辑
// lock.unlock()
Thread t = HashSet.get();// LinkedQueue.take()
LockSupport.unpart(t);// 唤醒线程t
以上,便是加锁,解锁的简单实现。但是除了自旋,LockSupport,Queue外还需要什么呢?在if判断的地方,必须要保证只有一个线程能够进入,通过synchronized就可以实现这个功能。此外,java中还有CAS可以实现和synchronized一样的功能,ReentrantLock是一种基于AQS框架的应用实现的。
// lock.lock()
while(true) {
if (cas加锁成功) {
break; // 跳出循环
}
HashSet.add(t); // LinkedQueue也可以
LockSupport.part();//阻塞线程,跳出循环
}
// todo 业务逻辑
// lock.unlock()
Thread t = HashSet.get();// LinkedQueue.take()
LockSupport.unpart(t);// 唤醒线程t
好啦,接下来就引入了本文的重点啦,继续往下看吧~
AQS(全称AbstractQueuedSynchronized),AQS定义了一套多线程访问共享资源 的同步器框架,是一个依赖状态(state)的同步器。
1-1.阻塞等待队列
1-2.共享/独占
1-3.公平/非公平
1-4.可重入
1-5.允许中断
2-1.AQS内部维护属性volatile int state (32位):state表示资源的可用状态,为0的时候表示当前锁并未被任何线程所持有。state三种访问方式:getState()、setState()、compareAndSetState()
2-2.AQS定义两种资源共享方式:Exclusive-独占(只有一个线程能执行,如ReentrantLock)、Share-共享(多个线程可以同时执行,如Semaphore/CountDownLatch)
2-3.AQS定义两种队列:同步等待队列、条件等待队列
不同的自定义同步器争用共享资源的方式也不同。自定义同步器在实现时只需要实现共享资源state的获取与释放方式即可,至于具体线程等待队列的维护(如获取资源失败入队/ 唤醒出队等),AQS已经在顶层实现好了。自定义同步器实现时主要实现以下几种方法:
isHeldExclusively():该线程是否正在独占资源。只有用到condition才需要去实现它。
tryAcquire(int):独占方式。尝试获取资源,成功则返回true,失败则返回false。
tryRelease(int):独占方式。尝试释放资源,成功则返回true,失败则返回false。
tryAcquireShared(int):共享方式。尝试获取资源。负数表示失败;0表示成功,但没有剩余可用资源;正数表示成功,且有剩余资源。
tryReleaseShared(int):共享方式。尝试释放资源,如果释放后允许唤醒后续等待结点返回true,否则返回false。
//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by FernFlower decompiler)
//
package java.util.concurrent.locks;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.concurrent.TimeUnit;
import sun.misc.Unsafe;
public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements Serializable {
private static final long serialVersionUID = 7373984972572414691L;
// 基于Node(prev、next)来构建的双向链表CLH
private transient volatile AbstractQueuedSynchronizer.Node head;
private transient volatile AbstractQueuedSynchronizer.Node tail;
// 状态器,为0的时候表示当前锁并未被任何线程所持有
private volatile int state;
static final long spinForTimeoutThreshold = 1000L;
private static final Unsafe unsafe = Unsafe.getUnsafe();
private static final long stateOffset;
private static final long headOffset;
private static final long tailOffset;
private static final long waitStatusOffset;
private static final long nextOffset;
protected AbstractQueuedSynchronizer() {
}
protected final int getState() {
return this.state;
}
protected final void setState(int var1) {
this.state = var1;
}
protected final boolean compareAndSetState(int var1, int var2) {
return unsafe.compareAndSwapInt(this, stateOffset, var1, var2);
}
private AbstractQueuedSynchronizer.Node enq(AbstractQueuedSynchronizer.Node var1) {
while(true) {
AbstractQueuedSynchronizer.Node var2 = this.tail;
if (var2 == null) {
if (this.compareAndSetHead(new AbstractQueuedSynchronizer.Node())) {
this.tail = this.head;
}
} else {
var1.prev = var2;
if (this.compareAndSetTail(var2, var1)) {
var2.next = var1;
return var2;
}
}
}
}
private AbstractQueuedSynchronizer.Node addWaiter(AbstractQueuedSynchronizer.Node var1) {
AbstractQueuedSynchronizer.Node var2 = new AbstractQueuedSynchronizer.Node(Thread.currentThread(), var1);
AbstractQueuedSynchronizer.Node var3 = this.tail;
if (var3 != null) {
var2.prev = var3;
if (this.compareAndSetTail(var3, var2)) {
var3.next = var2;
return var2;
}
}
this.enq(var2);
return var2;
}
private void setHead(AbstractQueuedSynchronizer.Node var1) {
this.head = var1;
var1.thread = null;
var1.prev = null;
}
private void unparkSuccessor(AbstractQueuedSynchronizer.Node var1) {
int var2 = var1.waitStatus;
if (var2 < 0) {
compareAndSetWaitStatus(var1, var2, 0);
}
AbstractQueuedSynchronizer.Node var3 = var1.next;
if (var3 == null || var3.waitStatus > 0) {
var3 = null;
for(AbstractQueuedSynchronizer.Node var4 = this.tail; var4 != null && var4 != var1; var4 = var4.prev) {
if (var4.waitStatus <= 0) {
var3 = var4;
}
}
}
if (var3 != null) {
LockSupport.unpark(var3.thread);
}
}
private void doReleaseShared() {
while(true) {
AbstractQueuedSynchronizer.Node var1 = this.head;
if (var1 != null && var1 != this.tail) {
int var2 = var1.waitStatus;
if (var2 == -1) {
if (!compareAndSetWaitStatus(var1, -1, 0)) {
continue;
}
this.unparkSuccessor(var1);
} else if (var2 == 0 && !compareAndSetWaitStatus(var1, 0, -3)) {
continue;
}
}
if (var1 == this.head) {
return;
}
}
}
private void setHeadAndPropagate(AbstractQueuedSynchronizer.Node var1, int var2) {
AbstractQueuedSynchronizer.Node var3 = this.head;
this.setHead(var1);
if (var2 > 0 || var3 == null || var3.waitStatus < 0 || (var3 = this.head) == null || var3.waitStatus < 0) {
AbstractQueuedSynchronizer.Node var4 = var1.next;
if (var4 == null || var4.isShared()) {
this.doReleaseShared();
}
}
}
private void cancelAcquire(AbstractQueuedSynchronizer.Node var1) {
if (var1 != null) {
var1.thread = null;
AbstractQueuedSynchronizer.Node var2;
for(var2 = var1.prev; var2.waitStatus > 0; var1.prev = var2 = var2.prev) {
}
AbstractQueuedSynchronizer.Node var3 = var2.next;
var1.waitStatus = 1;
if (var1 == this.tail && this.compareAndSetTail(var1, var2)) {
compareAndSetNext(var2, var3, (AbstractQueuedSynchronizer.Node)null);
} else {
int var4;
if (var2 != this.head && ((var4 = var2.waitStatus) == -1 || var4 <= 0 && compareAndSetWaitStatus(var2, var4, -1)) && var2.thread != null) {
AbstractQueuedSynchronizer.Node var5 = var1.next;
if (var5 != null && var5.waitStatus <= 0) {
compareAndSetNext(var2, var3, var5);
}
} else {
this.unparkSuccessor(var1);
}
var1.next = var1;
}
}
}
private static boolean shouldParkAfterFailedAcquire(AbstractQueuedSynchronizer.Node var0, AbstractQueuedSynchronizer.Node var1) {
int var2 = var0.waitStatus;
if (var2 == -1) {
return true;
} else {
if (var2 > 0) {
do {
var1.prev = var0 = var0.prev;
} while(var0.waitStatus > 0);
var0.next = var1;
} else {
compareAndSetWaitStatus(var0, var2, -1);
}
return false;
}
}
static void selfInterrupt() {
Thread.currentThread().interrupt();
}
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
final boolean acquireQueued(AbstractQueuedSynchronizer.Node var1, int var2) {
boolean var3 = true;
try {
boolean var4 = false;
while(true) {
AbstractQueuedSynchronizer.Node var5 = var1.predecessor();
if (var5 == this.head && this.tryAcquire(var2)) {
this.setHead(var1);
var5.next = null;
var3 = false;
boolean var6 = var4;
return var6;
}
if (shouldParkAfterFailedAcquire(var5, var1) && this.parkAndCheckInterrupt()) {
var4 = true;
}
}
} finally {
if (var3) {
this.cancelAcquire(var1);
}
}
}
private void doAcquireInterruptibly(int var1) throws InterruptedException {
AbstractQueuedSynchronizer.Node var2 = this.addWaiter(AbstractQueuedSynchronizer.Node.EXCLUSIVE);
boolean var3 = true;
try {
AbstractQueuedSynchronizer.Node var4;
do {
var4 = var2.predecessor();
if (var4 == this.head && this.tryAcquire(var1)) {
this.setHead(var2);
var4.next = null;
var3 = false;
return;
}
} while(!shouldParkAfterFailedAcquire(var4, var2) || !this.parkAndCheckInterrupt());
throw new InterruptedException();
} finally {
if (var3) {
this.cancelAcquire(var2);
}
}
}
private boolean doAcquireNanos(int var1, long var2) throws InterruptedException {
if (var2 <= 0L) {
return false;
} else {
long var4 = System.nanoTime() + var2;
AbstractQueuedSynchronizer.Node var6 = this.addWaiter(AbstractQueuedSynchronizer.Node.EXCLUSIVE);
boolean var7 = true;
try {
do {
AbstractQueuedSynchronizer.Node var8 = var6.predecessor();
boolean var9;
if (var8 == this.head && this.tryAcquire(var1)) {
this.setHead(var6);
var8.next = null;
var7 = false;
var9 = true;
return var9;
}
var2 = var4 - System.nanoTime();
if (var2 <= 0L) {
var9 = false;
return var9;
}
if (shouldParkAfterFailedAcquire(var8, var6) && var2 > 1000L) {
LockSupport.parkNanos(this, var2);
}
} while(!Thread.interrupted());
throw new InterruptedException();
} finally {
if (var7) {
this.cancelAcquire(var6);
}
}
}
}
private void doAcquireShared(int var1) {
AbstractQueuedSynchronizer.Node var2 = this.addWaiter(AbstractQueuedSynchronizer.Node.SHARED);
boolean var3 = true;
try {
boolean var4 = false;
while(true) {
AbstractQueuedSynchronizer.Node var5 = var2.predecessor();
if (var5 == this.head) {
int var6 = this.tryAcquireShared(var1);
if (var6 >= 0) {
this.setHeadAndPropagate(var2, var6);
var5.next = null;
if (var4) {
selfInterrupt();
}
var3 = false;
return;
}
}
if (shouldParkAfterFailedAcquire(var5, var2) && this.parkAndCheckInterrupt()) {
var4 = true;
}
}
} finally {
if (var3) {
this.cancelAcquire(var2);
}
}
}
private void doAcquireSharedInterruptibly(int var1) throws InterruptedException {
AbstractQueuedSynchronizer.Node var2 = this.addWaiter(AbstractQueuedSynchronizer.Node.SHARED);
boolean var3 = true;
try {
AbstractQueuedSynchronizer.Node var4;
do {
var4 = var2.predecessor();
if (var4 == this.head) {
int var5 = this.tryAcquireShared(var1);
if (var5 >= 0) {
this.setHeadAndPropagate(var2, var5);
var4.next = null;
var3 = false;
return;
}
}
} while(!shouldParkAfterFailedAcquire(var4, var2) || !this.parkAndCheckInterrupt());
throw new InterruptedException();
} finally {
if (var3) {
this.cancelAcquire(var2);
}
}
}
private boolean doAcquireSharedNanos(int var1, long var2) throws InterruptedException {
if (var2 <= 0L) {
return false;
} else {
long var4 = System.nanoTime() + var2;
AbstractQueuedSynchronizer.Node var6 = this.addWaiter(AbstractQueuedSynchronizer.Node.SHARED);
boolean var7 = true;
try {
do {
AbstractQueuedSynchronizer.Node var8 = var6.predecessor();
if (var8 == this.head) {
int var9 = this.tryAcquireShared(var1);
if (var9 >= 0) {
this.setHeadAndPropagate(var6, var9);
var8.next = null;
var7 = false;
boolean var10 = true;
return var10;
}
}
var2 = var4 - System.nanoTime();
if (var2 <= 0L) {
boolean var14 = false;
return var14;
}
if (shouldParkAfterFailedAcquire(var8, var6) && var2 > 1000L) {
LockSupport.parkNanos(this, var2);
}
} while(!Thread.interrupted());
throw new InterruptedException();
} finally {
if (var7) {
this.cancelAcquire(var6);
}
}
}
}
// 在AbstractQueuedSynchronizer中并没有实现,具体实现的逻辑都在子类中
protected boolean tryAcquire(int var1) {
throw new UnsupportedOperationException();
}
protected boolean tryRelease(int var1) {
throw new UnsupportedOperationException();
}
protected int tryAcquireShared(int var1) {
throw new UnsupportedOperationException();
}
protected boolean tryReleaseShared(int var1) {
throw new UnsupportedOperationException();
}
protected boolean isHeldExclusively() {
throw new UnsupportedOperationException();
}
public final void acquire(int var1) {
//this.acquireQueued(this.addWaiter(AbstractQueuedSynchronizer.Node.EXCLUSIVE), var1)加锁失败强制入队
if (!this.tryAcquire(var1) && this.acquireQueued(this.addWaiter(AbstractQueuedSynchronizer.Node.EXCLUSIVE), var1)) {
selfInterrupt();
}
}
public final void acquireInterruptibly(int var1) throws InterruptedException {
if (Thread.interrupted()) {
throw new InterruptedException();
} else {
if (!this.tryAcquire(var1)) {
this.doAcquireInterruptibly(var1);
}
}
}
public final boolean tryAcquireNanos(int var1, long var2) throws InterruptedException {
if (Thread.interrupted()) {
throw new InterruptedException();
} else {
return this.tryAcquire(var1) || this.doAcquireNanos(var1, var2);
}
}
public final boolean release(int var1) {
if (this.tryRelease(var1)) {
AbstractQueuedSynchronizer.Node var2 = this.head;
if (var2 != null && var2.waitStatus != 0) {
this.unparkSuccessor(var2);
}
return true;
} else {
return false;
}
}
public final void acquireShared(int var1) {
if (this.tryAcquireShared(var1) < 0) {
this.doAcquireShared(var1);
}
}
public final void acquireSharedInterruptibly(int var1) throws InterruptedException {
if (Thread.interrupted()) {
throw new InterruptedException();
} else {
if (this.tryAcquireShared(var1) < 0) {
this.doAcquireSharedInterruptibly(var1);
}
}
}
public final boolean tryAcquireSharedNanos(int var1, long var2) throws InterruptedException {
if (Thread.interrupted()) {
throw new InterruptedException();
} else {
return this.tryAcquireShared(var1) >= 0 || this.doAcquireSharedNanos(var1, var2);
}
}
public final boolean releaseShared(int var1) {
if (this.tryReleaseShared(var1)) {
this.doReleaseShared();
return true;
} else {
return false;
}
}
public final boolean hasQueuedThreads() {
return this.head != this.tail;
}
public final boolean hasContended() {
return this.head != null;
}
public final Thread getFirstQueuedThread() {
return this.head == this.tail ? null : this.fullGetFirstQueuedThread();
}
private Thread fullGetFirstQueuedThread() {
AbstractQueuedSynchronizer.Node var1;
AbstractQueuedSynchronizer.Node var2;
Thread var3;
if (((var1 = this.head) == null || (var2 = var1.next) == null || var2.prev != this.head || (var3 = var2.thread) == null) && ((var1 = this.head) == null || (var2 = var1.next) == null || var2.prev != this.head || (var3 = var2.thread) == null)) {
AbstractQueuedSynchronizer.Node var4 = this.tail;
Thread var5;
for(var5 = null; var4 != null && var4 != this.head; var4 = var4.prev) {
Thread var6 = var4.thread;
if (var6 != null) {
var5 = var6;
}
}
return var5;
} else {
return var3;
}
}
public final boolean isQueued(Thread var1) {
if (var1 == null) {
throw new NullPointerException();
} else {
for(AbstractQueuedSynchronizer.Node var2 = this.tail; var2 != null; var2 = var2.prev) {
if (var2.thread == var1) {
return true;
}
}
return false;
}
}
final boolean apparentlyFirstQueuedIsExclusive() {
AbstractQueuedSynchronizer.Node var1;
AbstractQueuedSynchronizer.Node var2;
return (var1 = this.head) != null && (var2 = var1.next) != null && !var2.isShared() && var2.thread != null;
}
public final boolean hasQueuedPredecessors() {
AbstractQueuedSynchronizer.Node var1 = this.tail;
AbstractQueuedSynchronizer.Node var2 = this.head;
AbstractQueuedSynchronizer.Node var3;
return var2 != var1 && ((var3 = var2.next) == null || var3.thread != Thread.currentThread());
}
public final int getQueueLength() {
int var1 = 0;
for(AbstractQueuedSynchronizer.Node var2 = this.tail; var2 != null; var2 = var2.prev) {
if (var2.thread != null) {
++var1;
}
}
return var1;
}
public final Collection<Thread> getQueuedThreads() {
ArrayList var1 = new ArrayList();
for(AbstractQueuedSynchronizer.Node var2 = this.tail; var2 != null; var2 = var2.prev) {
Thread var3 = var2.thread;
if (var3 != null) {
var1.add(var3);
}
}
return var1;
}
public final Collection<Thread> getExclusiveQueuedThreads() {
ArrayList var1 = new ArrayList();
for(AbstractQueuedSynchronizer.Node var2 = this.tail; var2 != null; var2 = var2.prev) {
if (!var2.isShared()) {
Thread var3 = var2.thread;
if (var3 != null) {
var1.add(var3);
}
}
}
return var1;
}
public final Collection<Thread> getSharedQueuedThreads() {
ArrayList var1 = new ArrayList();
for(AbstractQueuedSynchronizer.Node var2 = this.tail; var2 != null; var2 = var2.prev) {
if (var2.isShared()) {
Thread var3 = var2.thread;
if (var3 != null) {
var1.add(var3);
}
}
}
return var1;
}
public String toString() {
int var1 = this.getState();
String var2 = this.hasQueuedThreads() ? "non" : "";
return super.toString() + "[State = " + var1 + ", " + var2 + "empty queue]";
}
final boolean isOnSyncQueue(AbstractQueuedSynchronizer.Node var1) {
if (var1.waitStatus != -2 && var1.prev != null) {
return var1.next != null ? true : this.findNodeFromTail(var1);
} else {
return false;
}
}
private boolean findNodeFromTail(AbstractQueuedSynchronizer.Node var1) {
for(AbstractQueuedSynchronizer.Node var2 = this.tail; var2 != var1; var2 = var2.prev) {
if (var2 == null) {
return false;
}
}
return true;
}
final boolean transferForSignal(AbstractQueuedSynchronizer.Node var1) {
if (!compareAndSetWaitStatus(var1, -2, 0)) {
return false;
} else {
AbstractQueuedSynchronizer.Node var2 = this.enq(var1);
int var3 = var2.waitStatus;
if (var3 > 0 || !compareAndSetWaitStatus(var2, var3, -1)) {
LockSupport.unpark(var1.thread);
}
return true;
}
}
final boolean transferAfterCancelledWait(AbstractQueuedSynchronizer.Node var1) {
if (compareAndSetWaitStatus(var1, -2, 0)) {
this.enq(var1);
return true;
} else {
while(!this.isOnSyncQueue(var1)) {
Thread.yield();
}
return false;
}
}
final int fullyRelease(AbstractQueuedSynchronizer.Node var1) {
boolean var2 = true;
int var4;
try {
int var3 = this.getState();
if (!this.release(var3)) {
throw new IllegalMonitorStateException();
}
var2 = false;
var4 = var3;
} finally {
if (var2) {
var1.waitStatus = 1;
}
}
return var4;
}
public final boolean owns(AbstractQueuedSynchronizer.ConditionObject var1) {
return var1.isOwnedBy(this);
}
public final boolean hasWaiters(AbstractQueuedSynchronizer.ConditionObject var1) {
if (!this.owns(var1)) {
throw new IllegalArgumentException("Not owner");
} else {
return var1.hasWaiters();
}
}
public final int getWaitQueueLength(AbstractQueuedSynchronizer.ConditionObject var1) {
if (!this.owns(var1)) {
throw new IllegalArgumentException("Not owner");
} else {
return var1.getWaitQueueLength();
}
}
public final Collection<Thread> getWaitingThreads(AbstractQueuedSynchronizer.ConditionObject var1) {
if (!this.owns(var1)) {
throw new IllegalArgumentException("Not owner");
} else {
return var1.getWaitingThreads();
}
}
private final boolean compareAndSetHead(AbstractQueuedSynchronizer.Node var1) {
return unsafe.compareAndSwapObject(this, headOffset, (Object)null, var1);
}
private final boolean compareAndSetTail(AbstractQueuedSynchronizer.Node var1, AbstractQueuedSynchronizer.Node var2) {
return unsafe.compareAndSwapObject(this, tailOffset, var1, var2);
}
private static final boolean compareAndSetWaitStatus(AbstractQueuedSynchronizer.Node var0, int var1, int var2) {
return unsafe.compareAndSwapInt(var0, waitStatusOffset, var1, var2);
}
private static final boolean compareAndSetNext(AbstractQueuedSynchronizer.Node var0, AbstractQueuedSynchronizer.Node var1, AbstractQueuedSynchronizer.Node var2) {
return unsafe.compareAndSwapObject(var0, nextOffset, var1, var2);
}
static {
try {
stateOffset = unsafe.objectFieldOffset(AbstractQueuedSynchronizer.class.getDeclaredField("state"));
headOffset = unsafe.objectFieldOffset(AbstractQueuedSynchronizer.class.getDeclaredField("head"));
tailOffset = unsafe.objectFieldOffset(AbstractQueuedSynchronizer.class.getDeclaredField("tail"));
waitStatusOffset = unsafe.objectFieldOffset(AbstractQueuedSynchronizer.Node.class.getDeclaredField("waitStatus"));
nextOffset = unsafe.objectFieldOffset(AbstractQueuedSynchronizer.Node.class.getDeclaredField("next"));
} catch (Exception var1) {
throw new Error(var1);
}
}
public class ConditionObject implements Condition, Serializable {
private static final long serialVersionUID = 1173984872572414699L;
private transient AbstractQueuedSynchronizer.Node firstWaiter;
private transient AbstractQueuedSynchronizer.Node lastWaiter;
private static final int REINTERRUPT = 1;
private static final int THROW_IE = -1;
public ConditionObject() {
}
private AbstractQueuedSynchronizer.Node addConditionWaiter() {
AbstractQueuedSynchronizer.Node var1 = this.lastWaiter;
if (var1 != null && var1.waitStatus != -2) {
this.unlinkCancelledWaiters();
var1 = this.lastWaiter;
}
AbstractQueuedSynchronizer.Node var2 = new AbstractQueuedSynchronizer.Node(Thread.currentThread(), -2);
if (var1 == null) {
this.firstWaiter = var2;
} else {
var1.nextWaiter = var2;
}
this.lastWaiter = var2;
return var2;
}
private void doSignal(AbstractQueuedSynchronizer.Node var1) {
do {
if ((this.firstWaiter = var1.nextWaiter) == null) {
this.lastWaiter = null;
}
var1.nextWaiter = null;
} while(!AbstractQueuedSynchronizer.this.transferForSignal(var1) && (var1 = this.firstWaiter) != null);
}
private void doSignalAll(AbstractQueuedSynchronizer.Node var1) {
this.lastWaiter = this.firstWaiter = null;
AbstractQueuedSynchronizer.Node var2;
do {
var2 = var1.nextWaiter;
var1.nextWaiter = null;
AbstractQueuedSynchronizer.this.transferForSignal(var1);
var1 = var2;
} while(var2 != null);
}
private void unlinkCancelledWaiters() {
AbstractQueuedSynchronizer.Node var1 = this.firstWaiter;
AbstractQueuedSynchronizer.Node var3;
for(AbstractQueuedSynchronizer.Node var2 = null; var1 != null; var1 = var3) {
var3 = var1.nextWaiter;
if (var1.waitStatus != -2) {
var1.nextWaiter = null;
if (var2 == null) {
this.firstWaiter = var3;
} else {
var2.nextWaiter = var3;
}
if (var3 == null) {
this.lastWaiter = var2;
}
} else {
var2 = var1;
}
}
}
public final void signal() {
if (!AbstractQueuedSynchronizer.this.isHeldExclusively()) {
throw new IllegalMonitorStateException();
} else {
AbstractQueuedSynchronizer.Node var1 = this.firstWaiter;
if (var1 != null) {
this.doSignal(var1);
}
}
}
public final void signalAll() {
if (!AbstractQueuedSynchronizer.this.isHeldExclusively()) {
throw new IllegalMonitorStateException();
} else {
AbstractQueuedSynchronizer.Node var1 = this.firstWaiter;
if (var1 != null) {
this.doSignalAll(var1);
}
}
}
public final void awaitUninterruptibly() {
AbstractQueuedSynchronizer.Node var1 = this.addConditionWaiter();
int var2 = AbstractQueuedSynchronizer.this.fullyRelease(var1);
boolean var3 = false;
while(!AbstractQueuedSynchronizer.this.isOnSyncQueue(var1)) {
LockSupport.park(this);
if (Thread.interrupted()) {
var3 = true;
}
}
if (AbstractQueuedSynchronizer.this.acquireQueued(var1, var2) || var3) {
AbstractQueuedSynchronizer.selfInterrupt();
}
}
private int checkInterruptWhileWaiting(AbstractQueuedSynchronizer.Node var1) {
return Thread.interrupted() ? (AbstractQueuedSynchronizer.this.transferAfterCancelledWait(var1) ? -1 : 1) : 0;
}
private void reportInterruptAfterWait(int var1) throws InterruptedException {
if (var1 == -1) {
throw new InterruptedException();
} else {
if (var1 == 1) {
AbstractQueuedSynchronizer.selfInterrupt();
}
}
}
public final void await() throws InterruptedException {
if (Thread.interrupted()) {
throw new InterruptedException();
} else {
AbstractQueuedSynchronizer.Node var1 = this.addConditionWaiter();
int var2 = AbstractQueuedSynchronizer.this.fullyRelease(var1);
int var3 = 0;
while(!AbstractQueuedSynchronizer.this.isOnSyncQueue(var1)) {
LockSupport.park(this);
if ((var3 = this.checkInterruptWhileWaiting(var1)) != 0) {
break;
}
}
if (AbstractQueuedSynchronizer.this.acquireQueued(var1, var2) && var3 != -1) {
var3 = 1;
}
if (var1.nextWaiter != null) {
this.unlinkCancelledWaiters();
}
if (var3 != 0) {
this.reportInterruptAfterWait(var3);
}
}
}
public final long awaitNanos(long var1) throws InterruptedException {
if (Thread.interrupted()) {
throw new InterruptedException();
} else {
AbstractQueuedSynchronizer.Node var3 = this.addConditionWaiter();
int var4 = AbstractQueuedSynchronizer.this.fullyRelease(var3);
long var5 = System.nanoTime() + var1;
int var7;
for(var7 = 0; !AbstractQueuedSynchronizer.this.isOnSyncQueue(var3); var1 = var5 - System.nanoTime()) {
if (var1 <= 0L) {
AbstractQueuedSynchronizer.this.transferAfterCancelledWait(var3);
break;
}
if (var1 >= 1000L) {
LockSupport.parkNanos(this, var1);
}
if ((var7 = this.checkInterruptWhileWaiting(var3)) != 0) {
break;
}
}
if (AbstractQueuedSynchronizer.this.acquireQueued(var3, var4) && var7 != -1) {
var7 = 1;
}
if (var3.nextWaiter != null) {
this.unlinkCancelledWaiters();
}
if (var7 != 0) {
this.reportInterruptAfterWait(var7);
}
return var5 - System.nanoTime();
}
}
public final boolean awaitUntil(Date var1) throws InterruptedException {
long var2 = var1.getTime();
if (Thread.interrupted()) {
throw new InterruptedException();
} else {
AbstractQueuedSynchronizer.Node var4 = this.addConditionWaiter();
int var5 = AbstractQueuedSynchronizer.this.fullyRelease(var4);
boolean var6 = false;
int var7 = 0;
while(!AbstractQueuedSynchronizer.this.isOnSyncQueue(var4)) {
if (System.currentTimeMillis() > var2) {
var6 = AbstractQueuedSynchronizer.this.transferAfterCancelledWait(var4);
break;
}
LockSupport.parkUntil(this, var2);
if ((var7 = this.checkInterruptWhileWaiting(var4)) != 0) {
break;
}
}
if (AbstractQueuedSynchronizer.this.acquireQueued(var4, var5) && var7 != -1) {
var7 = 1;
}
if (var4.nextWaiter != null) {
this.unlinkCancelledWaiters();
}
if (var7 != 0) {
this.reportInterruptAfterWait(var7);
}
return !var6;
}
}
public final boolean await(long var1, TimeUnit var3) throws InterruptedException {
long var4 = var3.toNanos(var1);
if (Thread.interrupted()) {
throw new InterruptedException();
} else {
AbstractQueuedSynchronizer.Node var6 = this.addConditionWaiter();
int var7 = AbstractQueuedSynchronizer.this.fullyRelease(var6);
long var8 = System.nanoTime() + var4;
boolean var10 = false;
int var11;
for(var11 = 0; !AbstractQueuedSynchronizer.this.isOnSyncQueue(var6); var4 = var8 - System.nanoTime()) {
if (var4 <= 0L) {
var10 = AbstractQueuedSynchronizer.this.transferAfterCancelledWait(var6);
break;
}
if (var4 >= 1000L) {
LockSupport.parkNanos(this, var4);
}
if ((var11 = this.checkInterruptWhileWaiting(var6)) != 0) {
break;
}
}
if (AbstractQueuedSynchronizer.this.acquireQueued(var6, var7) && var11 != -1) {
var11 = 1;
}
if (var6.nextWaiter != null) {
this.unlinkCancelledWaiters();
}
if (var11 != 0) {
this.reportInterruptAfterWait(var11);
}
return !var10;
}
}
final boolean isOwnedBy(AbstractQueuedSynchronizer var1) {
return var1 == AbstractQueuedSynchronizer.this;
}
protected final boolean hasWaiters() {
if (!AbstractQueuedSynchronizer.this.isHeldExclusively()) {
throw new IllegalMonitorStateException();
} else {
for(AbstractQueuedSynchronizer.Node var1 = this.firstWaiter; var1 != null; var1 = var1.nextWaiter) {
if (var1.waitStatus == -2) {
return true;
}
}
return false;
}
}
protected final int getWaitQueueLength() {
if (!AbstractQueuedSynchronizer.this.isHeldExclusively()) {
throw new IllegalMonitorStateException();
} else {
int var1 = 0;
for(AbstractQueuedSynchronizer.Node var2 = this.firstWaiter; var2 != null; var2 = var2.nextWaiter) {
if (var2.waitStatus == -2) {
++var1;
}
}
return var1;
}
}
protected final Collection<Thread> getWaitingThreads() {
if (!AbstractQueuedSynchronizer.this.isHeldExclusively()) {
throw new IllegalMonitorStateException();
} else {
ArrayList var1 = new ArrayList();
for(AbstractQueuedSynchronizer.Node var2 = this.firstWaiter; var2 != null; var2 = var2.nextWaiter) {
if (var2.waitStatus == -2) {
Thread var3 = var2.thread;
if (var3 != null) {
var1.add(var3);
}
}
}
return var1;
}
}
}
static final class Node {
static final AbstractQueuedSynchronizer.Node SHARED = new AbstractQueuedSynchronizer.Node();
static final AbstractQueuedSynchronizer.Node EXCLUSIVE = null;
static final int CANCELLED = 1;
static final int SIGNAL = -1;
static final int CONDITION = -2;
static final int PROPAGATE = -3;
volatile int waitStatus;
volatile AbstractQueuedSynchronizer.Node prev;
volatile AbstractQueuedSynchronizer.Node next;
volatile Thread thread;
AbstractQueuedSynchronizer.Node nextWaiter;
final boolean isShared() {
return this.nextWaiter == SHARED;
}
final AbstractQueuedSynchronizer.Node predecessor() throws NullPointerException {
AbstractQueuedSynchronizer.Node var1 = this.prev;
if (var1 == null) {
throw new NullPointerException();
} else {
return var1;
}
}
Node() {
}
Node(Thread var1, AbstractQueuedSynchronizer.Node var2) {
this.nextWaiter = var2;
this.thread = var1;
}
Node(Thread var1, int var2) {
this.waitStatus = var2;
this.thread = var1;
}
}
}
1.在ReentrantLock内部定义了一个Sync的内部类,该类继承AbstractQueuedSynchronized,对该抽象类的部分方法做了实现;
2.定义了两个子类:
1>FairSync 公平锁的实现
2>NonfairSync 非公平锁的实现
这两个类都继承自Sync,也就是间接继承了AbstractQueuedSynchronized,所以这一个ReentrantLock同时具备公平与非公平特性。
//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by FernFlower decompiler)
//
package java.util.concurrent.locks;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.Serializable;
import java.util.Collection;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.AbstractQueuedSynchronizer.ConditionObject;
public class ReentrantLock implements Lock, Serializable {
private static final long serialVersionUID = 7373984872572414699L;
private final ReentrantLock.Sync sync;
public ReentrantLock() {
this.sync = new ReentrantLock.NonfairSync();
}
public ReentrantLock(boolean var1) {
this.sync = (ReentrantLock.Sync)(var1 ? new ReentrantLock.FairSync() : new ReentrantLock.NonfairSync());
}
public void lock() {
this.sync.lock();
}
public void lockInterruptibly() throws InterruptedException {
this.sync.acquireInterruptibly(1);
}
public boolean tryLock() {
return this.sync.nonfairTryAcquire(1);
}
public boolean tryLock(long var1, TimeUnit var3) throws InterruptedException {
return this.sync.tryAcquireNanos(1, var3.toNanos(var1));
}
public void unlock() {
this.sync.release(1);
}
public Condition newCondition() {
return this.sync.newCondition();
}
public int getHoldCount() {
return this.sync.getHoldCount();
}
public boolean isHeldByCurrentThread() {
return this.sync.isHeldExclusively();
}
public boolean isLocked() {
return this.sync.isLocked();
}
public final boolean isFair() {
return this.sync instanceof ReentrantLock.FairSync;
}
protected Thread getOwner() {
return this.sync.getOwner();
}
public final boolean hasQueuedThreads() {
return this.sync.hasQueuedThreads();
}
public final boolean hasQueuedThread(Thread var1) {
return this.sync.isQueued(var1);
}
public final int getQueueLength() {
return this.sync.getQueueLength();
}
protected Collection<Thread> getQueuedThreads() {
return this.sync.getQueuedThreads();
}
public boolean hasWaiters(Condition var1) {
if (var1 == null) {
throw new NullPointerException();
} else if (!(var1 instanceof ConditionObject)) {
throw new IllegalArgumentException("not owner");
} else {
return this.sync.hasWaiters((ConditionObject)var1);
}
}
public int getWaitQueueLength(Condition var1) {
if (var1 == null) {
throw new NullPointerException();
} else if (!(var1 instanceof ConditionObject)) {
throw new IllegalArgumentException("not owner");
} else {
return this.sync.getWaitQueueLength((ConditionObject)var1);
}
}
protected Collection<Thread> getWaitingThreads(Condition var1) {
if (var1 == null) {
throw new NullPointerException();
} else if (!(var1 instanceof ConditionObject)) {
throw new IllegalArgumentException("not owner");
} else {
return this.sync.getWaitingThreads((ConditionObject)var1);
}
}
public String toString() {
Thread var1 = this.sync.getOwner();
return super.toString() + (var1 == null ? "[Unlocked]" : "[Locked by thread " + var1.getName() + "]");
}
static final class FairSync extends ReentrantLock.Sync {
private static final long serialVersionUID = -3000897897090466540L;
FairSync() {
}
final void lock() {
this.acquire(1);
}
protected final boolean tryAcquire(int var1) {
Thread var2 = Thread.currentThread();
int var3 = this.getState();
if (var3 == 0) {
if (!this.hasQueuedPredecessors() && this.compareAndSetState(0, var1)) {
this.setExclusiveOwnerThread(var2);
return true;
}
} else if (var2 == this.getExclusiveOwnerThread()) {
int var4 = var3 + var1;
if (var4 < 0) {
throw new Error("Maximum lock count exceeded");
}
this.setState(var4);
return true;
}
return false;
}
}
static final class NonfairSync extends ReentrantLock.Sync {
private static final long serialVersionUID = 7316153563782823691L;
NonfairSync() {
}
final void lock() {
if (this.compareAndSetState(0, 1)) {
this.setExclusiveOwnerThread(Thread.currentThread());
} else {
this.acquire(1);
}
}
protected final boolean tryAcquire(int var1) {
return this.nonfairTryAcquire(var1);
}
}
abstract static class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = -5179523762034025860L;
Sync() {
}
abstract void lock();
final boolean nonfairTryAcquire(int var1) {
Thread var2 = Thread.currentThread();
int var3 = this.getState();
if (var3 == 0) {
if (this.compareAndSetState(0, var1)) {
this.setExclusiveOwnerThread(var2);
return true;
}
} else if (var2 == this.getExclusiveOwnerThread()) {
int var4 = var3 + var1;
if (var4 < 0) {
throw new Error("Maximum lock count exceeded");
}
this.setState(var4);
return true;
}
return false;
}
protected final boolean tryRelease(int var1) {
int var2 = this.getState() - var1;
if (Thread.currentThread() != this.getExclusiveOwnerThread()) {
throw new IllegalMonitorStateException();
} else {
boolean var3 = false;
if (var2 == 0) {
var3 = true;
this.setExclusiveOwnerThread((Thread)null);
}
this.setState(var2);
return var3;
}
}
protected final boolean isHeldExclusively() {
return this.getExclusiveOwnerThread() == Thread.currentThread();
}
final ConditionObject newCondition() {
return new ConditionObject(this);
}
final Thread getOwner() {
return this.getState() == 0 ? null : this.getExclusiveOwnerThread();
}
final int getHoldCount() {
return this.isHeldExclusively() ? this.getState() : 0;
}
final boolean isLocked() {
return this.getState() != 0;
}
private void readObject(ObjectInputStream var1) throws IOException, ClassNotFoundException {
var1.defaultReadObject();
this.setState(0);
}
}
}
通过FairSync 和NonfairSync中可以看到,加锁的时候调用的都是lock()方法,接下来,我们就从这个方法作为入口,来了解一下在ReentrantLock中是如何进行加锁的。
1>this.tryAcquire(var1):锁竞争逻辑
static final class FairSync extends ReentrantLock.Sync {
private static final long serialVersionUID = -3000897897090466540L;
FairSync() {
}
final void lock() {
this.acquire(1);
}
protected final boolean tryAcquire(int var1) {
Thread var2 = Thread.currentThread();
int var3 = this.getState();
if (var3 == 0) {
if (!this.hasQueuedPredecessors() && this.compareAndSetState(0, var1)) { // 当前线程加锁成功
this.setExclusiveOwnerThread(var2);
return true;
}
} else if (var2 == this.getExclusiveOwnerThread()) { // 当前线程持有锁,支持可重入,state+1
int var4 = var3 + var1;
if (var4 < 0) {
throw new Error("Maximum lock count exceeded");
}
this.setState(var4);
return true;
}
return false;
}
}
2>this.addWaiter(AbstractQueuedSynchronizer.Node.EXCLUSIVE):未获得锁的线程入队
private AbstractQueuedSynchronizer.Node addWaiter(AbstractQueuedSynchronizer.Node var1) {
//其中var1为AbstractQueuedSynchronizer.Node.EXCLUSIVE类型的节点,继续往下看可以知道,它就是我们链表中的头结点,无实际含义
AbstractQueuedSynchronizer.Node var2 = new AbstractQueuedSynchronizer.Node(Thread.currentThread(), var1);
AbstractQueuedSynchronizer.Node var3 = this.tail;
// 当队列为空的时候不会执行这段逻辑
if (var3 != null) {
var2.prev = var3;
if (this.compareAndSetTail(var3, var2)) {
var3.next = var2;
return var2;
}
}
this.enq(var2);
return var2;
}
private AbstractQueuedSynchronizer.Node enq(AbstractQueuedSynchronizer.Node var1) {
while(true) { // 一定要记住,这里是个循环哦,会循环执行
AbstractQueuedSynchronizer.Node var2 = this.tail;
// 1.当队列为空的时候,会先创建一个空的Node节点作为头结点
if (var2 == null) {
// 入队也存在竞争
if (this.compareAndSetHead(new AbstractQueuedSynchronizer.Node())) {
this.tail = this.head;
}
} else {
// 2.var2指的是头结点,var1指的是当前线程的节点。这段逻辑指的是使用尾插法,将节点插入到链表中
var1.prev = var2;
if (this.compareAndSetTail(var2, var1)) {
var2.next = var1;
return var2;
}
}
}
}
3>this.acquireQueued(this.addWaiter(AbstractQueuedSynchronizer.Node.EXCLUSIVE), var1)
final boolean acquireQueued(AbstractQueuedSynchronizer.Node var1, int var2) {
// var1指的是this.addWaiter(AbstractQueuedSynchronizer.Node.EXCLUSIVE)中得到的当前线程的节点Node(currentThread)
// var2指的是加锁中的1,this.acquire(1)
boolean var3 = true;
try {
boolean var4 = false;
while(true) {
// var5指的是var1的前驱节点
AbstractQueuedSynchronizer.Node var5 = var1.predecessor();
// 如果var5是头节点,则在节点阻塞之前又去尝试获取锁
// 能够获取到,则头节点出队(等待被gc回收),并且把head往后挪一个节点,新的头结点就是当前节点
if (var5 == this.head && this.tryAcquire(var2)) {
this.setHead(var1);
var5.next = null;
var3 = false;
boolean var6 = var4;
return var6;
}
// shouldParkAfterFailedAcquire(var5, var1)将头节点中的waitStatus设置为-1
// this.parkAndCheckInterrupt() 阻塞当前线程
if (shouldParkAfterFailedAcquire(var5, var1) && this.parkAndCheckInterrupt()) {
var4 = true;
}
}
} finally {
if (var3) {
this.cancelAcquire(var1);
}
}
}
private static boolean shouldParkAfterFailedAcquire(AbstractQueuedSynchronizer.Node var0, AbstractQueuedSynchronizer.Node var1) {
int var2 = var0.waitStatus;
if (var2 == -1) {
return true;
} else {
if (var2 > 0) {
do {
var1.prev = var0 = var0.prev;
} while(var0.waitStatus > 0);
var0.next = var1;
} else {
compareAndSetWaitStatus(var0, var2, -1);
}
return false;
}
}
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted(); //表示清除中断标记,如果当前线程中断,返回true,否则返回false
}
4>selfInterrupt();
static void selfInterrupt() {
Thread.currentThread().interrupt(); //为当前线程打中断标记
}
加锁的时候调用的都是lock()方法,那解锁呢,当然是从release(int var1)入手。
public final boolean release(int var1) {
if (this.tryRelease(var1)) {
AbstractQueuedSynchronizer.Node var2 = this.head;
// 加锁的时候将头结点的waitStatus设置为-1
// 这也就是为什么在shouldParkAfterFailedAcquire(p, node)中要把head节点的waitestate = 0 - > -1 ,因为持有锁的线程t0在释放锁的时候,得判断head节点的waitestate是否!=0,如果!=0成立,会再把waitstate = -1->0。
if (var2 != null && var2.waitStatus != 0) {
this.unparkSuccessor(var2);
}
return true;
} else {
return false;
}
}
protected final boolean tryRelease(int var1) {
int var2 = this.getState() - var1;
if (Thread.currentThread() != this.getExclusiveOwnerThread()) {
throw new IllegalMonitorStateException();
} else {
boolean var3 = false;
if (var2 == 0) {
var3 = true;
this.setExclusiveOwnerThread((Thread)null);
}
// state=state-1
this.setState(var2);
return var3;
}
}
private void unparkSuccessor(AbstractQueuedSynchronizer.Node var1) {
int var2 = var1.waitStatus;
if (var2 < 0) {
compareAndSetWaitStatus(var1, var2, 0);
}
AbstractQueuedSynchronizer.Node var3 = var1.next;
if (var3 == null || var3.waitStatus > 0) {
var3 = null;
for(AbstractQueuedSynchronizer.Node var4 = this.tail; var4 != null && var4 != var1; var4 = var4.prev) {
if (var4.waitStatus <= 0) {
var3 = var4;
}
}
}
if (var3 != null) {
LockSupport.unpark(var3.thread);
}
}