java并发--队列同步器原理一

独占式同步状态获取与释放原理

在同步队列中独占式获取锁的代码

public final void acquire(int arg) {
if (!tryAcquire(arg) &&      //独占
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}

上述代码主要完成了同步状态获取,节点构造 加入同步队列以及在同步队列中自旋等待的相关操作, 逻辑是 首先调用自定义同步器实现的tryAcquire方法 该方法保证线程安全的获取同步状态,如果同步状态获取失败 则构造同步节点(独占式node)并通过addWaiter将该节点加入到同步队列的尾部 最好调用 acquireQueued 是的该节点以死循环的方式获取同步状态 如果获取不到则阻塞节点中的线程

节点的构造及加入同步队列

/**
* Creates and enqueues node for given thread and mode.
*
* @param current the thread
* @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;
if (t == null) { // Must initialize
Node h = new Node(); // Dummy header
h.next = node;
node.prev = h;
if (compareAndSetHead(h)) {
tail = node;
return h;
}
}
else {
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}

节点加入同步队列后就进入了一个自旋的过程

final boolean acquireQueued(final Node node, int arg) {
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} catch (RuntimeException ex) {
cancelAcquire(node);
throw ex;
}
}

上述代码执行的结果

java.lang.Thread.State: WAITING
at sun.misc.Unsafe.park(Unsafe.java:-1)
at java.util.concurrent.locks.LockSupport.park(LockSupport.java:156)
at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:811)
at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireQueued(AbstractQueuedSynchronizer.java:842)
at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(AbstractQueuedSynchronizer.java:1178)
at concurrent.lock.Mutex.lock(Mutex.java:76)
at concurrent.lock.MutexTest$MutexRunnable.run(MutexTest.java:50)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:439)
at java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:303)
at java.util.concurrent.FutureTask.run(FutureTask.java:138)
at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:895)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:918)
at java.lang.Thread.run(Thread.java:662)

线程会处于waiting的状态