CountDownLatch原理浅析

public class CountDownLatchDemo {
    // 避免指令重排序
    private volatile  ThreadPoolExecutor threadPoolExecutor;
    public void countDownLatchDemo(){
        // 一共五十个任务
        CountDownLatch countDownLatch = new CountDownLatch(50);
        for (int i = 0; i < 50; i++) {
            getThreadPoolExecutor().execute(()->{
                // 线程做完对应的任务,任务数减一
                countDownLatch.countDown();
            });
        }
        try {
            // main 线程走到这里,如果50个任务没完成,就会被阻塞在这这里。
            // hello world 不会输出
            countDownLatch.await();
            System.out.println("hello world");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
      // 接着做对应的业务处理
    }
    private ThreadPoolExecutor getThreadPoolExecutor() {
        // 获取当前系统的逻辑核心数,我的是八核十六线程。
        int cpuNumbers = Runtime.getRuntime().availableProcessors();
        if (Objects.isNull(threadPoolExecutor)) {
            synchronized (CountDownLatchDemo.class) {
                // 为线程池起名,便于定位问题
                // CustomizableThreadFactory 来自于Spring
                ThreadFactory threadFactory = new CustomizableThreadFactory("fetch-data-pool-");
                threadPoolExecutor = new ThreadPoolExecutor(cpuNumbers, cpuNumbers, 1L, TimeUnit.HOURS, new LinkedBlockingDeque<>(), threadFactory);
                return threadPoolExecutor;
            }
        }
        return threadPoolExecutor;
    }
    public static void main(String[] args) {
        CountDownLatchDemo countDownLatchDemo = new CountDownLatchDemo();
        countDownLatchDemo.countDownLatchDemo();
    }
}

// 这个sync事实上是CountDownLatch中静态内部类Sync的实例
public void countDown() {
   //releaseShared调用的是父类AbstractQueuedSynchronizer的releaseShared方法
   sync.releaseShared(1);
 }
// tryReleaseShared为CountDownLatch中静态内部类Sync重写父类的方法
public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
 }
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
            // 减少count
            for (;;) {
                // 我们看getState和setState方法,getState和setState是父类的方法
                // 获取state
                int c = getState();
      // 如果等于0 返回false
                if (c == 0)
                    return false;
                // 否则就将state 变量 减一
                int nextc = c - 1;     
                // compareAndSetState调用的是AbstractQueuedSynchronizer的方法
                // 默认调用的是CAS,给AQS的state变量设置值。
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }
//下面是AbstractQueuedSynchronizer的compareAndSetState
// 这里的stateOffset用可以理解为字段state的偏移地址,通过this+stateOffset可以定位到state变量的内存地址
 protected final boolean compareAndSetState(int expect, int update) {
        // See below for intrinsics setup to support this
        return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
 }

/**
* 为当前线程创建一个排队结点,并且给一个模式
* Creates and enqueues node for current thread and given mode.
* mode 有两种类型, 一种是独占模式,一种是共享模式。有没有想到独占锁和共享锁呢
* @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;
     // 这个操作相当于设置尾结点
       // 首次添加为尾结点为null
        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 (;;) {
            // 获得tail的应用
            Node t = tail;
            // 如果tail 为空
            if (t == null) { // Must initialize
                // 设置头结点,此时头结点又是尾结点
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
                // 如果尾结点非空代表此时队列非空,尾结点成为该结点的前驱结点
                node.prev = t;
                // 将当前结点设置尾结点
                if (compareAndSetTail(t, node)) {
                    // 将尾结点的后继结点指向插入结点
                    t.next = node;
                    return t;
                }
            }
        }
 }

private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
     // 创建一个结点,并返回该结点
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            for (;;) {
                // 获取前驱结点
                final Node p = node.predecessor();
                // 如果是头结点
                if (p == head) {
                    // 获取state状态
                    // 大于零任务完成
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {                     
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        failed = false;
                        return;
                    }
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
}

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
       // 获取前驱结点的状态
     int ws = pred.waitStatus;
     // 前继结点处于唤醒,当前结点可以被安全的被阻塞
     if (ws == Node.SIGNAL)
            return true;
      // 通过看注释可以看明白 ws > 0，说明当前结点pred处于取消状态。
     // 将该结点pred从队列里面移除,接着向前遍历
        if (ws > 0) {         
            do {
               node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            // 将pred,也就是当前结点的前一个结点设置为-1.
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
 }

private final boolean parkAndCheckInterrupt() {
      // 用的是UnSafe,所以这里我们暂时不用关心它的实现原理，只需要知道线程调用这个方法
     // 相当于请求操作系统阻塞线程
        LockSupport.park(this);
     // 如果当前线程已经被中断,且满足shouldParkAfterFailedAcquire
     // 抛出异常。中断了就没有被阻塞这个概念。
        return Thread.interrupted();
}

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