AQS 的源码剖析
NonfairSync 加锁源码剖析
static final class NonfairSync extends Sync {
private static final long serialVersionUID = 7316153563782823691L;
/**
* Performs lock. Try immediate barge, backing up to normal
* acquire on failure.
*/
final void lock() {
// 插队获取，不必排队，能够进步吞吐量
if (compareAndSetState(0, 1))
// 获取锁胜利 设置 ExclusiveOwnerThread = currentThread
setExclusiveOwnerThread(Thread.currentThread());
else
// 重要 AQS 排队
acquire(1);
}
protected final boolean tryAcquire(int acquires) {return nonfairTryAcquire(acquires);
}
}
/**
* Acquires in exclusive mode, ignoring interrupts. Implemented
* by invoking at least once {@link #tryAcquire},
* returning on success. Otherwise the thread is queued, possibly
* repeatedly blocking and unblocking, invoking {@link
* #tryAcquire} until success. This method can be used
* to implement method {@link Lock#lock}.
*
* @param arg the acquire argument. This value is conveyed to
* {@link #tryAcquire} but is otherwise uninterpreted and
* can represent anything you like.
*/
public final void acquire(int arg) {// tryAcquire(1) 模版办法 , 再次尝试加锁, 加锁胜利返回 Ture，失败返回 False
// addWaiter(null) CAS 自旋 排到队尾
// acquireQueued CAS 自旋 获取锁 / 中断（park）if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
// 获取锁后 检测到 interrept = true，自我中断一次
selfInterrupt();}
````
/**
* Performs non-fair tryLock. tryAcquire is implemented in
* subclasses, but both need nonfair try for trylock method.
*/
final boolean nonfairTryAcquire(int acquires) {final Thread current = Thread.currentThread();
int c = getState();
// 再次获取锁，偏心锁须要判断是不是第一个期待的 Node
if (c == 0) {if (compareAndSetState(0, acquires)) {setExclusiveOwnerThread(current);
return true;
}
}
// 获取重入锁，重入锁间接 state + 1 就好
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;
}/**
* Creates and enqueues node for current thread and given mode.
*
* @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared
* @return the new node
*/
// 插入 队列尾部
private Node addWaiter(Node mode) {Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
Node pred = tail;
// 疾速入队 优化
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
// 自旋 入队
enq(node);
return node;
}
/**
* Inserts node into queue, initializing if necessary. See picture above.
* @param node the node to insert
* @return node's predecessor
*/
private Node enq(final Node node) {for (;;) {
Node t = tail;
// Head 为 null 时，先建设 Head
if (t == null) { // Must initialize
if (compareAndSetHead(new Node()))
tail = head;
// 把以后节点插入最初
} else {
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
/**
* Acquires in exclusive uninterruptible mode for thread already in
* queue. Used by condition wait methods as well as acquire.
*
* @param node the node
* @param arg the acquire argument
* @return {@code true} if interrupted while waiting
*/
// CAS 自旋 要么胜利，要么中断
final boolean acquireQueued(final Node node, int arg) {
// 加锁失败 标记
boolean failed = true;
try {
// 加锁过程中 是否中断过
boolean interrupted = false;
for (;;) {
// node.predecessor 前置节点
final Node p = node.predecessor();
// 判断 是否是 第一个排队期待的线程
// head 个别是 正在解决的线程节点，队列中第二个节点才是第一个排队期待的线程节点
if (p == head && tryAcquire(arg)) {setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
// 没有加锁胜利，判断是否中断
// 前一个 Node.waitStatus == -1 (前一个线程正在期待锁资源)
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {if (failed)
cancelAcquire(node);
}
}
/**
* Checks and updates status for a node that failed to acquire.
* Returns true if thread should block. This is the main signal
* control in all acquire loops. Requires that pred == node.prev.
*
* @param pred node's predecessor holding status
* @param node the node
* @return {@code true} if thread should block
*/
// 通过 pred.waitStatus 判断 node 是否须要挂起
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
// 前继节点（上一个线程）还没拿到锁，挂起
// pred.waitStatus == Node.SIGNAL，前继节点 在期待资源
if (ws == Node.SIGNAL)
/*
* This node has already set status asking a release
* to signal it, so it can safely park.
*/
return true;
if (ws > 0) {
/*
* Predecessor was cancelled. Skip over predecessors and
* indicate retry.
*/
// 删除 node 后面的有效节点
do {node.prev = pred = pred.prev;} while (pred.waitStatus > 0);
pred.next = node;
} else {
/*
* waitStatus must be 0 or PROPAGATE. Indicate that we
* need a signal, but don't park yet. Caller will need to
* retry to make sure it cannot acquire before parking.
*/
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
/**
* Convenience method to park and then check if interrupted
*
* @return {@code true} if interrupted
*/
private final boolean parkAndCheckInterrupt() {
// LockSupport.park 中断
LockSupport.park(this);
// 返回是否被中断标记
return Thread.interrupted();}