AQS
AbstractQuenedSynchronizer抽象的队列式同步器。是除了java自带的synchronized关键字之外的锁机制。
-
AQS主要结构
AQS就是基于CLH队列，用volatile修饰共享变量state，线程通过CAS去改变状态符，成功则获取锁成功，失败则进入等待队列(CLH队列)，等待被唤醒。
ReentrantLock源码
-
ReentrantLock也是基于AQS实现的框架他的结构为
主要里面包含着一个sync来控制
- lock方法(没特殊标志的都用公平锁来讲解,ReentrantLock初始化的时候其实是非公平锁)
public void lock() {
sync.lock();
}
final void lock() {
acquire(1);
}
java.util.concurrent.locks.AbstractQueuedSynchronizer#acquire
public final void acquire(int arg) {
//tryAcquire(arg)尝试获取锁，如果能获取成功的话则!tryAcquire(arg)为false,&&后面的逻辑也不会执行
//如果获取失败的话会执行 acquireQueued(addWaiter(Node.EXCLUSIVE), arg))添加入进入队列
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
tryAcquire方法(可重入锁实现地方)
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
//获取当前上锁的次数
int c = getState();
if (c == 0) {
//判断他前面是否还有人排队，如果没有人的话尝试获取锁
if (!hasQueuedPredecessors() &&
compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
//可重入锁实现地方，判断当前获取锁的线程是不是持有线程，如果是的话则加一层锁
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
进队列方法acquireQueued(addWaiter(Node.EXCLUSIVE), arg))方法
private Node addWaiter(Node mode) {
//新建一个node
Node node = new Node(Thread.currentThread(), mode);
//尾巴节点
Node pred = tail;
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
//如果tail节点为空，说明是第一次入队，还没有初始化，则初始化队列
enq(node);
return node;
}
private Node enq(final Node node) {
for (;;) {
//t置为当前tail的指针
Node t = tail;
if (t == null) {
//第一次进来初始化头尾节点(初始化头结点之后再复制尾巴节点)
if (compareAndSetHead(new Node()))
tail = head;
} else {
//第一次循环完后，将
node.prev = t;
//将tail置为node节点的指针(虽然上面把tail的指针复制给了t，但是这边直接改变了tail的指针，跟t没干系，t还是指向原来的tail指针)
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
//获取前置节点
final Node p = node.predecessor();
//如果他的前置节点是head节点，则再尝试获取锁
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
//如果获取不到锁
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
/*
* This node has already set status asking a release to signal it, so it can safely park.
* 节点已经设置好为 Node.SIGNAL状态，可以放心停止线程
*/
return true;
if (ws > 0) {
/*
* Predecessor was cancelled. Skip over predecessors and indicate retry.
* 前置节点已经cancel了，直接剔除
*/
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
/*
* 将他的waitStatus置为SIGNAL
*/
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
public static void park(Object blocker) {
Thread t = Thread.currentThread();
setBlocker(t, blocker);
//停止线程
UNSAFE.park(false, 0L);
setBlocker(t, null);
}
- unlock解锁方法
调用 sync.release(1);
public void unlock() {
sync.release(1);
}
public final boolean release(int arg) {
//释放锁
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
//唤醒等待线程
unparkSuccessor(h);
return true;
}
return false;
}
protected final boolean tryRelease(int releases) {
int c = getState() - releases;
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
//当前的锁持有数为0时(重入的话C可能不等于0)，则释放锁，返回true
if (c == 0) {
free = true;
setExclusiveOwnerThread(null);
}
//设置当前的锁状态为剩余c
setState(c);
return free;
}
private void unparkSuccessor(Node node) {
int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
Node s = node.next;
if (s == null || s.waitStatus > 0) {
s = null;
for (Node t = tail; t != null && t != node; t = t.prev)
if (t.waitStatus <= 0)
s = t;
}
if (s != null)
LockSupport.unpark(s.thread);
}
//当调用线程的unpark方法后会进入到休眠之前的方法
final boolean acquireQueued(final Node node, int arg) {
...
for (;;) {
//唤醒后会进入该自旋方法
//获取前置节点
final Node p = node.predecessor();
//如果他的前置节点是head节点，则再尝试获取锁
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
}
}
}
private void setHead(Node node) {
//将head修改为该节点的前置节点
head = node;
//置空
node.thread = null;
node.prev = null;
}
- 公平锁与非公平锁的差别,
1、非公平锁在lock调用的时候会直接尝试获取锁，不会看队列前面有没有等待的，公平锁则是要看前面有没有已经在排队的
非公平锁他尝试获取不到锁之后也会进入CLH队列，如果进入到队列之后就是排队了。这个跟公平锁是一样的
static final class NonfairSync extends Sync {
private static final long serialVersionUID = 7316153563782823691L;
final void lock() {
//差异1、非公平锁一进来就会尝试获取锁，如果没有获取到锁才进入队列
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
}
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
static final class FairSync extends Sync {
private static final long serialVersionUID = -3000897897090466540L;
final void lock() {
acquire(1);
}
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
//差异2、公平锁这边会判断前面是否还有节点，非公平锁不会判断吗，直接cas尝试获取
if (!hasQueuedPredecessors() &&
compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0)
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
}
网友评论