Java

什么是JUC?

JUC就是java.util.concurrent包,这个包俗称JUC,里面都是解决并发问题的一些东西
该包的位置位于java下面的rt.jar包下面

4大常用并发工具类

CountDownLatch
CyclicBarrier
Semaphore
ExChanger

CountDownLatch

CountDownLatch,俗称闭锁,作用是类似加强版的Join,是让一组线程等待其他的线程完成工作以后才执行
就比如在启动框架服务的时候,主线程需要在环境线程初始化完成之后才能启动,这时候就可以实现使用CountDownLatch来完成

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

在源码中可以看到,创建CountDownLatch时,需要传入一个int类型的参数,将决定在执行次扣减之后,等待的线程被唤醒
image.png
通过这个类图就可以知道其实CountDownLatch并没有多少东西
方法介绍:
CountDownLatch:初始化方法
await:等待方法,同时带参数的是超时重载方法
countDown:每执行一次,计数器减一,就是初始化传入的数字,也代表着一个线程完成了任务
getCount:获取当前值
toString:这个就不用说了
里面的Sync是一个内部类,外面的方法其实都是操作这个内部类的,这个内部类继承了AQS,实现的标准方法,AQS将在后面的章节写
2021-09-17-12-13-17-488745.png
主线程中创建CountDownLatch(3),然后主线程await阻塞,然后线程A,B,C各自完成了任务,调用了countDown,之后,每个线程调用一次计数器就会减一,初始是3,然后A线程调用后变成2,B线程调用后变成1,C线程调用后,变成0,这时就会唤醒正在await的主线程,然后主线程继续执行
上代码:休眠工具类,之后的代码都会用到

  1. import java.util.concurrent.TimeUnit;
  3. /**
  4.  * 类说明：线程休眠辅助工具类
  5.  */
  6. public class SleepTools {
  8.     /**
  9.      * 按秒休眠
  10.      * @param seconds 秒数
  11.      */
  12.     public static final void second(int seconds) {
  13.         try {
  14.             TimeUnit.SECONDS.sleep(seconds);
  15.         } catch (InterruptedException e) {
  16.         }
  17.     }
  19.     /**
  20.      * 按毫秒数休眠
  21.      * @param seconds 毫秒数
  22.      */
  23.     public static final void ms(int seconds) {
  24.         try {
  25.             TimeUnit.MILLISECONDS.sleep(seconds);
  26.         } catch (InterruptedException e) {
  27.         }
  28.     }
  29. }
  1. import org.dance.tools.SleepTools;
  3. import java.util.concurrent.CountDownLatch;
  5. /**
  6.  * CountDownLatch的使用,有五个线程,6个扣除点
  7.  * 扣除完成后主线程和业务线程,才能执行工作
  8.  * 扣除点一般都是大于等于需要初始化的线程的
  9.  */
  10. public class UseCountDownLatch {
  12.     /**
  13.      * 设置为6个扣除点
  14.      */
  15.     static CountDownLatch countDownLatch = new CountDownLatch(6);
  17.     /**
  18.      * 初始化线程
  19.      */
  20.     private static class InitThread implements Runnable {
  22.         @Override
  23.         public void run() {
  25.             System.out.println("thread_" + Thread.currentThread().getId() + " ready init work .....");
  27.             // 执行扣减 扣减不代表结束
  28.             countDownLatch.countDown();
  30.             for (int i = 0; i < 2; i++) {
  31.                 System.out.println("thread_" + Thread.currentThread().getId() + ".....continue do its work");
  32.             }
  34.         }
  35.     }
  37.     /**
  38.      * 业务线程
  39.      */
  40.     private static class BusiThread implements Runnable {
  42.         @Override
  43.         public void run() {
  45.             // 业务线程需要在等初始化完毕后才能执行
  46.             try {
  47.                 countDownLatch.await();
  48.                 for (int i = 0; i < 3; i++) {
  49.                     System.out.println("BusiThread " + Thread.currentThread().getId() + " do business-----");
  50.                 }
  51.             } catch (InterruptedException e) {
  52.                 e.printStackTrace();
  53.             }
  54.         }
  55.     }
  57.     public static void main(String[] args) {
  59.         // 创建单独的初始化线程
  60.         new Thread(){
  61.             @Override
  62.             public void run() {
  63.                 SleepTools.ms(1);
  64.                 System.out.println("thread_" + Thread.currentThread().getId() + " ready init work step 1st.....");
  65.                 // 扣减一次
  66.                 countDownLatch.countDown();
  67.                 System.out.println("begin stop 2nd.....");
  68.                 SleepTools.ms(1);
  69.                 System.out.println("thread_" + Thread.currentThread().getId() + " ready init work step 2nd.....");
  70.                 // 扣减一次
  71.                 countDownLatch.countDown();
  73.             }
  74.         }.start();
  75.         // 启动业务线程
  76.         new Thread(new BusiThread()).start();
  77.         // 启动初始化线程
  78.         for (int i = 0; i <= 3; i++) {
  79.             new Thread(new InitThread()).start();
  80.         }
  81.         // 主线程进入等待
  82.         try {
  83.             countDownLatch.await();
  84.             System.out.println("Main do ites work.....");
  85.         } catch (InterruptedException e) {
  86.             e.printStackTrace();
  87.         }
  89.     }
  91. }

返回结果:

  1. thread_13 ready init work .....
  2. thread_13.....continue do its work
  3. thread_13.....continue do its work
  4. thread_14 ready init work .....
  5. thread_14.....continue do its work
  6. thread_14.....continue do its work
  7. thread_15 ready init work .....
  8. thread_15.....continue do its work
  9. thread_11 ready init work step 1st.....
  10. begin stop 2nd.....
  11. thread_16 ready init work .....
  12. thread_16.....continue do its work
  13. thread_16.....continue do its work
  14. thread_15.....continue do its work
  15. thread_11 ready init work step 2nd.....
  16. Main do ites work.....
  17. BusiThread 12 do business-----
  18. BusiThread 12 do business-----
  19. BusiThread 12 do business-----

通过返回结果就可以很直接的看到业务线程是在初始化线程完全跑完之后,才开始执行的

CyclicBarrier

CyclicBarrier,俗称栅栏锁,作用是让一组线程到达某个屏障,被阻塞,一直到组内的最后一个线程到达,然后屏障开放,接着,所有的线程继续运行
这个感觉和CountDownLatch有点相似,但是其实是不一样的,所谓的差别,将在下面详解
CyclicBarrier的构造参数有两个

  1. /**
  2.      * Creates a new {@code CyclicBarrier} that will trip when the
  3.      * given number of parties (threads) are waiting upon it, and
  4.      * does not perform a predefined action when the barrier is tripped.
  5.      *
  6.      * @param parties the number of threads that must invoke {@link #await}
  7.      *        before the barrier is tripped
  8.      * @throws IllegalArgumentException if {@code parties} is less than 1
  9.      */
  10. public CyclicBarrier(int parties) {
  11.     this(parties, null);
  12. }
  14. /**
  15.      * Creates a new {@code CyclicBarrier} that will trip when the
  16.      * given number of parties (threads) are waiting upon it, and which
  17.      * will execute the given barrier action when the barrier is tripped,
  18.      * performed by the last thread entering the barrier.
  19.      *
  20.      * @param parties the number of threads that must invoke {@link #await}
  21.      *        before the barrier is tripped
  22.      * @param barrierAction the command to execute when the barrier is
  23.      *        tripped, or {@code null} if there is no action
  24.      * @throws IllegalArgumentException if {@code parties} is less than 1
  25.      */
  26. public CyclicBarrier(int parties, Runnable barrierAction) {
  27.     if (parties <= 0) throw new IllegalArgumentException();
  28.     this.parties = parties;
  29.     this.count = parties;
  30.     this.barrierCommand = barrierAction;
  31. }

很明显能感觉出来,上面的构造参数调用了下面的构造参数,是一个构造方法重载
首先这个第一个参数也树Int类型的,传入的是执行线程的个数,这个数量和CountDownLatch不一样,这个数量是需要和线程数量吻合的,CountDownLatch则不一样,CountDownLatch可以大于等于,而CyclicBarrier只能等于,然后是第二个参数,第二个参数是barrierAction,这个参数是当屏障开放后,执行的任务线程,如果当屏障开放后需要执行什么任务,可以写在这个线程中
2021-09-17-12-13-17-671753.png
主线程创建CyclicBarrier(3,barrierAction),然后由线程开始执行,线程A,B执行完成后都调用了await,然后他们都在一个屏障前阻塞者,需要等待线程C也,执行完成,调用await之后,然后三个线程都达到屏障后,屏障开放,然后线程继续执行,并且barrierAction在屏障开放的一瞬间也开始执行
上代码:

  1. import org.dance.tools.SleepTools;
  3. import java.util.Map;
  4. import java.util.Random;
  5. import java.util.concurrent.BrokenBarrierException;
  6. import java.util.concurrent.ConcurrentHashMap;
  7. import java.util.concurrent.CyclicBarrier;
  9. /**
  10.  * CyclicBarrier的使用
  11.  */
  12. public class UseCyclicBarrier {
  14.     /**
  15.      * 存放子线程工作结果的安全容器
  16.      */
  17.     private static ConcurrentHashMap<String, Long> resultMap = new ConcurrentHashMap<>();
  19.     private static CyclicBarrier cyclicBarrier = new CyclicBarrier(5,new CollectThread());
  21.     /**
  22.      * 结果打印线程
  23.      * 用来演示CyclicBarrier的第二个参数,barrierAction
  24.      */
  25.     private static class CollectThread implements Runnable {
  27.         @Override
  28.         public void run() {
  30.             StringBuffer result = new StringBuffer();
  32.             for (Map.Entry<String, Long> workResult : resultMap.entrySet()) {
  33.                 result.append("[" + workResult.getValue() + "]");
  34.             }
  36.             System.out.println("the result = " + result);
  37.             System.out.println("do other business.....");
  39.         }
  40.     }
  42.     /**
  43.      * 工作子线程
  44.      * 用于CyclicBarrier的一组线程
  45.      */
  46.     private static class SubThread implements Runnable {
  48.         @Override
  49.         public void run() {
  51.             // 获取当前线程的ID
  52.             long id = Thread.currentThread().getId();
  54.             // 放入统计容器中
  55.             resultMap.put(String.valueOf(id), id);
  57.             Random random = new Random();
  59.             try {
  60.                 if (random.nextBoolean()) {
  61.                     Thread.sleep(1000 + id);
  62.                     System.out.println("Thread_"+id+"..... do something");
  63.                 }
  64.                 System.out.println(id+" is await");
  65.                 cyclicBarrier.await();
  66.                 Thread.sleep(1000+id);
  67.                 System.out.println("Thread_"+id+".....do its business");
  68.             } catch (InterruptedException e) {
  69.                 e.printStackTrace();
  70.             } catch (BrokenBarrierException e) {
  71.                 e.printStackTrace();
  72.             }
  74.         }
  75.     }
  77.     public static void main(String[] args) {
  79.         for (int i = 0; i <= 4; i++) {
  80.             Thread thread = new Thread(new SubThread());
  81.             thread.start();
  82.         }
  84.     }
  86. }

返回结果:

  1. 11 is await
  2. 14 is await
  3. 15 is await
  4. Thread_12..... do something
  5. 12 is await
  6. Thread_13..... do something
  7. 13 is await
  8. the result = [11][12][13][14][15]
  9. do other business.....
  10. Thread_11.....do its business
  11. Thread_12.....do its business
  12. Thread_13.....do its business
  13. Thread_14.....do its business
  14. Thread_15.....do its business

通过返回结果可以看出前面的11 14 15三个线程没有进入if语句块,在执行到await的时候进入了等待,而另外12 13两个线程进入到了if语句块当中,多休眠了1秒多,然后当5个线程同时到达await的时候,屏障开放,执行了barrierAction线程,然后线程组继续执行
解释一下CountDownLatch和CyclicBarrier的区别吧!
首先就是CountDownLatch的构造参数传入的数量一般都是大于等于线程,数量的,因为他是有第三方控制的,可以扣减多次,然后就是CyclicBarrier的构造参数第一个参数传入的数量一定是等于线程的个数的,因为他是由一组线程自身控制的
区别

CountDownLatch CyclicBarrier
控制 第三方控制 自身控制
传入数量 大于等于线程数量 等于线程数量

Semaphore

Semaphore,俗称信号量,作用于控制同时访问某个特定资源的线程数量,用在流量控制
一说特定资源控制,那么第一时间就想到了数据库连接..
之前用等待超时模式写了一个数据库连接池,打算用这个Semaphone也写一个

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

在源码中可以看到在构建Semaphore信号量的时候,需要传入许可证的数量,这个数量就是资源的最大允许的访问的线程数
接下里用信号量实现一个数据库连接池
连接对象

  1. import org.dance.tools.SleepTools;
  3. import java.sql.*;
  4. import java.util.Map;
  5. import java.util.Properties;
  6. import java.util.concurrent.Executor;
  8. /**
  9.  * 数据库连接
  10.  */
  11. public class SqlConnection implements Connection {
  13.     /**
  14.      * 获取数据库连接
  15.      * @return
  16.      */
  17.     public static final Connection fetchConnection(){
  18.         return new SqlConnection();
  19.     }
  21.     @Override
  22.     public void commit() throws SQLException {
  23.         SleepTools.ms(70);
  24.     }
  26.     @Override
  27.     public Statement createStatement() throws SQLException {
  28.         SleepTools.ms(1);
  29.         return null;
  30.     }
  32.     @Override
  33.     public PreparedStatement prepareStatement(String sql) throws SQLException {
  34.         return null;
  35.     }
  37.     @Override
  38.     public CallableStatement prepareCall(String sql) throws SQLException {
  39.         return null;
  40.     }
  42.     @Override
  43.     public String nativeSQL(String sql) throws SQLException {
  44.         return null;
  45.     }
  47.     @Override
  48.     public void setAutoCommit(boolean autoCommit) throws SQLException {
  50.     }
  52.     @Override
  53.     public boolean getAutoCommit() throws SQLException {
  54.         return false;
  55.     }
  57.     @Override
  58.     public void rollback() throws SQLException {
  60.     }
  62.     @Override
  63.     public void close() throws SQLException {
  65.     }
  67.     @Override
  68.     public boolean isClosed() throws SQLException {
  69.         return false;
  70.     }
  72.     @Override
  73.     public DatabaseMetaData getMetaData() throws SQLException {
  74.         return null;
  75.     }
  77.     @Override
  78.     public void setReadOnly(boolean readOnly) throws SQLException {
  80.     }
  82.     @Override
  83.     public boolean isReadOnly() throws SQLException {
  84.         return false;
  85.     }
  87.     @Override
  88.     public void setCatalog(String catalog) throws SQLException {
  90.     }
  92.     @Override
  93.     public String getCatalog() throws SQLException {
  94.         return null;
  95.     }
  97.     @Override
  98.     public void setTransactionIsolation(int level) throws SQLException {
  100.     }
  102.     @Override
  103.     public int getTransactionIsolation() throws SQLException {
  104.         return 0;
  105.     }
  107.     @Override
  108.     public SQLWarning getWarnings() throws SQLException {
  109.         return null;
  110.     }
  112.     @Override
  113.     public void clearWarnings() throws SQLException {
  115.     }
  117.     @Override
  118.     public Statement createStatement(int resultSetType, int resultSetConcurrency) throws SQLException {
  119.         return null;
  120.     }
  122.     @Override
  123.     public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency) throws SQLException {
  124.         return null;
  125.     }
  127.     @Override
  128.     public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency) throws SQLException {
  129.         return null;
  130.     }
  132.     @Override
  133.     public Map<String, Class<?>> getTypeMap() throws SQLException {
  134.         return null;
  135.     }
  137.     @Override
  138.     public void setTypeMap(Map<String, Class<?>> map) throws SQLException {
  140.     }
  142.     @Override
  143.     public void setHoldability(int holdability) throws SQLException {
  145.     }
  147.     @Override
  148.     public int getHoldability() throws SQLException {
  149.         return 0;
  150.     }
  152.     @Override
  153.     public Savepoint setSavepoint() throws SQLException {
  154.         return null;
  155.     }
  157.     @Override
  158.     public Savepoint setSavepoint(String name) throws SQLException {
  159.         return null;
  160.     }
  162.     @Override
  163.     public void rollback(Savepoint savepoint) throws SQLException {
  165.     }
  167.     @Override
  168.     public void releaseSavepoint(Savepoint savepoint) throws SQLException {
  170.     }
  172.     @Override
  173.     public Statement createStatement(int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException {
  174.         return null;
  175.     }
  177.     @Override
  178.     public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException {
  179.         return null;
  180.     }
  182.     @Override
  183.     public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException {
  184.         return null;
  185.     }
  187.     @Override
  188.     public PreparedStatement prepareStatement(String sql, int autoGeneratedKeys) throws SQLException {
  189.         return null;
  190.     }
  192.     @Override
  193.     public PreparedStatement prepareStatement(String sql, int[] columnIndexes) throws SQLException {
  194.         return null;
  195.     }
  197.     @Override
  198.     public PreparedStatement prepareStatement(String sql, String[] columnNames) throws SQLException {
  199.         return null;
  200.     }
  202.     @Override
  203.     public Clob createClob() throws SQLException {
  204.         return null;
  205.     }
  207.     @Override
  208.     public Blob createBlob() throws SQLException {
  209.         return null;
  210.     }
  212.     @Override
  213.     public NClob createNClob() throws SQLException {
  214.         return null;
  215.     }
  217.     @Override
  218.     public SQLXML createSQLXML() throws SQLException {
  219.         return null;
  220.     }
  222.     @Override
  223.     public boolean isValid(int timeout) throws SQLException {
  224.         return false;
  225.     }
  227.     @Override
  228.     public void setClientInfo(String name, String value) throws SQLClientInfoException {
  230.     }
  232.     @Override
  233.     public void setClientInfo(Properties properties) throws SQLClientInfoException {
  235.     }
  237.     @Override
  238.     public String getClientInfo(String name) throws SQLException {
  239.         return null;
  240.     }
  242.     @Override
  243.     public Properties getClientInfo() throws SQLException {
  244.         return null;
  245.     }
  247.     @Override
  248.     public Array createArrayOf(String typeName, Object[] elements) throws SQLException {
  249.         return null;
  250.     }
  252.     @Override
  253.     public Struct createStruct(String typeName, Object[] attributes) throws SQLException {
  254.         return null;
  255.     }
  257.     @Override
  258.     public void setSchema(String schema) throws SQLException {
  260.     }
  262.     @Override
  263.     public String getSchema() throws SQLException {
  264.         return null;
  265.     }
  267.     @Override
  268.     public void abort(Executor executor) throws SQLException {
  270.     }
  272.     @Override
  273.     public void setNetworkTimeout(Executor executor, int milliseconds) throws SQLException {
  275.     }
  277.     @Override
  278.     public int getNetworkTimeout() throws SQLException {
  279.         return 0;
  280.     }
  282.     @Override
  283.     public <T> T unwrap(Class<T> iface) throws SQLException {
  284.         return null;
  285.     }
  287.     @Override
  288.     public boolean isWrapperFor(Class<?> iface) throws SQLException {
  289.         return false;
  290.     }
  291. }

连接池对象

  1. import java.sql.Connection;
  2. import java.util.ArrayList;
  3. import java.util.HashSet;
  4. import java.util.Iterator;
  5. import java.util.LinkedList;
  6. import java.util.concurrent.Semaphore;
  8. /**
  9.  * 使用信号量控制数据库的链接和释放
  10.  */
  11. public class DBPoolSemaphore {
  13.     /**
  14.      * 池容量
  15.      */
  16.     private final static int POOL_SIZE = 10;
  18.     /**
  19.      * useful 代表可用连接
  20.      * useless 代表已用连接
  21.      *  为什么要使用两个Semaphore呢?是因为,在连接池中不只有连接本身是资源,空位也是资源,也需要记录
  22.      */
  23.     private final Semaphore useful, useless;
  25.     /**
  26.      * 连接池
  27.      */
  28.     private final static LinkedList<Connection> POOL = new LinkedList<>();
  30.     /**
  31.      * 使用静态块初始化池
  32.      */
  33.     static {
  34.         for (int i = 0; i < POOL_SIZE; i++) {
  35.             POOL.addLast(SqlConnection.fetchConnection());
  36.         }
  37.     }
  39.     public DBPoolSemaphore() {
  40.         // 初始可用的许可证等于池容量
  41.         useful = new Semaphore(POOL_SIZE);
  42.         // 初始不可用的许可证容量为0
  43.         useless = new Semaphore(0);
  44.     }
  46.     /**
  47.      * 获取数据库连接
  48.      *
  49.      * @return 连接对象
  50.      */
  51.     public Connection takeConnection() throws InterruptedException {
  52.         // 可用许可证减一
  53.         useful.acquire();
  54.         Connection connection;
  55.         synchronized (POOL) {
  56.             connection = POOL.removeFirst();
  57.         }
  58.         // 不可用许可证数量加一
  59.         useless.release();
  60.         return connection;
  61.     }
  63.     /**
  64.      * 释放链接
  65.      *
  66.      * @param connection 连接对象
  67.      */
  68.     public void returnConnection(Connection connection) throws InterruptedException {
  69.         if(null!=connection){
  70.             // 打印日志
  71.             System.out.println("当前有"+useful.getQueueLength()+"个线程等待获取连接,,"
  72.                     +"可用连接有"+useful.availablePermits()+"个");
  73.             // 不可用许可证减一
  74.             useless.acquire();
  75.             synchronized (POOL){
  76.                 POOL.addLast(connection);
  77.             }
  78.             // 可用许可证加一
  79.             useful.release();
  80.         }
  81.     }
  83. }

测试类:

  1. import org.dance.tools.SleepTools;
  3. import java.sql.Connection;
  4. import java.util.Random;
  6. /**
  7.  * 测试Semaphore
  8.  */
  9. public class UseSemaphore {
  11.     /**
  12.      * 连接池
  13.      */
  14.     public static final DBPoolSemaphore pool = new DBPoolSemaphore();
  16.     private static class BusiThread extends Thread{
  17.         @Override
  18.         public void run() {
  19.             // 随机数工具类 为了让每个线程持有连接的时间不一样
  20.             Random random = new Random();
  21.             long start = System.currentTimeMillis();
  22.             try {
  23.                 Connection connection = pool.takeConnection();
  24.                 System.out.println("Thread_"+Thread.currentThread().getId()+
  25.                         "_获取数据库连接耗时["+(System.currentTimeMillis()-start)+"]ms.");
  26.                 // 模拟使用连接查询数据
  27.                 SleepTools.ms(100+random.nextInt(100));
  28.                 System.out.println("查询数据完成归还连接");
  29.                 pool.returnConnection(connection);
  30.             } catch (InterruptedException e) {
  31.                 e.printStackTrace();
  32.             }
  33.         }
  34.     }
  36.     public static void main(String[] args) {
  37.         for (int i = 0; i < 50; i++) {
  38.             BusiThread busiThread = new BusiThread();
  39.             busiThread.start();
  40.         }
  41.     }
  43. }

测试返回结果:

  1. Thread_11_获取数据库连接耗时[0]ms.
  2. Thread_12_获取数据库连接耗时[0]ms.
  3. Thread_13_获取数据库连接耗时[0]ms.
  4. Thread_14_获取数据库连接耗时[0]ms.
  5. Thread_15_获取数据库连接耗时[0]ms.
  6. Thread_16_获取数据库连接耗时[0]ms.
  7. Thread_17_获取数据库连接耗时[0]ms.
  8. Thread_18_获取数据库连接耗时[0]ms.
  9. Thread_19_获取数据库连接耗时[0]ms.
  10. Thread_20_获取数据库连接耗时[0]ms.
  11. 查询数据完成归还连接
  12. 当前有40个线程等待获取连接,,可用连接有0个
  13. Thread_21_获取数据库连接耗时[112]ms.
  14. 查询数据完成归还连接
  15. ...................
  16. 查询数据完成归还连接
  17. 当前有2个线程等待获取连接,,可用连接有0个
  18. Thread_59_获取数据库连接耗时[637]ms.
  19. 查询数据完成归还连接
  20. 当前有1个线程等待获取连接,,可用连接有0个
  21. Thread_60_获取数据库连接耗时[660]ms.
  22. 查询数据完成归还连接
  23. 当前有0个线程等待获取连接,,可用连接有0个
  24. 查询数据完成归还连接
  25. ...................
  26. 当前有0个线程等待获取连接,,可用连接有8个
  27. 查询数据完成归还连接
  28. 当前有0个线程等待获取连接,,可用连接有9个

通过执行结果可以很明确的看到,一上来就有10个线程获取到了连接,,然后后面的40个线程进入阻塞,然后只有释放链接之后,等待的线程就会有一个拿到,然后越后面的线程等待的时间就越长,然后一直到所有的线程执行完毕
最后打印的可用连接有九个不是因为少了一个是因为在释放之前打印的,不是错误
从结果中可以看到,对连接池中的资源的到了控制,这就是信号量的流量控制

Exchanger

Exchanger,俗称交换器,用于在线程之间交换数据,但是比较受限,因为只能两个线程之间交换数据

  1. /**
  2.  * Creates a new Exchanger.
  3.  */
  4. public Exchanger() {
  5.     participant = new Participant();
  6. }

这个构造函数没有什么好说的,也没有入参,只有在创建的时候指定一下需要交换的数据的泛型即可,下面看代码

  1. import java.util.HashSet;
  2. import java.util.Set;
  3. import java.util.concurrent.Exchanger;
  5. /**
  6.  * 线程之间交换数据
  7.  */
  8. public class UseExchange {
  10.     private static final Exchanger<Set<String>> exchanger = new Exchanger<>();
  12.     public static void main(String[] args) {
  14.         new Thread(){
  15.             @Override
  16.             public void run() {
  17.                 Set<String> aSet = new HashSet<>();
  18.                 aSet.add("A");
  19.                 aSet.add("B");
  20.                 aSet.add("C");
  21.                 try {
  22.                     Set<String> exchange = exchanger.exchange(aSet);
  23.                     for (String s : exchange) {
  24.                         System.out.println("aSet"+s);
  25.                     }
  26.                 } catch (InterruptedException e) {
  27.                     e.printStackTrace();
  28.                 }
  29.             }
  30.         }.start();
  32.         new Thread(){
  33.             @Override
  34.             public void run() {
  35.                 Set<String> bSet = new HashSet<>();
  36.                 bSet.add("1");
  37.                 bSet.add("2");
  38.                 bSet.add("3");
  39.                 try {
  40.                     Set<String> exchange = exchanger.exchange(bSet);
  41.                     for (String s : exchange) {
  42.                         System.out.println("bSet"+s);
  43.                     }
  44.                 } catch (InterruptedException e) {
  45.                     e.printStackTrace();
  46.                 }
  47.             }
  48.         }.start();
  50.     }
  52. }

执行结果:

  1. bSetA
  2. bSetB
  3. bSetC
  4. aSet1
  5. aSet2
  6. aSet3

通过执行结果可以清晰的看到,两个线程中的数据发生了交换,这就是Exchanger的线程数据交换了
以上就是JUC的4大常用并发工具类了。