詳解JUC 常用4大并發(fā)工具類

更新時間：2020年10月09日 11:05:22 作者：彼岸舞

這篇文章主要介紹了JUC 常用4大并發(fā)工具類的相關(guān)資料，幫助大家更好的理解和學習Java 并發(fā)編程，感興趣的朋友可以了解下

什么是JUC?

　　JUC就是java.util.concurrent包,這個包俗稱JUC,里面都是解決并發(fā)問題的一些東西

　　該包的位置位于java下面的rt.jar包下面

4大常用并發(fā)工具類:

　　CountDownLatch
　　CyclicBarrier
　　Semaphore
　　ExChanger

CountDownLatch:

　　CountDownLatch,俗稱閉鎖,作用是類似加強版的Join,是讓一組線程等待其他的線程完成工作以后才執(zhí)行

　　就比如在啟動框架服務的時候,我們主線程需要在環(huán)境線程初始化完成之后才能啟動,這時候我們就可以實現(xiàn)使用CountDownLatch來完成

/**
   * Constructs a {@code CountDownLatch} initialized with the given count.
   *
   * @param count the number of times {@link #countDown} must be invoked
   *    before threads can pass through {@link #await}
   * @throws IllegalArgumentException if {@code count} is negative
   */
  public CountDownLatch(int count) {
    if (count < 0) throw new IllegalArgumentException("count < 0");
    this.sync = new Sync(count);
  }

　　在源碼中可以看到,創(chuàng)建CountDownLatch時,需要傳入一個int類型的參數(shù),將決定在執(zhí)行次扣減之后,等待的線程被喚醒

　　通過這個類圖就可以知道其實CountDownLatch并沒有多少東西

　　方法介紹:

　　　　CountDownLatch:初始化方法
　　　　await:等待方法,同時帶參數(shù)的是超時重載方法
　　　　countDown:每執(zhí)行一次,計數(shù)器減一,就是初始化傳入的數(shù)字,也代表著一個線程完成了任務
　　　　getCount:獲取當前值
　　　　toString:這個就不用說了

　　里面的Sync是一個內(nèi)部類,外面的方法其實都是操作這個內(nèi)部類的,這個內(nèi)部類繼承了AQS,實現(xiàn)的標準方法,AQS將在后面的章節(jié)寫

主線程中創(chuàng)建CountDownLatch(3),然后主線程await阻塞,然后線程A,B,C各自完成了任務,調(diào)用了countDown,之后,每個線程調(diào)用一次計數(shù)器就會減一,初始是3,然后A線程調(diào)用后變成2,B線程調(diào)用后變成1,C線程調(diào)用后,變成0,這時就會喚醒正在await的主線程,然后主線程繼續(xù)執(zhí)行

說一千道一萬,不如代碼寫幾行,上代碼:

休眠工具類,之后的代碼都會用到

package org.dance.tools;

import java.util.concurrent.TimeUnit;

/**
 * 類說明：線程休眠輔助工具類
 */
public class SleepTools {

  /**
   * 按秒休眠
   * @param seconds 秒數(shù)
   */
  public static final void second(int seconds) {
    try {
      TimeUnit.SECONDS.sleep(seconds);
    } catch (InterruptedException e) {
    }
  }

  /**
   * 按毫秒數(shù)休眠
   * @param seconds 毫秒數(shù)
   */
  public static final void ms(int seconds) {
    try {
      TimeUnit.MILLISECONDS.sleep(seconds);
    } catch (InterruptedException e) {
    }
  }
}

package org.dance.day2.util;

import org.dance.tools.SleepTools;

import java.util.concurrent.CountDownLatch;

/**
 * CountDownLatch的使用,有五個線程,6個扣除點
 * 扣除完成后主線程和業(yè)務線程,才能執(zhí)行工作
 * 扣除點一般都是大于等于需要初始化的線程的
 * @author ZYGisComputer
 */
public class UseCountDownLatch {

  /**
   * 設置為6個扣除點
   */
  static CountDownLatch countDownLatch = new CountDownLatch(6);

  /**
   * 初始化線程
   */
  private static class InitThread implements Runnable {

    @Override
    public void run() {

      System.out.println("thread_" + Thread.currentThread().getId() + " ready init work .....");

      // 執(zhí)行扣減 扣減不代表結(jié)束
      countDownLatch.countDown();

      for (int i = 0; i < 2; i++) {
        System.out.println("thread_" + Thread.currentThread().getId() + ".....continue do its work");
      }

    }
  }

  /**
   * 業(yè)務線程
   */
  private static class BusiThread implements Runnable {

    @Override
    public void run() {

      // 業(yè)務線程需要在等初始化完畢后才能執(zhí)行
      try {
        countDownLatch.await();
        for (int i = 0; i < 3; i++) {
          System.out.println("BusiThread " + Thread.currentThread().getId() + " do business-----");
        }
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
  }

  public static void main(String[] args) {

    // 創(chuàng)建單獨的初始化線程
    new Thread(){
      @Override
      public void run() {
        SleepTools.ms(1);
        System.out.println("thread_" + Thread.currentThread().getId() + " ready init work step 1st.....");
        // 扣減一次
        countDownLatch.countDown();
        System.out.println("begin stop 2nd.....");
        SleepTools.ms(1);
        System.out.println("thread_" + Thread.currentThread().getId() + " ready init work step 2nd.....");
        // 扣減一次
        countDownLatch.countDown();

      }
    }.start();
    // 啟動業(yè)務線程
    new Thread(new BusiThread()).start();
    // 啟動初始化線程
    for (int i = 0; i <= 3; i++) {
      new Thread(new InitThread()).start();
    }
    // 主線程進入等待
    try {
      countDownLatch.await();
      System.out.println("Main do ites work.....");
    } catch (InterruptedException e) {
      e.printStackTrace();
    }

  }

}

返回結(jié)果:

thread_13 ready init work .....
thread_13.....continue do its work
thread_13.....continue do its work
thread_14 ready init work .....
thread_14.....continue do its work
thread_14.....continue do its work
thread_15 ready init work .....
thread_15.....continue do its work
thread_11 ready init work step 1st.....
begin stop 2nd.....
thread_16 ready init work .....
thread_16.....continue do its work
thread_16.....continue do its work
thread_15.....continue do its work
thread_11 ready init work step 2nd.....
Main do ites work.....
BusiThread 12 do business-----
BusiThread 12 do business-----
BusiThread 12 do business-----

通過返回結(jié)果就可以很直接的看到業(yè)務線程是在初始化線程完全跑完之后,才開始執(zhí)行的

CyclicBarrier:

　　CyclicBarrier,俗稱柵欄鎖,作用是讓一組線程到達某個屏障,被阻塞,一直到組內(nèi)的最后一個線程到達,然后屏障開放,接著,所有的線程繼續(xù)運行

　　這個感覺和CountDownLatch有點相似,但是其實是不一樣的,所謂的差別,將在下面詳解

　　CyclicBarrier的構(gòu)造參數(shù)有兩個

/**
   * Creates a new {@code CyclicBarrier} that will trip when the
   * given number of parties (threads) are waiting upon it, and
   * does not perform a predefined action when the barrier is tripped.
   *
   * @param parties the number of threads that must invoke {@link #await}
   *    before the barrier is tripped
   * @throws IllegalArgumentException if {@code parties} is less than 1
   */
  public CyclicBarrier(int parties) {
    this(parties, null);
  }

/**
   * Creates a new {@code CyclicBarrier} that will trip when the
   * given number of parties (threads) are waiting upon it, and which
   * will execute the given barrier action when the barrier is tripped,
   * performed by the last thread entering the barrier.
   *
   * @param parties the number of threads that must invoke {@link #await}
   *    before the barrier is tripped
   * @param barrierAction the command to execute when the barrier is
   *    tripped, or {@code null} if there is no action
   * @throws IllegalArgumentException if {@code parties} is less than 1
   */
  public CyclicBarrier(int parties, Runnable barrierAction) {
    if (parties <= 0) throw new IllegalArgumentException();
    this.parties = parties;
    this.count = parties;
    this.barrierCommand = barrierAction;
  }

很明顯能感覺出來,上面的構(gòu)造參數(shù)調(diào)用了下面的構(gòu)造參數(shù),是一個構(gòu)造方法重載

首先這個第一個參數(shù)也樹Int類型的,傳入的是執(zhí)行線程的個數(shù),這個數(shù)量和CountDownLatch不一樣,這個數(shù)量是需要和線程數(shù)量吻合的,CountDownLatch則不一樣,CountDownLatch可以大于等于,而CyclicBarrier只能等于,然后是第二個參數(shù),第二個參數(shù)是barrierAction,這個參數(shù)是當屏障開放后,執(zhí)行的任務線程,如果當屏障開放后需要執(zhí)行什么任務,可以寫在這個線程中

主線程創(chuàng)建CyclicBarrier(3,barrierAction),然后由線程開始執(zhí)行,線程A,B執(zhí)行完成后都調(diào)用了await,然后他們都在一個屏障前阻塞者,需要等待線程C也,執(zhí)行完成,調(diào)用await之后,然后三個線程都達到屏障后,屏障開放,然后線程繼續(xù)執(zhí)行,并且barrierAction在屏障開放的一瞬間也開始執(zhí)行

上代碼:

package org.dance.day2.util;

import org.dance.tools.SleepTools;

import java.util.Map;
import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CyclicBarrier;

/**
 * CyclicBarrier的使用
 *
 * @author ZYGisComputer
 */
public class UseCyclicBarrier {

  /**
   * 存放子線程工作結(jié)果的安全容器
   */
  private static ConcurrentHashMap<String, Long> resultMap = new ConcurrentHashMap<>();

  private static CyclicBarrier cyclicBarrier = new CyclicBarrier(5,new CollectThread());

  /**
   * 結(jié)果打印線程
   * 用來演示CyclicBarrier的第二個參數(shù),barrierAction
   */
  private static class CollectThread implements Runnable {

    @Override
    public void run() {

      StringBuffer result = new StringBuffer();

      for (Map.Entry<String, Long> workResult : resultMap.entrySet()) {
        result.append("[" + workResult.getValue() + "]");
      }

      System.out.println("the result = " + result);
      System.out.println("do other business.....");

    }
  }

  /**
   * 工作子線程
   * 用于CyclicBarrier的一組線程
   */
  private static class SubThread implements Runnable {

    @Override
    public void run() {

      // 獲取當前線程的ID
      long id = Thread.currentThread().getId();

      // 放入統(tǒng)計容器中
      resultMap.put(String.valueOf(id), id);

      Random random = new Random();

      try {
        if (random.nextBoolean()) {
          Thread.sleep(1000 + id);
          System.out.println("Thread_"+id+"..... do something");
        }
        System.out.println(id+" is await");
        cyclicBarrier.await();
        Thread.sleep(1000+id);
        System.out.println("Thread_"+id+".....do its business");
      } catch (InterruptedException e) {
        e.printStackTrace();
      } catch (BrokenBarrierException e) {
        e.printStackTrace();
      }

    }
  }

  public static void main(String[] args) {

    for (int i = 0; i <= 4; i++) {
      Thread thread = new Thread(new SubThread());
      thread.start();
    }

  }

}

返回結(jié)果:

11 is await
14 is await
15 is await
Thread_12..... do something
12 is await
Thread_13..... do something
13 is await
the result = [11][12][13][14][15]
do other business.....
Thread_11.....do its business
Thread_12.....do its business
Thread_13.....do its business
Thread_14.....do its business
Thread_15.....do its business

通過返回結(jié)果可以看出前面的11 14 15三個線程沒有進入if語句塊,在執(zhí)行到await的時候進入了等待,而另外12 13兩個線程進入到了if語句塊當中,多休眠了1秒多,然后當5個線程同時到達await的時候,屏障開放,執(zhí)行了barrierAction線程,然后線程組繼續(xù)執(zhí)行

解釋一下CountDownLatch和CyclicBarrier的卻別吧!

首先就是CountDownLatch的構(gòu)造參數(shù)傳入的數(shù)量一般都是大于等于線程,數(shù)量的,因為他是有第三方控制的,可以扣減多次,然后就是CyclicBarrier的構(gòu)造參數(shù)第一個參數(shù)傳入的數(shù)量一定是等于線程的個數(shù)的,因為他是由一組線程自身控制的

區(qū)別

	CountDownLatch	CyclicBarrier
控制	第三方控制	自身控制
傳入數(shù)量	大于等于線程數(shù)量	等于線程數(shù)量

Semaphore:

　　Semaphore,俗稱信號量,作用于控制同時訪問某個特定資源的線程數(shù)量,用在流量控制

　　一說特定資源控制,那么第一時間就想到了數(shù)據(jù)庫連接..

　　之前用等待超時模式寫了一個數(shù)據(jù)庫連接池,打算用這個Semaphone也寫一個

/**
   * Creates a {@code Semaphore} with the given number of
   * permits and nonfair fairness setting.
   *
   * @param permits the initial number of permits available.
   *    This value may be negative, in which case releases
   *    must occur before any acquires will be granted.
   */
  public Semaphore(int permits) {
    sync = new NonfairSync(permits);
  }

在源碼中可以看到在構(gòu)建Semaphore信號量的時候,需要傳入許可證的數(shù)量,這個數(shù)量就是資源的最大允許的訪問的線程數(shù)

接下里用信號量實現(xiàn)一個數(shù)據(jù)庫連接池

連接對象

package org.dance.day2.util.pool;

import org.dance.tools.SleepTools;

import java.sql.*;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Executor;

/**
 * 數(shù)據(jù)庫連接
 * @author ZYGisComputer
 */
public class SqlConnection implements Connection {

  /**
   * 獲取數(shù)據(jù)庫連接
   * @return
   */
  public static final Connection fetchConnection(){
    return new SqlConnection();
  }

  @Override
  public void commit() throws SQLException {
    SleepTools.ms(70);
  }

  @Override
  public Statement createStatement() throws SQLException {
    SleepTools.ms(1);
    return null;
  }

  @Override
  public PreparedStatement prepareStatement(String sql) throws SQLException {
    return null;
  }

  @Override
  public CallableStatement prepareCall(String sql) throws SQLException {
    return null;
  }

  @Override
  public String nativeSQL(String sql) throws SQLException {
    return null;
  }

  @Override
  public void setAutoCommit(boolean autoCommit) throws SQLException {

  }

  @Override
  public boolean getAutoCommit() throws SQLException {
    return false;
  }

  @Override
  public void rollback() throws SQLException {

  }

  @Override
  public void close() throws SQLException {

  }

  @Override
  public boolean isClosed() throws SQLException {
    return false;
  }

  @Override
  public DatabaseMetaData getMetaData() throws SQLException {
    return null;
  }

  @Override
  public void setReadOnly(boolean readOnly) throws SQLException {

  }

  @Override
  public boolean isReadOnly() throws SQLException {
    return false;
  }

  @Override
  public void setCatalog(String catalog) throws SQLException {

  }

  @Override
  public String getCatalog() throws SQLException {
    return null;
  }

  @Override
  public void setTransactionIsolation(int level) throws SQLException {

  }

  @Override
  public int getTransactionIsolation() throws SQLException {
    return 0;
  }

  @Override
  public SQLWarning getWarnings() throws SQLException {
    return null;
  }

  @Override
  public void clearWarnings() throws SQLException {

  }

  @Override
  public Statement createStatement(int resultSetType, int resultSetConcurrency) throws SQLException {
    return null;
  }

  @Override
  public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency) throws SQLException {
    return null;
  }

  @Override
  public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency) throws SQLException {
    return null;
  }

  @Override
  public Map<String, Class<?>> getTypeMap() throws SQLException {
    return null;
  }

  @Override
  public void setTypeMap(Map<String, Class<?>> map) throws SQLException {

  }

  @Override
  public void setHoldability(int holdability) throws SQLException {

  }

  @Override
  public int getHoldability() throws SQLException {
    return 0;
  }

  @Override
  public Savepoint setSavepoint() throws SQLException {
    return null;
  }

  @Override
  public Savepoint setSavepoint(String name) throws SQLException {
    return null;
  }

  @Override
  public void rollback(Savepoint savepoint) throws SQLException {

  }

  @Override
  public void releaseSavepoint(Savepoint savepoint) throws SQLException {

  }

  @Override
  public Statement createStatement(int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException {
    return null;
  }

  @Override
  public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException {
    return null;
  }

  @Override
  public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException {
    return null;
  }

  @Override
  public PreparedStatement prepareStatement(String sql, int autoGeneratedKeys) throws SQLException {
    return null;
  }

  @Override
  public PreparedStatement prepareStatement(String sql, int[] columnIndexes) throws SQLException {
    return null;
  }

  @Override
  public PreparedStatement prepareStatement(String sql, String[] columnNames) throws SQLException {
    return null;
  }

  @Override
  public Clob createClob() throws SQLException {
    return null;
  }

  @Override
  public Blob createBlob() throws SQLException {
    return null;
  }

  @Override
  public NClob createNClob() throws SQLException {
    return null;
  }

  @Override
  public SQLXML createSQLXML() throws SQLException {
    return null;
  }

  @Override
  public boolean isValid(int timeout) throws SQLException {
    return false;
  }

  @Override
  public void setClientInfo(String name, String value) throws SQLClientInfoException {

  }

  @Override
  public void setClientInfo(Properties properties) throws SQLClientInfoException {

  }

  @Override
  public String getClientInfo(String name) throws SQLException {
    return null;
  }

  @Override
  public Properties getClientInfo() throws SQLException {
    return null;
  }

  @Override
  public Array createArrayOf(String typeName, Object[] elements) throws SQLException {
    return null;
  }

  @Override
  public Struct createStruct(String typeName, Object[] attributes) throws SQLException {
    return null;
  }

  @Override
  public void setSchema(String schema) throws SQLException {

  }

  @Override
  public String getSchema() throws SQLException {
    return null;
  }

  @Override
  public void abort(Executor executor) throws SQLException {

  }

  @Override
  public void setNetworkTimeout(Executor executor, int milliseconds) throws SQLException {

  }

  @Override
  public int getNetworkTimeout() throws SQLException {
    return 0;
  }

  @Override
  public <T> T unwrap(Class<T> iface) throws SQLException {
    return null;
  }

  @Override
  public boolean isWrapperFor(Class<?> iface) throws SQLException {
    return false;
  }
}

連接池對象

package org.dance.day2.util.pool;

import java.sql.Connection;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.concurrent.Semaphore;

/**
 * 使用信號量控制數(shù)據(jù)庫的鏈接和釋放
 *
 * @author ZYGisComputer
 */
public class DBPoolSemaphore {

  /**
   * 池容量
   */
  private final static int POOL_SIZE = 10;

  /**
   * useful 代表可用連接
   * useless 代表已用連接
   * 為什么要使用兩個Semaphore呢?是因為,在連接池中不只有連接本身是資源,空位也是資源,也需要記錄
   */
  private final Semaphore useful, useless;

  /**
   * 連接池
   */
  private final static LinkedList<Connection> POOL = new LinkedList<>();

  /**
   * 使用靜態(tài)塊初始化池
   */
  static {
    for (int i = 0; i < POOL_SIZE; i++) {
      POOL.addLast(SqlConnection.fetchConnection());
    }
  }

  public DBPoolSemaphore() {
    // 初始可用的許可證等于池容量
    useful = new Semaphore(POOL_SIZE);
    // 初始不可用的許可證容量為0
    useless = new Semaphore(0);
  }

  /**
   * 獲取數(shù)據(jù)庫連接
   *
   * @return 連接對象
   */
  public Connection takeConnection() throws InterruptedException {
    // 可用許可證減一
    useful.acquire();
    Connection connection;
    synchronized (POOL) {
      connection = POOL.removeFirst();
    }
    // 不可用許可證數(shù)量加一
    useless.release();
    return connection;
  }

  /**
   * 釋放鏈接
   *
   * @param connection 連接對象
   */
  public void returnConnection(Connection connection) throws InterruptedException {
    if(null!=connection){
      // 打印日志
      System.out.println("當前有"+useful.getQueueLength()+"個線程等待獲取連接,,"
          +"可用連接有"+useful.availablePermits()+"個");
      // 不可用許可證減一
      useless.acquire();
      synchronized (POOL){
        POOL.addLast(connection);
      }
      // 可用許可證加一
      useful.release();
    }
  }

}

測試類:

package org.dance.day2.util.pool;

import org.dance.tools.SleepTools;

import java.sql.Connection;
import java.util.Random;

/**
 * 測試Semaphore
 * @author ZYGisComputer
 */
public class UseSemaphore {

  /**
   * 連接池
   */
  public static final DBPoolSemaphore pool = new DBPoolSemaphore();

  private static class BusiThread extends Thread{
    @Override
    public void run() {
      // 隨機數(shù)工具類 為了讓每個線程持有連接的時間不一樣
      Random random = new Random();
      long start = System.currentTimeMillis();
      try {
        Connection connection = pool.takeConnection();
        System.out.println("Thread_"+Thread.currentThread().getId()+
            "_獲取數(shù)據(jù)庫連接耗時["+(System.currentTimeMillis()-start)+"]ms.");
        // 模擬使用連接查詢數(shù)據(jù)
        SleepTools.ms(100+random.nextInt(100));
        System.out.println("查詢數(shù)據(jù)完成歸還連接");
        pool.returnConnection(connection);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
  }

  public static void main(String[] args) {
    for (int i = 0; i < 50; i++) {
      BusiThread busiThread = new BusiThread();
      busiThread.start();
    }
  }

}

測試返回結(jié)果:

Thread_11_獲取數(shù)據(jù)庫連接耗時[0]ms.
Thread_12_獲取數(shù)據(jù)庫連接耗時[0]ms.
Thread_13_獲取數(shù)據(jù)庫連接耗時[0]ms.
Thread_14_獲取數(shù)據(jù)庫連接耗時[0]ms.
Thread_15_獲取數(shù)據(jù)庫連接耗時[0]ms.
Thread_16_獲取數(shù)據(jù)庫連接耗時[0]ms.
Thread_17_獲取數(shù)據(jù)庫連接耗時[0]ms.
Thread_18_獲取數(shù)據(jù)庫連接耗時[0]ms.
Thread_19_獲取數(shù)據(jù)庫連接耗時[0]ms.
Thread_20_獲取數(shù)據(jù)庫連接耗時[0]ms.
查詢數(shù)據(jù)完成歸還連接
當前有40個線程等待獲取連接,,可用連接有0個
Thread_21_獲取數(shù)據(jù)庫連接耗時[112]ms.
查詢數(shù)據(jù)完成歸還連接
...................查詢數(shù)據(jù)完成歸還連接
當前有2個線程等待獲取連接,,可用連接有0個
Thread_59_獲取數(shù)據(jù)庫連接耗時[637]ms.
查詢數(shù)據(jù)完成歸還連接
當前有1個線程等待獲取連接,,可用連接有0個
Thread_60_獲取數(shù)據(jù)庫連接耗時[660]ms.
查詢數(shù)據(jù)完成歸還連接
當前有0個線程等待獲取連接,,可用連接有0個
查詢數(shù)據(jù)完成歸還連接...................
當前有0個線程等待獲取連接,,可用連接有8個
查詢數(shù)據(jù)完成歸還連接
當前有0個線程等待獲取連接,,可用連接有9個

通過執(zhí)行結(jié)果可以很明確的看到,一上來就有10個線程獲取到了連接,,然后后面的40個線程進入阻塞,然后只有釋放鏈接之后,等待的線程就會有一個拿到,然后越后面的線程等待的時間就越長,然后一直到所有的線程執(zhí)行完畢

最后打印的可用連接有九個不是因為少了一個是因為在釋放之前打印的,不是錯誤

從結(jié)果中可以看到,我們對連接池中的資源的到了控制,這就是信號量的流量控制

Exchanger:

　　Exchanger,俗稱交換器,用于在線程之間交換數(shù)據(jù),但是比較受限,因為只能兩個線程之間交換數(shù)據(jù)

/**
   * Creates a new Exchanger.
   */
  public Exchanger() {
    participant = new Participant();
  }

這個構(gòu)造函數(shù)沒有什么好說的,也沒有入?yún)?只有在創(chuàng)建的時候指定一下需要交換的數(shù)據(jù)的泛型即可,下面看代碼

package org.dance.day2.util;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.Exchanger;

/**
 * 線程之間交換數(shù)據(jù)
 * @author ZYGisComputer
 */
public class UseExchange {

  private static final Exchanger<Set<String>> exchanger = new Exchanger<>();

  public static void main(String[] args) {

    new Thread(){
      @Override
      public void run() {
        Set<String> aSet = new HashSet<>();
        aSet.add("A");
        aSet.add("B");
        aSet.add("C");
        try {
          Set<String> exchange = exchanger.exchange(aSet);
          for (String s : exchange) {
            System.out.println("aSet"+s);
          }
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    }.start();

    new Thread(){
      @Override
      public void run() {
        Set<String> bSet = new HashSet<>();
        bSet.add("1");
        bSet.add("2");
        bSet.add("3");
        try {
          Set<String> exchange = exchanger.exchange(bSet);
          for (String s : exchange) {
            System.out.println("bSet"+s);
          }
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }
    }.start();

  }

}

執(zhí)行結(jié)果: