package com.fuzhulei;
import java.util.concurrent.*;
​
/**
 * 減法計(jì)數(shù)器，主要是countDown（計(jì)數(shù)器1） 和 await（阻塞）方法，只有當(dāng)計(jì)數(shù)器減為0的時(shí)候，當(dāng)前線程才可以往下繼續(xù)執(zhí)行。
 * 主要用于允許一個(gè)或多個(gè)線程等待其他線程完成操作
 * @author Huxudong
 * @createTime 2020-04-05 00:04:36
 **/
public class CountDownDemo {
  public static void main(String[] args) throws InterruptedException {
    /** 使用其構(gòu)造函數(shù)，創(chuàng)建一個(gè)數(shù)值為6的計(jì)數(shù)器 */
    CountDownLatch countDownLatch = new CountDownLatch(6);
    /** 自定義線程池使用 */
    ExecutorService pool = new ThreadPoolExecutor(
        6,  // 核心線程池大小
        9, // 最大線程池的大小（根據(jù)是IO密集型，還是CPU密集型來(lái)確定大?。?
        3L,  // 超時(shí)等待時(shí)間
        TimeUnit.SECONDS,  // 時(shí)間的單位
        new LinkedBlockingQueue<>(5), // 阻塞隊(duì)列是哪一種
        Executors.defaultThreadFactory(),   // 默認(rèn)線程創(chuàng)建工廠
        new ThreadPoolExecutor.AbortPolicy()  // 四大拒絕策略，選擇一種
    );
    try{
      for (int i = 0; i < 6; i++) {
        /** 這個(gè)線程的提交，沒有返回值的任務(wù) */
        pool.execute(()->{
          countDownLatch.countDown();
          System.out.println(Thread.currentThread().getName()+"執(zhí)行一次減法");
        });
​
      }
    } catch(Exception e) {
      e.printStackTrace();
    } finally {
      /** 關(guān)閉線程池 */
      pool.shutdown();
    }
​
    countDownLatch.await();
    System.out.println("執(zhí)行完成了");
​
  }
}

package com.fuzhulei;
​
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
​
/**
 * CyclicBarrier是一個(gè)加法計(jì)數(shù)器，即同步屏障，可循環(huán)的屏障，讓一組線程到達(dá)一個(gè)屏障（也可以叫做同步點(diǎn)）時(shí)被阻塞，直到最后一個(gè)線程到達(dá)屏障，達(dá)到了一開始初始化的屏障的數(shù)值，
 * 屏障才可以打開門，所有被攔截的線程才可以繼續(xù)工作，主要是通過調(diào)用await方法來(lái)實(shí)現(xiàn)的
 * @author Huxudong
 * @createTime 2020-04-04 22:53:50
 **/
public class CyclicBarrierDemo {
  public static void main(String[] args) {
    CyclicBarrier cyclicBarrier = new CyclicBarrier(3);
    new Thread(()->{
​
      try {
        cyclicBarrier.await();
      } catch (InterruptedException e) {
        e.printStackTrace();
      } catch (BrokenBarrierException e) {
        e.printStackTrace();
      }
      System.out.println("線程A已經(jīng)到達(dá)屏障");
    },"A").start();
​
    new Thread(()->{
      try {
        cyclicBarrier.await();
      } catch (InterruptedException e) {
        e.printStackTrace();
      } catch (BrokenBarrierException e) {
        e.printStackTrace();
      }
      System.out.println("線程B已經(jīng)到達(dá)屏障");
    },"B").start();
​
    new Thread(()->{
      try {
        cyclicBarrier.await();
      } catch (InterruptedException e) {
        e.printStackTrace();
      } catch (BrokenBarrierException e) {
        e.printStackTrace();
      }
      System.out.println("線程C已經(jīng)到達(dá)屏障");
    },"C").start();
​
​
  }
}

package com.fuzhulei;
​
import java.util.concurrent.*;
​
/**
 * 用來(lái)控制同時(shí)訪問特定資源的線程數(shù)量，通過協(xié)調(diào)各個(gè)線程，以保證合理的使用公用的資源
 * @author Huxudong
 * @createTime 2020-04-04 23:45:29
 **/
public class SemaphoreDemo {
  public static void main(String[] args) {
    Semaphore semaphore = new Semaphore(5);
    ExecutorService pool = new ThreadPoolExecutor(
        10,
        20,
        3L,
        TimeUnit.SECONDS,
        new LinkedBlockingQueue<>(20),
        Executors.defaultThreadFactory(),
        new ThreadPoolExecutor.AbortPolicy());
​
    try{
      for (int i = 0; i < 60; i++) {
        pool.execute(() ->{
          try {
            semaphore.acquire();
            System.out.println(Thread.currentThread().getName()+"限流成功");
            semaphore.release();
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
        });
      }
    } catch(Exception e) {
      e.printStackTrace();
    } finally {
      pool.shutdown();
    }
  }
}