AQS
AQS 内部维护了一个FIFO双向队列,Node内部类就是队列的节点;(关键: FIFO、双向)
head,此队列头结点
tail,此队列尾节点
state,表示同步状态,共享模式下,一个同步锁可能被多个线程持有,所以用int表示
/**
* Head of the wait queue, lazily initialized. Except for
* initialization, it is modified only via method setHead. Note:
* If head exists, its waitStatus is guaranteed not to be
* CANCELLED.
*/
private transient volatile Node head;
/**
* Tail of the wait queue, lazily initialized. Modified only via
* method enq to add new wait node.
*/
private transient volatile Node tail;
/**
* The synchronization state.
*/
private volatile int state;
/** Marker to indicate a node is waiting in shared mode */
static final Node SHARED = new Node();
/** Marker to indicate a node is waiting in exclusive mode */
static final Node EXCLUSIVE = null;
waitStatus 节点状态
CANCELLED 1 被取消的线程
SIGNAL -1 需要被唤醒的线程
CONDITION -2 表示线程正在等待条件
PROPAGATE -3 表示下一个acquisitionShared应 无条件地传播
0 默认值,无意义
/** waitStatus value to indicate thread has cancelled */
static final int CANCELLED = 1;
/** waitStatus value to indicate successor's thread needs unparking */
static final int SIGNAL = -1;
/** waitStatus value to indicate thread is waiting on condition */
static final int CONDITION = -2;
/**
* waitStatus value to indicate the next acquireShared should
* unconditionally propagate
*/
static final int PROPAGATE = -3;
/**
* Status field, taking on only the values:
* SIGNAL: The successor of this node is (or will soon be)
* blocked (via park), so the current node must
* unpark its successor when it releases or
* cancels. To avoid races, acquire methods must
* first indicate they need a signal,
* then retry the atomic acquire, and then,
* on failure, block.
* CANCELLED: This node is cancelled due to timeout or interrupt.
* Nodes never leave this state. In particular,
* a thread with cancelled node never again blocks.
* CONDITION: This node is currently on a condition queue.
* It will not be used as a sync queue node
* until transferred, at which time the status
* will be set to 0. (Use of this value here has
* nothing to do with the other uses of the
* field, but simplifies mechanics.)
* PROPAGATE: A releaseShared should be propagated to other
* nodes. This is set (for head node only) in
* doReleaseShared to ensure propagation
* continues, even if other operations have
* since intervened.
* 0: None of the above
*
* The values are arranged numerically to simplify use.
* Non-negative values mean that a node doesn't need to
* signal. So, most code doesn't need to check for particular
* values, just for sign.
*
* The field is initialized to 0 for normal sync nodes, and
* CONDITION for condition nodes. It is modified using CAS
* (or when possible, unconditional volatile writes).
*/
volatile int waitStatus;
/**
* Link to predecessor node that current node/thread relies on
* for checking waitStatus. Assigned during enqueuing, and nulled
* out (for sake of GC) only upon dequeuing. Also, upon
* cancellation of a predecessor, we short-circuit while
* finding a non-cancelled one, which will always exist
* because the head node is never cancelled: A node becomes
* head only as a result of successful acquire. A
* cancelled thread never succeeds in acquiring, and a thread only
* cancels itself, not any other node.
*/
volatile Node prev;
/**
* Link to the successor node that the current node/thread
* unparks upon release. Assigned during enqueuing, adjusted
* when bypassing cancelled predecessors, and nulled out (for
* sake of GC) when dequeued. The enq operation does not
* assign next field of a predecessor until after attachment,
* so seeing a null next field does not necessarily mean that
* node is at end of queue. However, if a next field appears
* to be null, we can scan prev's from the tail to
* double-check. The next field of cancelled nodes is set to
* point to the node itself instead of null, to make life
* easier for isOnSyncQueue.
*/
volatile Node next;
/**
* The thread that enqueued this node. Initialized on
* construction and nulled out after use.
*/
volatile Thread thread;
/**
* Link to next node waiting on condition, or the special
* value SHARED. Because condition queues are accessed only
* when holding in exclusive mode, we just need a simple
* linked queue to hold nodes while they are waiting on
* conditions. They are then transferred to the queue to
* re-acquire. And because conditions can only be exclusive,
* we save a field by using special value to indicate shared
* mode.
*/
Node nextWaiter;
- 独占模式 EXCLUSIVE
- 共享模式 SHARED
获取锁的两种模式
- 尝试获取锁,但实际是否获得锁并不重要。获取到锁直接返回true;没有获取到锁直接返回false。
不会阻塞
- 没有获得锁的话就会一直阻塞,直到获得锁
这两种模式在AQS中都有实现,分别是 tryAcquire()
和 acquire()
tryAcquire()
tryAcquire()用于尝试获取锁,获取结果会立即返回。
AQS中的tryAcquire()方法并没有提供具体的实现,要求AQS的继承者必须自己实现此方法,例如ReentryLock、CountDownLacth,都自己实现了tryAcquire()方法。如果没有实现电话,会抛出UnsupportedOperationException() 异常
/**
* Attempts to acquire in exclusive mode. This method should query
* if the state of the object permits it to be acquired in the
* exclusive mode, and if so to acquire it.
*
* <p>This method is always invoked by the thread performing
* acquire. If this method reports failure, the acquire method
* may queue the thread, if it is not already queued, until it is
* signalled by a release from some other thread. This can be used
* to implement method {@link Lock#tryLock()}.
*
* <p>The default
* implementation throws {@link UnsupportedOperationException}.
*
* @param arg the acquire argument. This value is always the one
* passed to an acquire method, or is the value saved on entry
* to a condition wait. The value is otherwise uninterpreted
* and can represent anything you like.
* @return {@code true} if successful. Upon success, this object has
* been acquired.
* @throws IllegalMonitorStateException if acquiring would place this
* synchronizer in an illegal state. This exception must be
* thrown in a consistent fashion for synchronization to work
* correctly.
* @throws UnsupportedOperationException if exclusive mode is not supported
*/
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}
acquire()
用于在独占模式下获取锁,直到成功
执行步骤:
- 执行tryAcquire(),如果获取到锁(返回ture),直接执行selfInterrupt()
- tryAcquire()没有获取到锁(返回false),执行 acquireQueued(addWaiter(Node.EXCLUSIVE), arg),
节点入队
- addWaiter(Node.EXCLUSIVE)
- 将当前线程封装成Node,并加入等待队列
- acquireQueued(node,arg)
- 如果获取锁失败,会执行 if 的代码 selfInterrupt(),将当前线程中断
/**
* 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) {
if (
!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
)
selfInterrupt();
}
1. 将当前线程,封装成node
2. 如果尾节点不为空,则用cas将当前节点设为尾节点
3. 如果cas失败,则进入完整入队方法 enq()
/**
* 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;
}
入队方法,将节点node加入队列
1. 如果尾节点为空(说明队列还未初始化),则用cas初始化一个队列
2. 如果尾节点不为空,则用cas将当前节点设为尾节点
/**
* 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;
if (t == null) { // Must initialize
if (compareAndSetHead(new Node()))
tail = head;
} else {
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
配合release方法,对线程进行挂起和响应。
对已经在队列中的线程以不间断的方式进行获取。被条件等待方法和获取方法所调用。
1. 判断
cancelAcquire(node)
/**
* 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
*/
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
// 判断当前节点的前一节点是否是头结点
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
// 判断当前线程是否需要被挂起,如果成功挂起,则interrupted置为true
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
*/
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.
*/
// 前置节点状态为signal,前置节点也在等待拿锁,那当前节点就可以挂起等待。
return true;
if (ws > 0) {
/*
* Predecessor was cancelled. Skip over predecessors and
* indicate retry.
*/
// 前置节点状态为cancel,则删除前面所有waitStatus为CANCEL的节点
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.
*/
// 其他状态,将前置节点waitStatus置为SIGNAL
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(this);
return Thread.interrupted();
}
只有Head的后一个节点有资格拿锁,且需要保证同时有且只有这一个节点在拿锁。
被挂起的线程何时被唤醒呢?
释放锁
tryRelease()由继承子类自行定义逻辑
protected boolean tryRelease(int arg) {
throw new UnsupportedOperationException();
}
1. 调用tryRealease()方法尝试释放锁
2. 如果tryRealease()成功,返回true,则唤醒队列的下一节点
/**
* Releases in exclusive mode. Implemented by unblocking one or
* more threads if {@link #tryRelease} returns true.
* This method can be used to implement method {@link Lock#unlock}.
*
* @param arg the release argument. This value is conveyed to
* {@link #tryRelease} but is otherwise uninterpreted and
* can represent anything you like.
* @return the value returned from {@link #tryRelease}
*/
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
node 为头结点(也就是刚刚执行结束的线程)
1. 将node的waitStatus 通过cas置为0
2. 寻找node的继任者,通常及时node的next节点,但有时也不是,这是就要查找
3. 查找逻辑:
从队列尾节点开始向前寻找 最靠近node的waitStatus<=0 的节点
/**
* Wakes up node's successor, if one exists.
*
* @param node the node
*/
private void unparkSuccessor(Node node) {
/*
* If status is negative (i.e., possibly needing signal) try
* to clear in anticipation of signalling. It is OK if this
* fails or if status is changed by waiting thread.
*/
int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
被唤醒的线程被保留在后继者中,通常就是下一个节点。
但如果被取消或显然是空的,就从尾部向前遍历,以找到实际的未被取消的继任者。
/*
* Thread to unpark is held in successor, which is normally
* just the next node. But if cancelled or apparently null,
* traverse backwards from tail to find the actual
* non-cancelled successor.
*/
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);
}
!为什么寻找successor时要从后往前查找?
首先要明确,从后往前寻找时,并不是找到第一个符合s == null || s.waitStatus > 0
就行,而是一直找到离node最近的那一个。所以就有了这个疑问,既然要找的是离node最近的一个,为什么不直接从node开始向后寻找呢?
原因在入队方法enq()逻辑里
重点关注入队语句,将一个节点node入队,即放在尾节点Tail后,有两个步骤:
//
tail.next = node;
node.prev = tail;
从逻辑上讲,单线程环境下,这两个步骤谁先谁后无所谓。而多线程下,就有问题了(因为这两个操作不是原子操作),回到AQS的入队操作:
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
上面逻辑中,compareAndSetTail(t, node)
,利用cas将node的prev指针设置为tail,成功后再执行if括号里的内容;但是括号里的操作不是同步操作。这个时刻,node的prev指针已经指向了tail,但是tail.next
指针还没建立。若此时其他线程调用了unpark操作,那tail的next指针就无法成功建立。
所以从头往后就无法完整遍历队列,从后往前就可以
private Node enq(final Node node) {
for (;;) { // 自旋操作
Node t = tail;
if (t == null) { // Must initialize
if (compareAndSetHead(new Node()))
tail = head;
} else {
node.prev = t;
// 入队语句在这里
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
Head头结点
头结点是个虚节点,不存储实际线程;一个线程的节点被设为head节点,意外该线程节点完成,出队。
private void setHead(Node node) {
head = node;
node.thread = null;
node.prev = null;
}