多线程进阶——JUC并发编程之CountDownLatch源码一探究竟?

JDK的并发包中提供了几个非常有用的并发工具类。    **CountDownLatch**、    **CyclicBarrier**和    **Semaphore**工具类提供了一种并发流程控制的手段。本文将介绍CountDownLatch(闭锁)的实现原理。在了解闭锁之前需要先了解AQS，因为CountDownLatch的实现需要依赖于AQS共享锁的实现机制。   

// 计数器
public class CountDownLatchDemo {
    public static void main(String[] args) throws InterruptedException {
        // 总数是6，必须要执行任务的时候，再使用！
        CountDownLatch countDownLatch = new CountDownLatch(6);
        for (int i = 1; i <=6 ; i++) {
            new Thread(()->{
                System.out.println(Thread.currentThread().getName()+" Go out");
                countDownLatch.countDown(); // 数量-1
            },String.valueOf(i)).start();
        }
        countDownLatch.await(); // 等待计数器归零，然后再向下执行
        System.out.println("Close Door");
    }
}

package java.util.concurrent;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;

public class CountDownLatch {
   
    private static final class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 4982264981922014374L;
        //设置同步状态
        Sync(int count) {
            setState(count);
        }
        //获取同步状态的值
        int getCount() {
            return getState();
        }
        //获取共享锁,1、getState>1返回1：表示获取到共享锁，-1：表示没有获取到共享锁
        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;
                //通过CAS设置同步状态值，如果设置失败则说明同一时刻有其它线程在设置，但是会通过自旋的方式最终设置成功
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }
    }

    private final Sync sync;

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

public class CountDownLatch {
    /**
     * Synchronization control For CountDownLatch.
     * Uses AQS state to represent count.
     */
    private static final class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 4982264981922014374L;

        Sync(int count) {
            setState(count);
        }

        int getCount() {
            return getState();
        }

    private volatile int state;

    protected final void setState(int newState) {
        state = newState;
    }

    public void await() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }
    //AQS获取共享资源时，该方法是响应中断的
    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        //如果线程中断则抛出异常
        if (Thread.interrupted())
            throw new InterruptedException();
        //tryAcquireShared(arg)是AQS提供的模板方法
        //尝试看当前计数器值是否为0，为0则直接返回，否则进入AQS的等待队列
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }

 //尝试看当前计数器值是否为0，为0则表示获取到共享锁，阻塞的等待的线程将会被唤醒      
 protected int tryAcquireShared(int acquires) {
       return (getState() == 0) ? 1 : -1;
  }

    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;
        //只有当当前节点状态为Singal才返回true
        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;
    }

线程已经被取消
static final int CANCELLED =  1;
线程需要去被唤醒
static final int SIGNAL    = -1;
线程正在唤醒等待条件
static final int CONDITION = -2; 
//线程的共享锁应该被无条件传播
static final int PROPAGATE = -3;

    private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this);
        return Thread.interrupted();
    }
    public static void park(Object blocker) {
        Thread t = Thread.currentThread();
        setBlocker(t, blocker);
        UNSAFE.park(false, 0L);
        setBlocker(t, null);
    }

public native void park(boolean var1, long var2);

//调用countDown()释放同步状态，每次调用同步状态值-1
 public void countDown() {
       sync.releaseShared(1);
 }

public final boolean releaseShared(int arg) {
        //tryReleaseShared方法必须保证同步状态线程安全释放，一般是通过CAS和循环来实现
        if (tryReleaseShared(arg)) {
           //唤醒后续处于等待的节点
            doReleaseShared();
            return true;
        }
        return false;
    }

protected boolean tryReleaseShared(int releases) {
            // Decrement count; signal when transition to zero
            for (;;) {//自旋锁
                int c = getState();
                //在并发情况下，可能已经被其他线程修改为0，则直接返回false
                if (c == 0)
                    return false;
                int nextc = c-1;
                if (compareAndSetState(c, nextc))//CAS自旋减一
                    return nextc == 0;//next为0返回true
            }
        }
    }

//释放共享锁，通知后续节点，主要是唤醒调用了await方法的线程（一般为主线程）
private void doReleaseShared() {
        for (;;) {
            Node h = head;//获取头节点
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                //SIGNAL为-1，后继节点的线程处于等待状态，当前节点的线程如果释放了同步状态或者被取消，将会通知后继节点
                if (ws == Node.SIGNAL) { //头结点的状态为Node.SIGNAL
                    //将头结点的状态值设置为0
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;            // loop to recheck cases
                    unparkSuccessor(h);//这里唤醒后继节点
                }
                //WaitStatus为0的时候表示为初始状态，设置当前节点为-3，表示线程的共享锁应该被无条件传播
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
           //如果h还是指向头结点，说明没有其他节点对头结点进行修改
            if (h == head)                   // loop if head changed
                break;
        }
    }

private void unparkSuccessor(Node node) {
       
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);
        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);//唤醒线程（解锁的过程）
    }

private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head; // Record old head for check below
        setHead(node);//设置Head为头结点
    
        //如果走到了setHeadAndPropagate方法，那么propagate的值一定大于1，if条件成立
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) {
            //获取当前节点的下一个节点
            Node s = node.next;
            if (s == null || s.isShared())
                doReleaseShared();//进入这里
        }
    }

private void doReleaseShared() {
        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 cases
                    unparkSuccessor(h);//唤醒后续等待被唤醒的线程
                }
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
           //明白这里为什么要加一次判断了吧！！！,被唤醒的线程会在执行该函数
            if (h == head)                   // loop if head changed
                break;
        }
    }

    private void unparkSuccessor(Node node) {
      
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);
        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);
    }