await

调用sync.acquireSharedInterruptibly

public void await() throws InterruptedException {sync.acquireSharedInterruptibly(1);
}

sync.acquireSharedInterruptibly
调用tryAcquireShared方法返回<0执行doAcquireSharedInterruptibly

public final void acquireSharedInterruptibly(int arg) throws InterruptedException {if (Thread.interrupted())throw new InterruptedException();if (tryAcquireShared(arg) < 0)doAcquireSharedInterruptibly(arg);
}

tryAcquireShared
尝试获取共享锁,获取成功返回1，否则-1

protected int tryAcquireShared(int acquires) {return (getState() == 0) ? 1 : -1;
}

doAcquireSharedInterruptibly

private void doAcquireSharedInterruptibly(int arg)throws InterruptedException {final Node node = addWaiter(Node.SHARED);boolean failed = true;try {for (;;) {final Node p = node.predecessor();//如果前一个node为队头，则通过tryAcquireShared尝试获取共享锁if (p == head) {int r = tryAcquireShared(arg);if (r >= 0) {//获取到锁执行setHeadAndPropagate(node, r);p.next = null; // help GCfailed = false;return;}}if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())throw new InterruptedException();}} finally {//产生异常执行if (failed)cancelAcquire(node);}
}

addWaiter
调用addWaiter方法把队尾设置为当前node；如果队尾为空或者设置失败则调用enq方法

private Node addWaiter(Node mode) {Node node = new Node(Thread.currentThread(), mode);// Try the fast path of enq; backup to full enq on failureNode pred = tail;if (pred != null) {node.prev = pred;if (compareAndSetTail(pred, node)) {pred.next = node;return node;}}enq(node);return node;
}

enq
调用enq方法队尾为空则创建空的队尾和队头，否则重新设置队尾为当前node，设置成功返回。enq和addWaiter方法不同在于enq循环执行一定会执行成功，不存在失败情况

private Node enq(final Node node) {for (;;) {Node t = tail;if (t == null) { // Must initializeif (compareAndSetHead(new Node()))tail = head;} else {node.prev = t;if (compareAndSetTail(t, node)) {t.next = node;return t;}}}
}

predecessor
调用predecessor方法获取前一个node

final Node predecessor() throws NullPointerException {Node p = prev;if (p == null)throw new NullPointerException();elsereturn p;
}static final int CANCELLED = 1; //取消
static final int SIGNAL = -1; //下个节点需要被唤醒
static final int CONDITION = -2; //线程在等待条件触发
static final int PROPAGATE = -3; //（共享锁）状态需要向后传播

shouldParkAfterFailedAcquire
获取当前node的前一个note的线程等待状态，如果为SIGNAL,那么返回true,大于0通过循环将当前节点之前所有取消状态的节点移出队列;其他状时,利用compareAndSetWaitStatus使前节点的状态为-1;如果是第一次await时ws状态是0，多次await时ws状态是0，最后肯定返回true

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {int ws = pred.waitStatus;if (ws == Node.SIGNAL)return true;if (ws > 0) {do {node.prev = pred = pred.prev;} while (pred.waitStatus > 0);pred.next = node;} else {compareAndSetWaitStatus(pred, ws, Node.SIGNAL);}return false;
}

parkAndCheckInterrupt
调用park并返回线程是否已经中断

private final boolean parkAndCheckInterrupt() {LockSupport.park(this);return Thread.interrupted();
}

park
调用UNSAFE.park阻塞当前线程

public static void park(Object blocker) {Thread t = Thread.currentThread();setBlocker(t, blocker);UNSAFE.park(false, 0L);setBlocker(t, null);
}

setBlocker
在当前线程t的parkBlockerOffset位置设置blocker的引用

private static void setBlocker(Thread t, Object arg) {// Even though volatile, hotspot doesn't need a write barrier here.UNSAFE.putObject(t, parkBlockerOffset, arg);
}

UNSAFE.park

/*** 阻塞一个线程直到<a href="#unpark"><code>unpark</code></a>出现、线程* 被中断或者timeout时间到期。如果一个<code>unpark</code>调用已经出现了，* 这里只计数。timeout为0表示永不过期.当<code>isAbsolute</code>为true时，* timeout是相对于新纪元之后的毫秒。否则这个值就是超时前的纳秒数。这个方法执行时* 也可能不合理地返回(没有具体原因)* * @param isAbsolute true if the timeout is specified in milliseconds from*                   the epoch.*                   如果为true timeout的值是一个相对于新纪元之后的毫秒数* @param time either the number of nanoseconds to wait, or a time in*             milliseconds from the epoch to wait for.*             可以是一个要等待的纳秒数，或者是一个相对于新纪元之后的毫秒数直到*             到达这个时间点*/
UNSAFE.park(false, 0L);

countDown

调用sync.releaseShared

public void countDown() {sync.releaseShared(1);
}

releaseShared
执行tryReleaseShared成功后执行doReleaseShared

public final boolean releaseShared(int arg) {if (tryReleaseShared(arg)) {doReleaseShared();return true;}return false;
}

tryReleaseShared
更新state值为state-1，如果state新值为0返回true，否则false

protected boolean tryReleaseShared(int releases) {// Decrement count; signal when transition to zerofor (;;) {int c = getState();if (c == 0)return false;int nextc = c-1;if (compareAndSetState(c, nextc))return nextc == 0;}
}

doReleaseShared
只要等待队列有数据，获取队头等待状态，队头状态=-1其他node为等待时，则把队头等待状态置为初始，且调用unparkSuccessor方法；队头状态=0时，把队头状态置为-3传播到下一node

private void doReleaseShared() {/** Ensure that a release propagates, even if there are other* in-progress acquires/releases.  This proceeds in the usual* way of trying to unparkSuccessor of head if it needs* signal. But if it does not, status is set to PROPAGATE to* ensure that upon release, propagation continues.* Additionally, we must loop in case a new node is added* while we are doing this. Also, unlike other uses of* unparkSuccessor, we need to know if CAS to reset status* fails, if so rechecking.*/for (;;) {Node h = head;if (h != null && h != tail) {int ws = h.waitStatus;if (ws == Node.SIGNAL) {if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))continue;            // loop to recheck casesunparkSuccessor(h);}else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))continue;                // loop on failed CAS}if (h == head)                   // loop if head changedbreak;}
}

unparkSuccessor
上面调用unparkSuccessor时，node的状态已经更改为0，且node.next存在，执行unpark方法

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);
}

unpark
unpark执行完之后是如何更改head的?

public static void unpark(Thread thread) {if (thread != null)UNSAFE.unpark(thread);
}

UNSAFE.unpark

/*** Releases the block on a thread created by * <a href="#park"><code>park</code></a>.  This method can also be used* to terminate a blockage caused by a prior call to <code>park</code>.* This operation is unsafe, as the thread must be guaranteed to be* live.  This is true of Java, but not native code.* 释放被<a href="#park"><code>park</code></a>创建的在一个线程上的阻塞.这个* 方法也可以被使用来终止一个先前调用<code>park</code>导致的阻塞.* 这个操作操作时不安全的,因此线程必须保证是活的.这是java代码不是native代码。* @param thread the thread to unblock.*           要解除阻塞的线程*/
UNSAFE.unpark(thread);