Java并發(fā)工具輔助類代碼實例

更新時間：2020年04月16日 10:04:40 作者：CloverLucky

這篇文章主要介紹了Java并發(fā)工具輔助類代碼實例,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下

java中的并發(fā)工具類

一：等待多線程完成的CountDownLatch

CountDownLatch允許一個或多個線程等待其他線程完成操作。

package com.fuzhulei;
import java.util.concurrent.*;

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

      }
    } catch(Exception e) {
      e.printStackTrace();
    } finally {
      /** 關閉線程池 */
      pool.shutdown();
    }

    countDownLatch.await();
    System.out.println("執(zhí)行完成了");

  }
}

正確執(zhí)行結果：

但是如果我們設置計數(shù)器的容量大于6的話（相對于我的程序而言），就會被阻塞在那里

會發(fā)現(xiàn)執(zhí)行完成了沒有被打印出來，而且程序一直沒有停止，這個時候就是因為計數(shù)器沒有歸0，所以當前線程被阻塞，不能向下面繼續(xù)進行。

二：同步屏障CyclicBarrier

CyclicBarrier的翻譯大致就是可循環(huán)的屏障。它主要的作用就是讓一組線程到達一個屏障（也可以叫做同步點）時被阻塞，直到最后一份線程到達屏障時，屏障才會開門，所有被屏障攔截的線程才會繼續(xù)運行。

package com.fuzhulei;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

/**
 * CyclicBarrier是一個加法計數(shù)器，即同步屏障，可循環(huán)的屏障，讓一組線程到達一個屏障（也可以叫做同步點）時被阻塞，直到最后一個線程到達屏障，達到了一開始初始化的屏障的數(shù)值，
 * 屏障才可以打開門，所有被攔截的線程才可以繼續(xù)工作，主要是通過調(diào)用await方法來實現(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)到達屏障");
    },"A").start();

    new Thread(()->{
      try {
        cyclicBarrier.await();
      } catch (InterruptedException e) {
        e.printStackTrace();
      } catch (BrokenBarrierException e) {
        e.printStackTrace();
      }
      System.out.println("線程B已經(jīng)到達屏障");
    },"B").start();

    new Thread(()->{
      try {
        cyclicBarrier.await();
      } catch (InterruptedException e) {
        e.printStackTrace();
      } catch (BrokenBarrierException e) {
        e.printStackTrace();
      }
      System.out.println("線程C已經(jīng)到達屏障");
    },"C").start();


  }
}

執(zhí)行的結果如下：

但是如果把定義的容量大于3（相對于我的程序而言），就會發(fā)現(xiàn)什么都不會輸出了，看截圖

并且程序一直還沒有停止，這就是屏障起到了作用，因為屏障要求至少需要4個（假設），但是此時只有三個線程到達，所以不滿足，屏障就一直阻攔不放路，那么所有的線程也就被阻塞不能向下面繼續(xù)運行，除非知道第四個過來，滿足條件才會運行。

三：控制并發(fā)線程數(shù)的Semaphore

用來控制同時訪問特定資源的線程數(shù)量，通過協(xié)調(diào)各個線程，以保證合理的使用公用的資源。

package com.fuzhulei;

import java.util.concurrent.*;

/**
 * 用來控制同時訪問特定資源的線程數(shù)量，通過協(xié)調(diào)各個線程，以保證合理的使用公用的資源
 * @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();
    }
  }
}

執(zhí)行的結果如下：