package com.conrrentcy.juc;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class CountDownLatchExample {
    private static final Logger log = LoggerFactory.getLogger(CountDownLatchExample.class);
     //線程數(shù)量
    private static final int THREAD_NUM = 10;
    // CountdownLatch阻塞模擬
    public static void main(String[] args) throws InterruptedException {
        // 創(chuàng)建線程池 用于執(zhí)行線程
        ExecutorService executorService = Executors.newCachedThreadPool();
        //創(chuàng)建countDownLatch
        final CountDownLatch countDownLatch = new CountDownLatch(THREAD_NUM);
        long startTime = System.currentTimeMillis();
        //循環(huán)創(chuàng)建線程
        for (int i = 0; i < THREAD_NUM; i++) {
            final int a = i;
            executorService.execute(() -> {
                try {
                    test(a);
                } catch (Exception e) {
                    log.error("Exception", e);
                } finally {
                    countDownLatch.countDown();
                }
            });
        }
        countDownLatch.await();
        long endTime = System.currentTimeMillis();
        log.info("執(zhí)行完畢,{}-{}",startTime,endTime);
        executorService.shutdown();
    }
    private static void test(int num) throws InterruptedException {
        Thread.sleep(100);
        log.info("{}-{}", num,System.currentTimeMillis());
        Thread.sleep(100);
    }
}

//當(dāng)前對象中私有阻塞工具
private final Sync sync;
// 模板方法模式重寫AQS工具
private static final class Sync extends AbstractQueuedSynchronizer {
   private static final long serialVersionUID = 4982264981922014374L;
   // 共享阻塞AQS
   Sync(int count) {
       setState(count);
   }
   // 獲取當(dāng)前還剩多少資源可以使用
   int getCount() {
       return getState();
   }
   protected int tryAcquireShared(int acquires) {
       return (getState() == 0) ? 1 : -1;
   }
   protected boolean tryReleaseShared(int releases) {
       for (;;) {
           int c = getState();
           if (c == 0)
               return false;
           int nextc = c-1;
           if (compareAndSetState(c, nextc))
               return nextc == 0;
       }
   }
}
//構(gòu)造方法創(chuàng)建一個鎖對象
public CountDownLatch(int count) {
   if (count < 0) throw new IllegalArgumentException("count < 0");
   this.sync = new Sync(count);
}

// 該方法時釋放一個共享鎖。當(dāng)所有鎖都被釋放完成后主線程就能繼續(xù)執(zhí)行了。
public void countDown() {
    sync.releaseShared(1);
}

//攔截主線程的方法。主線程在這里等待條件達(dá)成后繼續(xù)執(zhí)行。
public void await() throws InterruptedException {
    //在這里阻塞線程的執(zhí)行
    sync.acquireSharedInterruptibly(1);
}
public final void acquireSharedInterruptibly(int arg)
       throws InterruptedException {
   if (Thread.interrupted())
       throw new InterruptedException();
    //這里判斷是否還有可以共享的資源
   // 如果有則返回-1 否則返回 1，重寫AQS的方法參見（1.AQS框架以及構(gòu)造方法）
   if (tryAcquireShared(arg) < 0)
       // 有資源則運(yùn)行阻塞自旋等待所有線程執(zhí)行完畢
       doAcquireSharedInterruptibly(arg);
   // 無資源可用就讓線程繼續(xù)執(zhí)行
}
// 帶延遲的減少數(shù)據(jù)攔截方法
// 返回的結(jié)果是沒有跑完全部線程就繼續(xù)執(zhí)行下一步了。
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();
    // 表示如果線程被執(zhí)行完了直接返回成功，如果沒有執(zhí)行完則看等待時間來決定是否要繼續(xù)執(zhí)行。
    return tryAcquireShared(arg) >= 0 ||
        doAcquireSharedNanos(arg, nanosTimeout);
}

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