JUC同步锁原理源码解析三----CountDownLatch、CyclicBarrier

1.CountDownLatch的来源

A synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads completes.

​ CountDownLatch的出现是为了一个或者多个线程等待一系列操作完成后，继续往下执行。因而在日常工作中最经常使用的场景：一个大任务需要等待子任务完成后，继续往后执行其他操作。

2.CountDownLatch的底层实现

​ CountDownLatch的底层实现依旧依赖于AQS的共享锁机制。

3.CyclicBarrier的来源

A synchronization aid that allows a set of threads to all wait for each other to reach a common barrier point.

​ 允许一组线程都等待彼此到达一个公共屏障点。 也即允许一组线程阻塞在同一个点上，直到所有的线程都到达这个点上，再继续往下执行。

4.CyclicBarrier的实现

​ CyclicBarrier的实现底层依赖于ReentransLock。由于CyclicBarrier是可以复用，判断其参与者以及滚动到下一个代的操作，需要保证其原子性。

Node节点

 static final class Node {
        /** 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 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;

        static final int PROPAGATE = -3;

        volatile int waitStatus;

        volatile Node prev;

        volatile Node next;

        volatile Thread thread;

        Node nextWaiter;
}

AbstractQueuedSynchronizer类

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {

     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;//最重要的一个变量

}

ConditionObject类

public class ConditionObject implements Condition, java.io.Serializable {
        private static final long serialVersionUID = 1173984872572414699L;
        /** First node of condition queue. */
        private transient Node firstWaiter;
        /** Last node of condition queue. */
        private transient Node lastWaiter;
}

accquire方法

public final void acquire(int arg) {
    if (!tryAcquire(arg) &&//尝试获取锁
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))//如果获取锁失败，添加到队列中，由于ReentrantLock是独占锁所以节点必须是EXCLUSIVE类型
        selfInterrupt();//添加中断标识位
}

addWaiter方法

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);//cas失败后执行入队操作，继续尝试
     return node;
 }

enq方法

private Node enq(final Node node) {
    for (;;) {
        Node t = tail;//获取尾指针
        if (t == null) { //代表当前队列没有节点
            if (compareAndSetHead(new Node()))//将当前节点置为头结点
                tail = head;
        } else {//当前队列有节点
            node.prev = t;//
            if (compareAndSetTail(t, node)) {//将当前节点置为尾结点
                t.next = node;
                return t;
            }
        }
    }
}

acquireQueued方法

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)) {//前驱节点等于头节点尝试cas抢锁。
                setHead(node);//抢锁成功将当前节点设置为头节点
                p.next = null; // help GC  当头结点置空
                failed = false;
                return interrupted;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&//当队列中有节点在等待，判断是否应该阻塞
                parkAndCheckInterrupt())//阻塞等待，检查中断标识位
                interrupted = true;//将中断标识位置为true
        }
    } finally {
        if (failed)//
            cancelAcquire(node);//取消当前节点
    }
}

 private void cancelAcquire(Node node) {
     // Ignore if node doesn't exist
     if (node == null)//当前节点为空直接返回
         return;

     node.thread = null;//要取消了将当前节点的线程置为空
     // Skip cancelled predecessors
     Node pred = node.prev;//获取到当前节点的前驱节点
     while (pred.waitStatus > 0)//如果当前节点的前驱节点的状态大于0,代表是取消状态，一直找到不是取消状态的节点
         node.prev = pred = pred.prev;
     Node predNext = pred.next;//将当前要取消的节点断链

     node.waitStatus = Node.CANCELLED;//将当前节点的等待状态置为CANCELLED
     // If we are the tail, remove ourselves.
     if (node == tail && compareAndSetTail(node, pred)) {//如果当前节点是尾结点，将尾结点替换为浅语节点
         compareAndSetNext(pred, predNext, null);//将当前节点的下一个节点置为空，因为当前节点是最后一个节点没有next指针
     } else {
         // If successor needs signal, try to set pred's next-link
         // so it will get one. Otherwise wake it up to propagate.
         int ws;
         if (pred != head &&//前驱节点不等于头结点
             ((ws = pred.waitStatus) == Node.SIGNAL ||//前驱节点的状态不等于SIGNAL
              (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&//前驱节点的状态小于0，并且cas将前驱节点的等待置为SIGNAL
             pred.thread != null) {//前驱节点的线程补位空
             Node next = node.next;//获取当前节点的next指针
             if (next != null && next.waitStatus <= 0)//如果next指针不等于空并且等待状态小于等于0，标识节点有效
                 compareAndSetNext(pred, predNext, next);//将前驱节点的next指针指向下一个有效节点
         } else {
             unparkSuccessor(node);//唤醒后续节点 条件：1.前驱节点是头结点 2.当前节点不是signal,在ReentransLock中基本不会出现，在读写锁时就会出现
         }

         node.next = node; // help GC 将引用指向自身
     }
 }

 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);//cas将当前节点置为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;//将s置为空
         for (Node t = tail; t != null && t != node; t = t.prev)//从尾结点遍历找到一个不是取消状态的节点
             if (t.waitStatus <= 0)
                 s = t;
     }
     if (s != null)//如果s不等于空
         LockSupport.unpark(s.thread);//唤醒当前节点s
 }

shouldParkAfterFailedAcquire方法

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;//获取上一个节点的等待状态
    if (ws == Node.SIGNAL)//如果状态为SIGNAL,代表后续节点有节点可以唤醒，可以安心阻塞去
        /*
             * This node has already set status asking a release
             * to signal it, so it can safely park.
             */
        return true;
    if (ws > 0) {//如果当前状态大于0，代表节点为CANCELLED状态
        /*
             * Predecessor was cancelled. Skip over predecessors and
             * indicate retry.
             */
        do {
            node.prev = pred = pred.prev;//从尾节点开始遍历，找到下一个状态不是CANCELLED的节点。将取消节点断链移除
        } 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.
             */
        //这里需要注意ws>0时，已经找到了一个不是取消状态的前驱节点。
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);//将找到的不是CANCELLED节点的前驱节点，将其等待状态置为SIGNAL
    }
    return false;
}

cancelAcquire方法

 private void cancelAcquire(Node node) {
     // Ignore if node doesn't exist
     if (node == null)//当前节点为空直接返回
         return;

     node.thread = null;//要取消了将当前节点的线程置为空
     // Skip cancelled predecessors
     Node pred = node.prev;//获取到当前节点的前驱节点
     while (pred.waitStatus > 0)//如果当前节点的前驱节点的状态大于0,代表是取消状态，一直找到不是取消状态的节点
         node.prev = pred = pred.prev;
     Node predNext = pred.next;//将当前要取消的节点断链

     node.waitStatus = Node.CANCELLED;//将当前节点的等待状态置为CANCELLED
     // If we are the tail, remove ourselves.
     if (node == tail && compareAndSetTail(node, pred)) {//如果当前节点是尾结点，将尾结点替换为浅语节点
         compareAndSetNext(pred, predNext, null);//将当前节点的下一个节点置为空，因为当前节点是最后一个节点没有next指针
     } else {
         // If successor needs signal, try to set pred's next-link
         // so it will get one. Otherwise wake it up to propagate.
         int ws;
         if (pred != head &&//前驱节点不等于头结点
             ((ws = pred.waitStatus) == Node.SIGNAL ||//前驱节点的状态不等于SIGNAL
              (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&//前驱节点的状态小于0，并且cas将前驱节点的等待置为SIGNAL
             pred.thread != null) {//前驱节点的线程补位空
             Node next = node.next;//获取当前节点的next指针
             if (next != null && next.waitStatus <= 0)//如果next指针不等于空并且等待状态小于等于0，标识节点有效
                 compareAndSetNext(pred, predNext, next);//将前驱节点的next指针指向下一个有效节点
         } else {
             unparkSuccessor(node);//唤醒后续节点 条件：1.前驱节点是头结点 2.当前节点不是signal,在ReentransLock中基本不会出现，在读写锁时就会出现
         }

         node.next = node; // help GC 将引用指向自身
     }
 }

unparkSuccessor方法

 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);//cas将当前节点置为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;//将s置为空
         for (Node t = tail; t != null && t != node; t = t.prev)//从尾结点遍历找到一个不是取消状态的节点
             if (t.waitStatus <= 0)
                 s = t;
     }
     if (s != null)//如果s不等于空
         LockSupport.unpark(s.thread);//唤醒当前节点s
 }

release方法

public final boolean release(int arg) {
    if (tryRelease(arg)) {//子类实现如何释放锁
        Node h = head;//获取到头结点
        if (h != null && h.waitStatus != 0)//获取到头结点，如果头结点不为空，等待状态不为0,唤醒后续节点
            unparkSuccessor(h);
        return true;
    }
    return false;
}

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)//如果等待状态小于0,标识节点属于有效节点
        compareAndSetWaitStatus(node, ws, 0);//将当前节点的等待状态置为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);//唤醒当前等待的有效节点S
}

acquireShared方法

public final void acquireShared(int arg) {
    if (tryAcquireShared(arg) < 0)//由子类实现
        doAcquireShared(arg);
}

doAcquireShared方法

private void doAcquireShared(int arg) {
    final Node node = addWaiter(Node.SHARED);//将共享节点也即读线程入队并返回
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();//找到节点的前驱节点
            if (p == head) {//如果前驱节点等于头结点
                int r = tryAcquireShared(arg);//尝试获取共享锁数量
                if (r >= 0) {//如果锁的数量大于0，表示还有多余的共享锁。这里等于0也需要进一步判断。由于如果当执行到这里时，有另外的线程释放了共享锁，如果不进行判断，将会导致释放锁的线程没办法唤醒其他线程。所以这里会伪唤醒一个节点，唤醒的节点后续如果没有锁释放，依旧阻塞在当前parkAndCheckInterrupt方法中
                    setHeadAndPropagate(node, r);//将当前节点的等待状态设置为Propagate。
                    p.next = null; // help GC
                    if (interrupted)//判断是会否中断过
                        selfInterrupt();//设置中断标识位
                    failed = false;
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node) &&//判断是否应该阻塞等待
                parkAndCheckInterrupt方法中())//阻塞并检查中断标识
                interrupted = true;//重置中断标识位
        }
    } finally {
        if (failed)//如果失败
            cancelAcquire(node);//取消节点
    }
}

setHeadAndPropagate方法

private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head; // Record old head for check below
        setHead(node);//将当前节点置为头结点
        /*
         * Try to signal next queued node if:
         *   Propagation was indicated by caller,
         *     or was recorded (as h.waitStatus either before
         *     or after setHead) by a previous operation
         *     (note: this uses sign-check of waitStatus because
         *      PROPAGATE status may transition to SIGNAL.)
         * and
         *   The next node is waiting in shared mode,
         *     or we don't know, because it appears null
         *
         * The conservatism in both of these checks may cause
         * unnecessary wake-ups, but only when there are multiple
         * racing acquires/releases, so most need signals now or soon
         * anyway.
         */
        if (propagate > 0 //可获取的共享锁也即读锁的数量，对于ReentrantReadWriteLock而言，永远都是1，所以会继续唤醒下一个读线程
            || h == null //如果旧的头结点为空
            || h.waitStatus < 0 ||//头结点的等待状态不为0
            (h = head) == null || h.waitStatus < 0) {//旧头节点不为空并且等待状态小于0也即是有效节点
            Node s = node.next;//获取到node的下一个节点
            if (s == null || s.isShared())//如果node的下一个节点为空或者是共享节点
                doReleaseShared();//唤醒下一个线程
        }
    }

releaseShared方法

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {//子类实现释放锁
        doReleaseShared();//唤醒后续线程
        return true;//释放成功
    }
    return false;//释放是吧
}

doReleaseShared方法

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) {//如果状态为SIGNAL
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))//cas将SIGNAL状态置为0。SIGNAL标识后续有线程需要唤醒
                    continue;            // loop to recheck cases
                unparkSuccessor(h);//唤醒后续线程
            }
            else if (ws == 0 &&//如果当前状态为0。表示有线程将其置为0
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))//cas将0状态置为PROPAGATE。在多个共享锁同时释放时，方便继续进行读传播，唤醒后续节点
                continue;                // loop on failed CAS
        }
        if (h == head)//如果头结点没有改变，证明没有必要继续循环等待了，直接退出吧，如果头结点放生变化，可能有其他线程释放了锁。
            break;
    }
}

await()

public final void await() throws InterruptedException {
    if (Thread.interrupted())//线程是否发生中断，是，就抛出中断异常
        throw new InterruptedException();
    Node node = addConditionWaiter();//加入条件等待队列
    int savedState = fullyRelease(node);//释放锁，并返回。因为当前线程需要等待
    int interruptMode = 0;
    while (!isOnSyncQueue(node)) {//判断是否在竞争队列中。AQS分为两个队列一个是竞争队列，等待调度执行，一个是等待队列等待在ConditionObject上。
        LockSupport.park(this);//阻塞等待
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)//重新去获取锁并判断当前中断模式不是THROW_IE
        interruptMode = REINTERRUPT;//将中断模式置为REINTERRUPT
    if (node.nextWaiter != null) // clean up if cancelled如果当前节点的下一个节点不为空
        unlinkCancelledWaiters();//清除等待队列中已经取消的节点
    if (interruptMode != 0)//如果当前中断模式不等于0
        reportInterruptAfterWait(interruptMode);
}

private void reportInterruptAfterWait(int interruptMode)
    throws InterruptedException {
    if (interruptMode == THROW_IE)//如果是THROW_IE直接抛出异常
        throw new InterruptedException();
    else if (interruptMode == REINTERRUPT)//如果是REINTERRUPT
        selfInterrupt();//重置中断标识位
}

addConditionWaiter方法

private Node addConditionWaiter() {
    Node t = lastWaiter;//获取到最后一个节点
    // If lastWaiter is cancelled, clean out.
    if (t != null && t.waitStatus != Node.CONDITION) {//最后一个节点不等于空，并且等待状态不等于CONDITION
        unlinkCancelledWaiters();//将取消节点断链,标准的链表操作
        t = lastWaiter;//获取到最后一个有效的节点
    }
    Node node = new Node(Thread.currentThread(), Node.CONDITION);//将当前节点封装成node
    if (t == null)//如果最后一个节点为空，表示当前节点是第一个入队的节点
        firstWaiter = node;
    else
        t.nextWaiter = node;//否则将当前node挂在链表末尾
    lastWaiter = node;//设置最后节点的指针指向当前node
    return node;
}

fullyRelease方法

final int fullyRelease(Node node) {
    boolean failed = true;
    try {
        int savedState = getState();//获取当前state状态
        if (release(savedState)) {//释放锁尝试
            failed = false;
            return savedState;//返回
        } else {
            throw new IllegalMonitorStateException();//抛出释放锁异常
        }
    } finally {
        if (failed)
            node.waitStatus = Node.CANCELLED;//如果失败将节点置为取消状态
    }
}

public final boolean release(int arg) {
    if (tryRelease(arg)) {//尝试释放锁，在CyclciBarrier中由于线程需要去阻塞，所以需要将锁释放，后续重新拿锁
        Node h = head;
        if (h != null && h.waitStatus != 0)//从头结点开始唤醒
            unparkSuccessor(h);
        return true;
    }
    return false;
}

isOnSyncQueue方法

final boolean isOnSyncQueue(Node node) {
    if (node.waitStatus == Node.CONDITION || node.prev == null)//如果当前节点是Condition或者node.pre节点为空，标识不在竞争队列中，返回faslse
        return false;
    if (node.next != null) // If has successor, it must be on queue  表示在竞争队列中
        return true;
    /*
         * node.prev can be non-null, but not yet on queue because
         * the CAS to place it on queue can fail. So we have to
         * traverse from tail to make sure it actually made it.  It
         * will always be near the tail in calls to this method, and
         * unless the CAS failed (which is unlikely), it will be
         * there, so we hardly ever traverse much.
         */
    return findNodeFromTail(node);//从竞争队列的尾结点开始找当前node，找到就返回true，否则为false
}

private boolean findNodeFromTail(Node node) {
    Node t = tail;//获取到尾结点
    for (;;) {
        if (t == node)
            return true;
        if (t == null)
            return false;
        t = t.prev;
    }
}

findNodeFromTail方法

private int checkInterruptWhileWaiting(Node node) {
    return Thread.interrupted() ?//判断当前是否中断过
        (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) ://如果移动到竞争队列中并入队成功，返回THROW_IE，否则返回REINTERRUPT
    0;//没有中断过直接返回0
}

//走到这里表示条件队列的条件满足，可以将节点移动到竞争队列中执行
final boolean transferAfterCancelledWait(Node node) {
    if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {//尝试将当前为Condition的节点置为0，并移动到竞争队列中
        enq(node);
        return true;
    }
    /*
         * If we lost out to a signal(), then we can't proceed
         * until it finishes its enq().  Cancelling during an
         * incomplete transfer is both rare and transient, so just
         * spin.
         */
    while (!isOnSyncQueue(node))//如果不在竞争队列中返回false
        Thread.yield();
    return false;
}

signalAll方法

public final void signalAll() {
    if (!isHeldExclusively())//是不是持有独占锁
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;//获取等待队列的第一个节点
    if (first != null)//如果节点不为空
        doSignalAll(first);//唤醒所有线程
}

//从头指针一直遍历等待队列，将其移动到竞争队列中
private void doSignalAll(Node first) {
    lastWaiter = firstWaiter = null;
    do {
        Node next = first.nextWaiter;
        first.nextWaiter = null;
        transferForSignal(first);//
        first = next;
    } while (first != null);
}

transferForSignal方法

final boolean transferForSignal(Node node) {
    /*
     * If cannot change waitStatus, the node has been cancelled.
     */
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))//cas自旋将其等待状态改为0
        return false;

    /*
     * Splice onto queue and try to set waitStatus of predecessor to
     * indicate that thread is (probably) waiting. If cancelled or
     * attempt to set waitStatus fails, wake up to resync (in which
     * case the waitStatus can be transiently and harmlessly wrong).
     */
    Node p = enq(node);//将其放入竞争队列
    int ws = p.waitStatus;//获取节点的等待状态
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))//如果节点是取消状态或者cas将其置为signal失败，唤醒当前线程，让他自己处理，后续在竞争队列中会自动移除取消节点
        LockSupport.unpark(node.thread);
    return true;
}

总结：AQS提供了统一的模板，对于如何入队出队以及线程的唤醒都由AQS提供默认的实现，只需要子类实现自己上锁和解锁的逻辑。

基本使用

import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class RWDictionary {
    private final Map<String, String> m = new TreeMap<String, String>();
    private final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
    private final Lock r = rwl.readLock();
    private final Lock w = rwl.writeLock();

    public String get(String key) {
      r.lock();
      try { return m.get(key); }
      finally { r.unlock(); }
    }
    public String[] allKeys() {
      r.lock();
      try {
          return (String[])m.keySet().toArray();
      }
      finally { r.unlock(); }
    }
    public String put(String key, String value) {
      w.lock();
      try { return m.put(key, value); }
      finally { w.unlock(); }
    }
    public void clear() {
      w.lock();
      try { m.clear(); }
      finally { w.unlock(); }
    }
}

Sync类

import java.util.concurrent.CountDownLatch;

public class Driver {
    void main() throws InterruptedException {
        int N = 10;
      CountDownLatch startSignal = new CountDownLatch(1);
      CountDownLatch doneSignal = new CountDownLatch(N);

      for (int i = 0; i < N; ++i) // create and start threads
        new Thread(new Worker(startSignal, doneSignal)).start();

//      doSomethingElse();            // don't let run yet
      startSignal.countDown();      // let all threads proceed
//      doSomethingElse();
      doneSignal.await();           // wait for all to finish
    }

  class Worker implements Runnable {
    private final CountDownLatch startSignal;
    private final CountDownLatch doneSignal;
    Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {
      this.startSignal = startSignal;
      this.doneSignal = doneSignal;
    }
    public void run() {
      try {
        startSignal.await();
        doWork();
        doneSignal.countDown();
      } catch (InterruptedException ex) {} // return;
    }

    void doWork() {  }
  }

CountDownLatch的构造器

public CountDownLatch(int count) {
    if (count < 0) throw new IllegalArgumentException("count < 0");
    this.sync = new Sync(count);
}

private static final class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 4982264981922014374L;

        Sync(int count) {
            setState(count);//设置共享锁的数量
        }

        int getCount() {
            return getState();
        }

        protected int tryAcquireShared(int acquires) {
            return (getState() == 0) ? 1 : -1;//只要state不为0,表示还有共享锁，可以直接获取
        }

        protected boolean tryReleaseShared(int releases) {
            // Decrement count; signal when transition to zero
            for (;;) {
                int c = getState();//获取state变量
                if (c == 0)//共享锁数量为0，表示没有共享锁需要释放
                    return false;
                int nextc = c-1;//将共享锁数量减1
                if (compareAndSetState(c, nextc))//cas设置state变量
                    return nextc == 0;//如果共享搜数量已经为0了，表示可以释放锁，无需等待。继续执行吧
            }
        }
    }

    private final Sync sync;

countDown方法

public void countDown() {
    sync.releaseShared(1);
}
//调用AQS的模板方法
public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {//子类自己实现，请查看上面sync类中的tryReleaseShared
        doReleaseShared();//详情请查看上文AQS的doReleaseShared
        return true;
    }
    return false;
}

await()方法:

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

public final void acquireSharedInterruptibly(int arg)
    throws InterruptedException {
    if (Thread.interrupted())//判断是否中断过
        throw new InterruptedException();//抛出异常
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);//实现方法与doAcquireShared不可中断的方法一样，只不过增加了抛出中断异常的判断
}

await(long timeout, TimeUnit unit)方法

public boolean await(long timeout, TimeUnit unit)
    throws InterruptedException {
    return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}

public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout)
    throws InterruptedException {
    if (Thread.interrupted())//判断是否中断过
        throw new InterruptedException();//抛出异常
    return tryAcquireShared(arg) >= 0 ||//CountDownLatch自己实现释放锁的方法
        doAcquireSharedNanos(arg, nanosTimeout);//doAcquireSharedNanos与doAcquireShare基本一致，不过是增加了超时以及异常抛出的处理，这里不过多赘述
}

基本使用

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierDemo {
    public static void main(String[] args) {
        //CyclicBarrier barrier = new CyclicBarrier(10);
        CyclicBarrier barrier = new CyclicBarrier(10, () -> System.out.println("线程到达"));
        for(int i=0; i<30; i++) {
                new Thread(()->{
                    try {
                        barrier.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (BrokenBarrierException e) {
                        e.printStackTrace();
                    }
                }).start();
        }
    }
}

CyclicBarrier类的基本变量

public class CyclicBarrier {
    private static class Generation {
        boolean broken = false;
    }
    /** The lock for guarding barrier entry */
    private final ReentrantLock lock = new ReentrantLock();
    /** Condition to wait on until tripped */
    private final Condition trip = lock.newCondition();
    /** The number of parties */
    private final int parties;
    /* The command to run when tripped */
    private final Runnable barrierCommand;
    /** The current generation */
    private Generation generation = new Generation();
    /**
     * Number of parties still waiting. Counts down from parties to 0
     * on each generation.  It is reset to parties on each new
     * generation or when broken.
     */
    private int count;

    public CyclicBarrier(int parties, Runnable barrierAction) {
        if (parties <= 0) throw new IllegalArgumentException();
        this.parties = parties;
        this.count = parties;
        this.barrierCommand = barrierAction;
    }

    public CyclicBarrier(int parties) {
        this(parties, null);
    }
}

await()方法

public int await() throws InterruptedException, BrokenBarrierException {
    try {
        return dowait(false, 0L);//不待超时，所以置为0；
    } catch (TimeoutException toe) {
        throw new Error(toe); // cannot happen
    }
}

public int await(long timeout, TimeUnit unit)
        throws InterruptedException,
               BrokenBarrierException,
               TimeoutException {
        return dowait(true, unit.toNanos(timeout));//带超时时间
}

dowait方法

private int dowait(boolean timed, long nanos)
    throws InterruptedException, BrokenBarrierException,
TimeoutException {
    final ReentrantLock lock = this.lock;//获取到锁
    lock.lock();//获取到锁
    try {
        final Generation g = generation;//获取到当前轮次

        if (g.broken)//是否被打破过
            throw new BrokenBarrierException();//抛出异常

        if (Thread.interrupted()) {//如果当前线程被中过
            breakBarrier();//打破当前轮次，唤醒所有线程
            throw new InterruptedException();
        }

        int index = --count;//先将等待的线程数减1
        if (index == 0) {  // tripped 代表当前线程是最后一个线程，才会去唤醒等待的线程
            boolean ranAction = false;
            try {
                final Runnable command = barrierCommand;//获取回调任务
                if (command != null)//任务补位空执行
                    command.run();
                ranAction = true;
                nextGeneration();//滚动到下一个轮次，并唤醒等待线程
                return 0;
            } finally {
                if (!ranAction)//发生异常，没有正常结束
                    breakBarrier();//打破当前轮次，唤醒所有线程
            }
        }

        // loop until tripped, broken, interrupted, or timed out
        for (;;) {
            try {
                if (!timed)//是否设置超时时间
                    trip.await();//没有设置超时时间就等待
                else if (nanos > 0L)//如果超时时间大于0
                    nanos = trip.awaitNanos(nanos);//等待固定的超时时间，此方法与trip.await()类似，只是加了超时时间的处理而已
            } catch (InterruptedException ie) {
                if (g == generation && ! g.broken) {//g还是当前轮次，并且没有被打破
                    breakBarrier();//由于发生异常，打破当前轮次，唤醒所有线程。
                    throw ie;
                } else {
                    // We're about to finish waiting even if we had not
                    // been interrupted, so this interrupt is deemed to
                    // "belong" to subsequent execution.
                    Thread.currentThread().interrupt();//中断
                }
            }

            if (g.broken)//轮次已经被打破，抛出中断异常
                throw new BrokenBarrierException();

            if (g != generation)//当前轮次不等于现在的轮次。g获取时是临时变量，中间可能会发生变化
                return index;

            if (timed && nanos <= 0L) {//设置了超时，并且已经超时
                breakBarrier();//打破当前轮次，唤醒所有线程
                throw new TimeoutException();
            }
        }
    } finally {
        lock.unlock();//解锁
    }
}

breakBarrier方法

private void breakBarrier() {
    generation.broken = true;//broken设置为true
    count = parties;//count设置为参与者的数量
    trip.signalAll();//唤醒所有等待的线程
}

nextGeneration方法

private void nextGeneration() {
        // signal completion of last generation
        trip.signalAll();//唤醒所有线程,AQS的方法
        // set up next generation
        count = parties;//count设置为参与者的数量
        generation = new Generation();//赋值一个新的Generation，代表不同的轮次
    }

本文章只是JUC 中AQS的一部分，后续的文章会对基于AQS锁实现的子类进行拓展讲解，以上文章内容基于个人以及结合别人文章的理解，如果有问题或者不当之处欢迎大家留言交流。由于为了保证观看流畅性，其中一部分源码有重复的地方。请见谅