JUC 常用的并發(fā)工具類

/**
   * 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);
  }

package org.dance.tools;

import java.util.concurrent.TimeUnit;

/**
 * 類說(shuō)明：線程休眠輔助工具類
 */
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的使用,有五個(gè)線程,6個(gè)扣除點(diǎn)
 * 扣除完成后主線程和業(yè)務(wù)線程,才能執(zhí)行工作
 * 扣除點(diǎn)一般都是大于等于需要初始化的線程的
 * @author ZYGisComputer
 */
public class UseCountDownLatch {

  /**
   * 設(shè)置為6個(gè)扣除點(diǎn)
   */
  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è)務(wù)線程
   */
  private static class BusiThread implements Runnable {

    @Override
    public void run() {

      // 業(yè)務(wù)線程需要在等初始化完畢后才能執(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)建單獨(dú)的初始化線程
    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();
    // 啟動(dòng)業(yè)務(wù)線程
    new Thread(new BusiThread()).start();
    // 啟動(dòng)初始化線程
    for (int i = 0; i <= 3; i++) {
      new Thread(new InitThread()).start();
    }
    // 主線程進(jìn)入等待
    try {
      countDownLatch.await();
      System.out.println("Main do ites work.....");
    } catch (InterruptedException e) {
      e.printStackTrace();
    }

  }

}

/**
   * 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;
  }

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é)果打印線程
   * 用來(lái)演示CyclicBarrier的第二個(gè)參數(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() {

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

      // 放入統(tǒng)計(jì)容器中
      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();
    }

  }

}

/**
   * 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);
  }

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ù)庫(kù)連接
 * @author ZYGisComputer
 */
public class SqlConnection implements Connection {

  /**
   * 獲取數(shù)據(jù)庫(kù)連接
   * @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<&#63;>> getTypeMap() throws SQLException {
    return null;
  }

  @Override
  public void setTypeMap(Map<String, Class<&#63;>> 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<&#63;> 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;

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

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

  /**
   * useful 代表可用連接
   * useless 代表已用連接
   * 為什么要使用兩個(gè)Semaphore呢?是因?yàn)?在連接池中不只有連接本身是資源,空位也是資源,也需要記錄
   */
  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ù)庫(kù)連接
   *
   * @return 連接對(duì)象
   */
  public Connection takeConnection() throws InterruptedException {
    // 可用許可證減一
    useful.acquire();
    Connection connection;
    synchronized (POOL) {
      connection = POOL.removeFirst();
    }
    // 不可用許可證數(shù)量加一
    useless.release();
    return connection;
  }

  /**
   * 釋放鏈接
   *
   * @param connection 連接對(duì)象
   */
  public void returnConnection(Connection connection) throws InterruptedException {
    if(null!=connection){
      // 打印日志
      System.out.println("當(dāng)前有"+useful.getQueueLength()+"個(gè)線程等待獲取連接,,"
          +"可用連接有"+useful.availablePermits()+"個(gè)");
      // 不可用許可證減一
      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;

/**
 * 測(cè)試Semaphore
 * @author ZYGisComputer
 */
public class UseSemaphore {

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

  private static class BusiThread extends Thread{
    @Override
    public void run() {
      // 隨機(jī)數(shù)工具類 為了讓每個(gè)線程持有連接的時(shí)間不一樣
      Random random = new Random();
      long start = System.currentTimeMillis();
      try {
        Connection connection = pool.takeConnection();
        System.out.println("Thread_"+Thread.currentThread().getId()+
            "_獲取數(shù)據(jù)庫(kù)連接耗時(shí)["+(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();
    }
  }

}

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

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();

  }

}