CountDownLatch、CyclicBarrier、 Semaphore、ReentrantLock和AQS

学习自http://www.importnew.com/21889.html

https://blog.csdn.net/yanyan19880509/article/details/52349056

https://www.cnblogs.com/waterystone/p/4920797.html

下面看一个例子大家就清楚CountDownLatch的用法了：

执行结果：

在new CountDownLatch(2)的时候

对于 CountDownLatch 来说，state=2表示所有调用await方法的线程都应该阻塞，等到同一个latch被调用两次countDown后才能唤醒沉睡的线程

为了充分了解AQS里的链表，这里假设上面挂起等待的线程数为2个

当latch被成功减到0后，AQS的state就成了0。那个成功减到0的那个线程。然后节点3被唤醒了。当节点3醒来后，发现自己是通知状态，然后删除自己，唤醒节点4。

上面的流程，如果落实到代码，把 state置为0的那个线程，会判断head指向节点的状态，如果为通知状态，则唤醒后续节点，即线程3节点，然后head指向线程3节点，head指向的旧节点会被删除掉。当线程3恢复执行后，发现自身为通知状态，又会把head指向线程4节点，然后删除自身节点，并唤醒线程4。

线程节点的状态是什么时候设置上去的？其实，一个线程在阻塞之前，就会把它前面的节点设置为通知状态，这样便可以实现链式唤醒机制了。

AQS是一些同步的抽象，简单介绍一下

两大元素

1.volatile int state->共享资源

2.FIFO线程等待队列，多线程争用资源被阻塞时会进入此队列

AQS定义两种资源占用方式：

1.Exclusive（独占，只有一个线程能执行，如ReentrantLock）

2.Share（共享，多个线程可同时执行，如Semaphore/CountDownLatch）

自定义同步器在实现时只需要实现state的增减即可，至于具体线程等待队列的维护（如获取资源失败入队/唤醒出队等），AQS已经在底层实现好了

自定义同步器的时候，主要实现以下几种方法：

1.isHeldExclusively()，该线程是否正在独占资源，只有用到condition的时候才需要去实现这个方法

2.tryAcquire(int)，独占。尝试获取资源，成功返回true，失败返回fasle

3.tryRelease(int)

4.tryAcquireShared(int)，共享。尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。

5.tryReleaseShared(int)

以ReentrantLock为例

state初始化为0，表示未锁定状态。A线程lock()时，会调用tryAcquire()独占该锁并将state+1。此后，其他线程再tryAcquire()时就会失败，直到A线程unlock()到state=0（即释放锁）为止，其它线程才有机会获取该锁。当然，释放锁之前，A线程自己是可以重复获取此锁的（state会累加），这就是可重入的概念。但要注意，获取多少次就要释放多么次，这样才能保证state是能回到零态的。

以CountDownLatch以例

任务分为N个子线程去执行，state也初始化为N（注意N要与线程个数一致）。这N个子线程是并行执行的，每个子线程执行完后countDown()一次，state会CAS减1。等到所有子线程都执行完后(即state=0)，会unpark()主调用线程，然后主调用线程就会从await()函数返回，继续后余动作。

深入看一下CountDownLatch

首先是他的内部类

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

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

private final Sync sync;

tryAcquireShared不知道有啥用，理论上应该是来获取资源的，但是对于CountDownLatch不需要了，因为他只需要一开始设定state即可。不过CountDownLatch为什么是共享锁我理解了，他允许多个线程同时使用CPU资源。

Node.waitStatus

-CANCELLED：值为1，在同步队列中等待的线程等待超时或被中断，需要从同步队列中取消该Node的结点，其结点的waitStatus为CANCELLED，即结束状态，进入该状态后的结点将不会再变化。
-SIGNAL：值为-1，被标识为该等待唤醒状态的后继结点，当其前继结点的线程释放了同步锁或被取消，将会通知该后继结点的线程执行。说白了，就是处于唤醒状态，只要前继结点释放锁，就会通知标识为SIGNAL状态的后继结点的线程执行。
-CONDITION：值为-2，与Condition相关，该标识的结点处于等待队列中，结点的线程等待在Condition上，当其他线程调用了Condition的signal()方法后，CONDITION状态的结点将从等待队列转移到同步队列中，等待获取同步锁。
-PROPAGATE：值为-3，与共享模式相关，在共享模式中，该状态标识结点的线程处于可运行状态。
0状态：值为0，代表初始化状态。

AQS在判断状态时，通过用waitStatus>0表示取消状态，而waitStatus<0表示有效状态。

CountDownLatch的-1方法

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

AQS中

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

前面讲到：releaseShared就是让state-1，当你减到0的时候，减到0的那个线程开始唤醒线程。

这里和代码逻辑对应一下。

tryReleaseShared就是让state-1。如果减到0，就返回true了，开始执行doReleaseShared，唤醒线程。

private void doReleaseShared() {
    for (;;) {
        Node h = head;//头结点
        if (h != null && h != tail) {//如果不为空，且不是尾节点
            int ws = h.waitStatus;
            if (ws == Node.SIGNAL) {//如果是待唤醒，waitStatus改为初始化状态
                if (!h.compareAndSetWaitStatus(Node.SIGNAL, 0))
                    continue;            // loop to recheck cases
                unparkSuccessor(h);//找到他后面的第一个节点，也就是允许使用cpu，同时解锁了
            }
            else if (ws == 0 &&//如果是初始化状态 希望把他变成可运行状态
                     !h.compareAndSetWaitStatus(0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        if (h == head)//如果head变了，就继续loop                   // loop if head changed
            break;
    }
}

所以doReleaseShared的整体意思是：唤醒下一个线程

unparkSuccessor

private void unparkSuccessor(Node node) {
    int ws = node.waitStatus;
    if (ws < 0)
        node.compareAndSetWaitStatus(ws, 0);//如果是可运行、可唤醒，那就置为初始化状态

    Node s = node.next;
    if (s == null || s.waitStatus > 0) {
        s = null;//如果为空或者cancel
        for (Node p = tail; p != node && p != null; p = p.prev)//从后往前找
            if (p.waitStatus <= 0)//目的是为了找到一个可运行、可唤醒的
                s = p;
    }
    if (s != null)//如果找到了
        LockSupport.unpark(s.thread);//解锁这个线程
}

整体的意思就是，找到他后面的第一个节点，也就是允许使用cpu，同时解锁了

那么被唤醒的线程是在哪里唤醒下一个线程的呢？

需要从await入手

public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0)//如果state!=0，之前初始化已经让state=2了
        doAcquireSharedInterruptibly(arg);
}

private void doAcquireSharedInterruptibly(int arg)
    throws InterruptedException {
    final Node node = addWaiter(Node.SHARED);//尾部加一个node
    try {
        for (;;) {
            final Node p = node.predecessor();//前一个node
            if (p == head) {//是head
                int r = tryAcquireShared(arg);
                if (r >= 0) {//state==0
                    setHeadAndPropagate(node, r);//那就开始，并唤醒下一个线程
                    p.next = null; // help GC
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())//锁住咯，并让上一个线程置为通知状态
                throw new InterruptedException();
        }
    } catch (Throwable t) {
        cancelAcquire(node);
        throw t;
    }
}

CountDownLatch就分析的差不多了

为何CountDownLatch被称为共享锁？

因为CountDownLatch latch = new CountDownLatch(2)，他允许了多个线程同时使用CPU资源，而ReentrantLock是单行道。在这个实例化中，锁可以认为有2道（state），每个线程执行完后，通过cas方式正确地减了一道。

再讲一下CyclicBarrier、 Semaphore，不过就不扯源码了

CyclicBarrier

可以实现让一组线程等待至某个状态之后再全部同时执行

例子

续

执行结果：

总结CountDownLatch、CyclicBarrie的区别

前者是手动-1，另外线程先阻塞着。state减到0后，另外线程可以跑了

后者是若干线程先阻塞着，等到阻塞到若干数量后，所有线程就可以跑了

Semaphore

若一个工厂有5台机器，但是有8个工人，一台机器同时只能被一个工人使用，只有使用完了，其他工人才能继续使用。那么我们就可以通过Semaphore来实现：

执行结果：

其实到这里都明白了，state就代表了共享资源池，在不同的并发器里有不同的意义。Semaphore的意义是资源，我可以acquire1个，也可以acquire多个。

state在CountDownLatch中的意义是，当前资源有几个人在使用，直到0个人使用，锁才能解除；

state在CyclicBarrier中的意义是当前资源几个人已经用完了，直到x个人用完了，锁才能解除。

AQS就不继续深究了，本文叙述其共享实现就够了