private static void countDownLatch() {
    Thread[] threads = new Thread[10];
    CountDownLatch latch = new CountDownLatch(threads.length);

    for(int i=0; i<threads.length; i++) {
        threads[i] = new Thread(()->{
            int result = 0;
            for(int j=0; j<10; j++) result += j;
            // 线程执行完毕result 之后，计数器 -1 如果-1 后latch 值还是 > 0 则阻塞
            latch.countDown();
            System.out.println(Thread.currentThread().getName() + ":finsh result:" + result);
        });
        threads[i].start();
    }

    try {
        // CountDownLatch 保证 只有在所有的线程执行完毕之后，才能执行 也就是 CountDownLatch  的值为 0
        latch.await();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    
    System.out.println("end latch");
}

private static final class Sync extends AbstractQueuedSynchronizer {
       // 当 status 不为0 则返回失败，根据AQS 的特性，一旦获取失败，则进入等待队列。
       // 具体的 要参考 AQS 的实现
       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;
           }
       }
   }

   private final Sync sync;

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

public void await() throws InterruptedException {
    // 获取共享锁
    sync.acquireSharedInterruptibly(1);
}

public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
        // 依赖于 获取共享锁，如果是 返回获取失败，则表示 需要加锁，在构造函数的时候，将status 设置成了一个具体的值，
        // 所以在调用 await 的时候，线程阻塞
    if (tryAcquireShared(arg) < 0)
        // 这里加入的都是共享锁的Node
        doAcquireSharedInterruptibly(arg);
}

public void countDown() {
    sync.releaseShared(1);
}
// AQS
public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        // 可以看到，这里如果 tryReleaseShared 释放成功，以为这status == 0 
        // 释放所有的共享Node
        doReleaseShared();
        return true;
    }
    return false;
}

private  static void testTestCyclicBarrier() {
    Thread[] threads = new Thread[5];
    
    CyclicBarrier barrier = new CyclicBarrier(threads.length, () -> {
        System.out.println("============ wake up all ============");
    });


    for (int i = 0; i < threads.length; i++) {
        threads[i] = new Thread(()->{
            String name = Thread.currentThread().getName();
            long l = System.currentTimeMillis();
            try {
                System.out.println(name + ": await");
                barrier.await();
            } catch (InterruptedException | BrokenBarrierException e) {
                e.printStackTrace();
            }
            System.out.println(name + ": wake up wait:" + (System.currentTimeMillis() - l));
        });
        threads[i].start();
    }
}

public class CyclicBarrier {
    /**
     * 静态内部类，当前屏障是否被破坏
     */
    private static class Generation {
        boolean broken = false;
    }

    /** 实现的Lock */
    private final ReentrantLock lock = new ReentrantLock();
    /** Condition用来实现wait */
    private final Condition trip = lock.newCondition();
    /** 等待的屏障数 */
    private final int parties;
    /* 到达屏障要执行的Runnable */
    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);
    }
}

private int dowait(boolean timed, long nanos)
    throws InterruptedException, BrokenBarrierException,
           TimeoutException {
    /**获取CyclicBaerrier的内部锁*/
    final ReentrantLock lock = this.lock;
    /**获取锁*/
    lock.lock();
    try {
        /**存储当前的Generation*/
        final Generation g = generation;
        /**判断当前的屏障是否被破坏，如果破坏则抛出BrokenBarrierException异常*/
        if (g.broken)
            throw new BrokenBarrierException();
        /**判断当前线程是否被interrupted，如果被打断，则breakBarrier破坏屏障*/
        if (Thread.interrupted()) {
            breakBarrier();
            throw new InterruptedException();
        }
       /**记录当前屏障等待个数*/ 
       int index = --count;
       if (index == 0) {  // 最后一个预留到达屏障的线程
           boolean ranAction = false;
           try {
               final Runnable command = barrierCommand;
               /**执行barrierCommand指令*/
               if (command != null)
                   command.run();
               ranAction = true;
               /**执行下一个Generation*/
               nextGeneration();
               return 0;
           } finally {
               /**如果barrierCommand执行失败，进行屏障破坏处理*/
               if (!ranAction)
                   breakBarrier();
           }
       }

        // 如果当前线程不是最后一个到达的线程
        for (;;) {
            try {
                if (!timed)///调用Condition的await()方法阻塞
                    trip.await();
                else if (nanos > 0L)///调用Condition的awaitNanos()方法阻塞
                    nanos = trip.awaitNanos(nanos);
            } catch (InterruptedException ie) {
                /**如果当前线程被中断，则判断是否有其他线程已经使屏障破坏。若没有则进行屏障破坏处理，并抛出异常；否则再次中断当前线程*/
                if (g == generation && ! g.broken) {
                    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)
                return index;

            if (timed && nanos <= 0L) {
                breakBarrier();
                throw new TimeoutException();
            }
        }
    } finally {
        lock.unlock();
    }
}

private void nextGeneration() {
    // signal completion of last generation
    trip.signalAll();
    // set up next generation
    count = parties;
    generation = new Generation();
}

private  static void testSemaphore () throws InterruptedException {
    int solt = 10;
    Map<Integer, List<String>> washRoom= new HashMap<>(solt);

    for (int i = 0; i < solt; i++) {
        washRoom.put(i, new ArrayList<String>());
    }
    // 是有10个槽位，所以每次是能有10个线程执行
    Semaphore semaphore = new Semaphore(solt);
    Runnable r  = () -> {

        try {
            // 如果没有获取到坑位，则阻塞
            semaphore.acquire();
            int name = Integer.parseInt(Thread.currentThread().getName());
            int index = name % 10;
            List<String> list = washRoom.get(index);
            list.add(name + "");
            TimeUnit.MILLISECONDS.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            semaphore.release();
        }
    };
    Thread[] threads = new Thread[solt * 3];
    for (int i = 0; i < threads.length; i++) {
        Thread thread = new Thread(r);
        thread.setName(i + "");
        threads[i] =  thread;
        thread.start();
    }

    for (Thread t : threads) {
        t.join();
    }
    System.out.println(washRoom);
}

public Semaphore(int permits) {
    sync = new NonfairSync(permits);
}

public Semaphore(int permits, boolean fair) {
    sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

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

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

public static void PhaserTestExample1() throws InterruptedException {
      int solt = 3;
    Phaser phaser = new Phaser();
    Thread[] threads = new Thread[solt];
    Runnable r = () ->{
        System.out.println(Thread.currentThread().getName() + ": 执行开始任务:" + phaser.getPhase());
        int i = phaser.arriveAndAwaitAdvance();
        System.out.println(Thread.currentThread().getName() + ": 执行完任务，当前phase:" + i + "=" + phaser.getPhase());
    };
    // 注册solt个
    for (int i = 0; i < solt; i++) {
        phaser.register();
    }
    // 或者使用
    //phaser.bulkRegister(solt);

    for (int i = 0; i < solt; i++) {
        threads[i] = new Thread(r);
        threads[i].setName("PhaserTestExample1-" + i);
        threads[i].start();
    }

    for (Thread t : threads) {
        t.join();
    }
    int registeredParties = phaser.getRegisteredParties();
    int arrivedParties = phaser.getArrivedParties();
    System.out.println("registeredParties:" + registeredParties);
    System.out.println("arrivedParties:" + arrivedParties);
}

public static void PhaserTestExample2() throws InterruptedException, IOException {
    int solt = 3;
    //注册主线程,当外部条件满足时,由主线程打开开关
    Phaser phaser = new Phaser(1);
    Thread[] threads = new Thread[solt];
    Runnable r = () ->{
        System.out.println(Thread.currentThread().getName() + ":任务准备");
        int i = phaser.arriveAndAwaitAdvance();
        System.out.println(Thread.currentThread().getName() + ": 执行完任务，当前phase =" + i + "");
    };

    phaser.bulkRegister(solt);

    for (int i = 0; i < solt; i++) {
        threads[i] = new Thread(r);
        threads[i].setName("PhaserTestExample2-" + i);
        threads[i].start();
    }

    // 外部条件:等待用户输入命令
    System.out.println("Press ENTER to continue");
    BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
    reader.readLine();
    // 放开阀门 主线程打开了开关
    phaser.arriveAndDeregister();

    int registeredParties = phaser.getRegisteredParties();
    int arrivedParties = phaser.getArrivedParties();
    System.out.println("registeredParties:" + registeredParties);
    System.out.println("arrivedParties:" + arrivedParties);
}

public static void PhaserTestExample3() throws Exception {
    int solt = 3;    // 指定任务最多执行的次数
    Thread[] threads = new Thread[solt];
    Phaser phaser = new Phaser() {
        @Override
        protected boolean onAdvance(int phase, int registeredParties) {
            System.out.println("---------------PHASE[" + phase + "],Parties[" + registeredParties + "] ---------------");
            //如果返回 True 则Phaser结束
            return phase + 1 >= solt  || registeredParties == 0;
        }
    };

    phaser.bulkRegister(solt);
    Runnable r = () ->{
        while (!phaser.isTerminated()) { //只要Phaser没有终止, 各个线程的任务就会一直执行
            int i = phaser.arriveAndAwaitAdvance(); // 等待其它参与者线程到达
            System.out.println(Thread.currentThread().getName() + ": 执行完任务");
        }
    };

    for (int i = 0; i < solt; i++) {
        threads[i] = new Thread(r);
        threads[i].setName("PhaserTestExample3-" + i);
        threads[i].start();
    }
}

---------------PHASE[0],Parties[3] ---------------
PhaserTestExample3-2: 执行完任务
PhaserTestExample3-1: 执行完任务
PhaserTestExample3-0: 执行完任务
---------------PHASE[1],Parties[3] ---------------
PhaserTestExample3-0: 执行完任务
PhaserTestExample3-2: 执行完任务
PhaserTestExample3-1: 执行完任务
---------------PHASE[2],Parties[3] ---------------
PhaserTestExample3-1: 执行完任务
PhaserTestExample3-2: 执行完任务
PhaserTestExample3-0: 执行完任务

public class PhaserTest4 {
    private static final int TASKS_PER_PHASER = 4;      // 每个Phaser对象对应的工作线程（任务）数

    public static void main(String[] args) throws IOException {

        int repeats = 3;    // 指定任务最多执行的次数
        Phaser phaser = new Phaser() {
            @Override
            protected boolean onAdvance(int phase, int registeredParties) {
                System.out.println("---------------PHASE[" + phase + "],Parties[" + registeredParties + "] ---------------");
                return phase + 1 >= repeats || registeredParties == 0;
            }
        };

        Tasker[] taskers = new Tasker[10];
        build(taskers, 0, taskers.length, phaser);       // 根据任务数,为每个任务分配Phaser对象

        for (int i = 0; i < taskers.length; i++) {          // 执行任务
            Thread thread = new Thread(taskers[i]);
            thread.start();
        }
    }

    private static void build(Tasker[] taskers, int lo, int hi, Phaser phaser) {
        if (hi - lo > TASKS_PER_PHASER) {
            for (int i = lo; i < hi; i += TASKS_PER_PHASER) {
                int j = Math.min(i + TASKS_PER_PHASER, hi);
                build(taskers, i, j, new Phaser(phaser));
            }
        } else {
            for (int i = lo; i < hi; ++i)
                taskers[i] = new Tasker(i, phaser);
        }

    }

    static class Tasker implements Runnable {
        private final Phaser phaser;
        private int i;
        Tasker(int i, Phaser phaser) {
            this.i = i;
            this.phaser = phaser;
            this.phaser.register();
        }

        @Override
        public void run() {
            while (!phaser.isTerminated()) {   //只要Phaser没有终止, 各个线程的任务就会一直执行
                int i = phaser.arriveAndAwaitAdvance();     // 等待其它参与者线程到达
                // do something
                System.out.println(Thread.currentThread().getName() + ": 执行完任务, index:" + this.i);
            }
        }
    }
}

public static void exchangerTestExample() throws Exception {
    final Exchanger<String> exchanger = new Exchanger<String>();
    Runnable r = () ->{
        try {
            String myName = Thread.currentThread().getName();
            Random random = new Random();
            TimeUnit.SECONDS.sleep(random.nextInt(5));
            System.out.println(myName + " want to exchange data and i waiting");
            String hisName = exchanger.exchange(Thread.currentThread().getName());
            System.out.println(myName + ": " + hisName);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    };
    Thread[] threads = new Thread[2];
    for (int i = 0; i < 2; i++) {
        threads[i] = new Thread(r);
        threads[i].setName("exchangerTestExample-" + i);
        threads[i].start();
    }
}

static final class Participant extends ThreadLocal<Node> {
    public Node initialValue() { return new Node(); }
}
// Node 结构如下
@sun.misc.Contended static final class Node {
    int index;                  // arena的下标，多个槽位的时候利用
    int bound;                  // 上一次记录的Exchanger.bound
    int collides;               // 在当前bound下CAS失败的次数；
    int hash;                   // 用于自旋；
    Object item;                // 这个线程的当前项，也就是需要交换的数据；
    volatile Object match;      //做releasing操作的线程传递的项；
    volatile Thread parked;     //挂起时设置线程值，其他情况下为null；
}

public V exchange(V x) throws InterruptedException {
    Object v;
    // NULL_ITEM 是一个替代的对象，这里可以简单理解为默认值
    // 主要是因为 slotExchange 和  arenaExchange 并没有做非空的判断
    Object item = (x == null) ? NULL_ITEM : x; // translate null args
    if ((arena != null ||
        (v = slotExchange(item, false, 0L)) == null) &&
        ((Thread.interrupted() || // disambiguates null return
        (v = arenaExchange(item, false, 0L)) == null)))
        throw new InterruptedException();
    return (v == NULL_ITEM) ? null : (V)v;
}
// 
/**
 * Elimination array; null until enabled (within slotExchange).
 * Element accesses use emulation of volatile gets and CAS.
 */
private volatile Node[] arena;

private final Participant participant;
private volatile Node[] arena;
private volatile Node slot;

private final Object slotExchange(Object item, boolean timed, long ns) {
    // 获取当前线程node对象
    Node p = participant.get();
    // 当前线程
    Thread t = Thread.currentThread();  
    // 若果线程被中断，就直接返回null
    if (t.isInterrupted())
        return null;
    // 自旋
    for (Node q;;) {
        // 将slot值赋给q
        if ((q = slot) != null) {
            // slot 不为null，即表示已有线程已经把需要交换的数据设置在slot中了
            // 通过CAS将slot设置成null
            if (U.compareAndSwapObject(this, SLOT, q, null)) {
                Object v = q.item;
                q.match = item;
                Thread w = q.parked;
                if (w != null)
                    U.unpark(w);
                return v;
            }
            // create arena on contention, but continue until slot null
            if (NCPU > 1 && bound == 0 &&
                U.compareAndSwapInt(this, BOUND, 0, SEQ))
                arena = new Node[(FULL + 2) << ASHIFT];
        }
        else if (arena != null)
            return null; // caller must reroute to arenaExchange
        else {
            // 这里表示当前线程是以第一个线程进来交换数据
            // 或者表示之前的数据交换已进行完毕，这里可以看作是第一个线程
            // 将需要交换的数据先存放在当前线程变量p中 
            // 这里将 第一个想成的Node 放到了 solt 当中
            p.item = item;
            if (U.compareAndSwapObject(this, SLOT, null, p))
                break;
            
            // CAS操作失败，表示有其它线程刚好先于当前线程将数据设置到交换区slot
            // 将当前线程变量中的item设置为null，然后自旋获取其它线程存放在交换区slot的数据
            p.item = null;
        }
    }

    // await release
    // 执行到这里表示当前线程已将需要的交换的数据放置于交换区slot中了，
    // 等待其它线程交换数据然后唤醒当前线程
    int h = p.hash;
    long end = timed ? System.nanoTime() + ns : 0L;
    // 自旋次数
    int spins = (NCPU > 1) ? SPINS : 1;
    Object v;
    // 自旋等待直到p.match不为null，也就是说等待其它线程将需要交换的数据放置于交换区slot
    // 如果另一个线程 将p.match 设置上，这里就可以跳出自选了
    while ((v = p.match) == null) {
         // 下面的逻辑主要是自旋等待，直到spins递减到0为止
        if (spins > 0) {
            // h 表示一个随机数 这样计算似的 h 可以在正负当中来回摆动 概率是1：1
            h ^= h << 1; h ^= h >>> 3; h ^= h << 10;
            if (h == 0)
                h = SPINS | (int)t.getId();
            else if (h < 0 && (--spins & ((SPINS >>> 1) - 1)) == 0)
                // 让出cpu
                Thread.yield();
        }
        // 如果slot ！= p 证明已经有一个线程准备搞好了，这时候需要回到 while 判断match 是否存在
        else if (slot != p)
            spins = SPINS;
        // 此处表示未设置超时或者时间未超时 需要将本线程阻塞
        else if (!t.isInterrupted() && arena == null &&
                (!timed || (ns = end - System.nanoTime()) > 0L)) {
            U.putObject(t, BLOCKER, this);
            // 给p挂机线程的值赋值
            p.parked = t;
            if (slot == p)
                // 如果slot还没有被置为null，也就表示暂未有线程过来交换数据，需要将当前线程挂起
                U.park(false, ns);
            p.parked = null;
            U.putObject(t, BLOCKER, null);
        }
        //arena不为null则v为null,其它为超时则v为超时对象TIMED_OUT，并且跳出循环
        else if (U.compareAndSwapObject(this, SLOT, p, null)) {
            v = timed && ns <= 0L && !t.isInterrupted() ? TIMED_OUT : null;
            break;
        }
    }
    U.putOrderedObject(p, MATCH, null);
    p.item = null;
    p.hash = h;
    return v;
}

for (Node q;;) {
    // 将slot值赋给q
    if ((q = slot) != null) {
        // slot 不为null，即表示已有线程已经把需要交换的数据设置在slot中了
        // 通过CAS将slot设置成null
        if (U.compareAndSwapObject(this, SLOT, q, null)) {
            Object v = q.item;
            q.match = item;
            Thread w = q.parked;
            if (w != null)
                U.unpark(w);
            return v;
        }
        // create arena on contention, but continue until slot null
        if (NCPU > 1 && bound == 0 &&
            U.compareAndSwapInt(this, BOUND, 0, SEQ))
            arena = new Node[(FULL + 2) << ASHIFT];
    }
    // ...
}

// timed 为true表示设置了超时时间，ns为>0的值，反之没有设置超时时间
 private final Object arenaExchange(Object item, boolean timed, long ns) {
        Node[] a = arena;
        // 获取当前线程中的存放的node
        Node p = participant.get();
        for (int i = p.index;;) {                      // access slot at i
            int b, m, c; long j;                       // j is raw array offset
            Node q = (Node)U.getObjectVolatile(a, j = (i << ASHIFT) + ABASE);
            if (q != null && U.compareAndSwapObject(a, j, q, null)) {
                Object v = q.item;                     // release
                q.match = item;
                Thread w = q.parked;
                if (w != null)
                    U.unpark(w);
                return v;
            }
            else if (i <= (m = (b = bound) & MMASK) && q == null) {
                p.item = item;                         // offer
                if (U.compareAndSwapObject(a, j, null, p)) {
                    long end = (timed && m == 0) ? System.nanoTime() + ns : 0L;
                    Thread t = Thread.currentThread(); // wait
                    for (int h = p.hash, spins = SPINS;;) {
                        Object v = p.match;
                        if (v != null) {
                            U.putOrderedObject(p, MATCH, null);
                            p.item = null;             // clear for next use
                            p.hash = h;
                            return v;
                        }
                        else if (spins > 0) {
                            h ^= h << 1; h ^= h >>> 3; h ^= h << 10; // xorshift
                            if (h == 0)                // initialize hash
                                h = SPINS | (int)t.getId();
                            else if (h < 0 &&          // approx 50% true
                                     (--spins & ((SPINS >>> 1) - 1)) == 0)
                                Thread.yield();        // two yields per wait
                        }
                        else if (U.getObjectVolatile(a, j) != p)
                            spins = SPINS;       // releaser hasn't set match yet
                        else if (!t.isInterrupted() && m == 0 &&
                                 (!timed ||
                                  (ns = end - System.nanoTime()) > 0L)) {
                            U.putObject(t, BLOCKER, this); // emulate LockSupport
                            p.parked = t;              // minimize window
                            if (U.getObjectVolatile(a, j) == p)
                                U.park(false, ns);
                            p.parked = null;
                            U.putObject(t, BLOCKER, null);
                        }
                        else if (U.getObjectVolatile(a, j) == p &&
                                 U.compareAndSwapObject(a, j, p, null)) {
                            if (m != 0)                // try to shrink
                                U.compareAndSwapInt(this, BOUND, b, b + SEQ - 1);
                            p.item = null;
                            p.hash = h;
                            i = p.index >>>= 1;        // descend
                            if (Thread.interrupted())
                                return null;
                            if (timed && m == 0 && ns <= 0L)
                                return TIMED_OUT;
                            break;                     // expired; restart
                        }
                    }
                }
                else
                    p.item = null;                     // clear offer
            }
            else {
                if (p.bound != b) {                    // stale; reset
                    p.bound = b;
                    p.collides = 0;
                    i = (i != m || m == 0) ? m : m - 1;
                }
                else if ((c = p.collides) < m || m == FULL ||
                         !U.compareAndSwapInt(this, BOUND, b, b + SEQ + 1)) {
                    p.collides = c + 1;
                    i = (i == 0) ? m : i - 1;          // cyclically traverse
                }
                else
                    i = m + 1;                         // grow
                p.index = i;
            }
        }
    }