【多线程】多线程之并发工具类

痛定思痛。 2022-11-26 09:43 273阅读 0赞

# 一、概述 #

jdk 1.5 后，为我们提供的并发工具类有：

<table> 
 <tbody> 
  <tr> 
   <td><strong>名称</strong></td> 
   <td><strong>描述</strong></td> 
   <td><strong>详细</strong></td> 
  </tr> 
  <tr> 
   <td><strong>CountDownLatch</strong></td> 
   <td>同步计数器</td> 
   <td> <p>初始化时，传入需要计数的线程等待数，并用 await() 阻塞当前线程，其他线程中可以调用&nbsp;countDown（）方法让计数器减一，当计数器为 0 时，则放行</p> </td> 
  </tr> 
  <tr> 
   <td><strong>CyclicBarrier</strong></td> 
   <td>栅栏</td> 
   <td>让一组线程达到某个屏障被阻塞，一直到组内最后一个线程达到屏障时，屏障开放，所有被阻塞的线程才会继续运行</td> 
  </tr> 
  <tr> 
   <td><strong>Semaphore</strong></td> 
   <td>信号量/令牌桶</td> 
   <td>类似于令牌桶算法，通过控制令牌数量，让持有令牌的线程运行，未持有的等待，实现多线程的并发控制，常用于流量控制</td> 
  </tr> 
  <tr> 
   <td><strong>Exchange</strong></td> 
   <td>两个线程的数据交换器</td> 
   <td>通过在两个线程中定义同步点，当两个线程都到达同步点时，交换数据结构</td> 
  </tr> 
 </tbody> 
</table>

# 二、详述 #

## 2.1 CountDownLatch 同步计数器 ##

● **概念**

CountDownLatch 是一个同步计数器，初始化时，传入需要计数的线程等待数，并用 await() 阻塞当前线程，其他线程中可以调用 countDown（）方法让计数器减一，当计数器为 0 时，则放行。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0NDE2MzMx_size_16_color_FFFFFF_t_70][]

● **作用**

用来协调多个线程之间的同步，是一组线程等待其他的线程完成工作以后在执行，相当于加强版join。例如A线程的继续执行需要依赖于线程B、C、D都执行完成，A才能继续处理的情况

● **关键方法**

<table> 
 <tbody> 
  <tr> 
   <td><strong>方法</strong></td> 
   <td><strong>作用</strong></td> 
  </tr> 
  <tr> 
   <td>wait()</td> 
   <td>用于阻塞当前线程，等待计数器计数值减到0</td> 
  </tr> 
  <tr> 
   <td>countDown()</td> 
   <td>将计数器减一</td> 
  </tr> 
 </tbody> 
</table>

● **示例代码**

public class CountDownLatchDemo {
    
        private static CountDownLatch countDownLatch = new CountDownLatch(6);
    
        private static class InitThread extends Thread {
            @Override
            public void run() {
                System.out.println("线程" + Thread.currentThread().getName() + "开始初始化....");
                // 计数减一
                countDownLatch.countDown();
            }
        }
    
        public static void main(String[] args) {
    
            System.out.println("main 开始");
            for (int i = 0; i < 5; i++) {
                new InitThread().start();
            }
    
            try {
                countDownLatch.await();
                System.out.println("main 结束");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

## 2.2 CyclicBarrier 栅栏 ##

● **概念**

CyclicBarrier 可以让一组线程达到某个屏障被阻塞，一直到组内最后一个线程达到屏障时，屏障开放，所有被阻塞的线程才会继续运行

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0NDE2MzMx_size_16_color_FFFFFF_t_70 1][]

● **作用**

协调多线程的同步，让多个线程阻塞在栅栏处，直到相关线程都到了栅栏才会一起放行

● **关键方法**

<table> 
 <tbody> 
  <tr> 
   <td><strong>方法</strong></td> 
   <td><strong>作用</strong></td> 
  </tr> 
  <tr> 
   <td>await()</td> 
   <td>用于阻塞当前线程，等待所有线程都到达屏障</td> 
  </tr> 
 </tbody> 
</table>

● **示例代码**

public class CyclicBarrierDemo {
    
        // 初始化栅栏， parties为栅栏拦住的线程数， CallThread 为自定义的，当栅栏放开后调用的函数
        private static CyclicBarrier cyclicBarrier = new CyclicBarrier(6, new CallThread());
    
        private static ConcurrentHashMap<Long, String> concurrentHashMap = new ConcurrentHashMap<>();
    
        private static class CallThread extends Thread {
            @Override
            public void run() {
                System.out.println("栅栏放开~");
            }
        }
    
        private static class SubThread extends Thread {
            @Override
            public void run() {
                System.out.println("线程" + Thread.currentThread().getName() + "正在等待");
                try {
                    cyclicBarrier.await();
                    System.out.println("线程" + Thread.currentThread().getName() + "处理完毕！");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }
        }
    
        public static void main(String[] args) {
            for (int i = 0; i < 5; i++) {
                new SubThread().start();
            }
    
            try {
                Thread.sleep(3000);
                cyclicBarrier.await();
            } catch (InterruptedException | BrokenBarrierException e) {
                e.printStackTrace();
            }
            System.out.println("主线程完成");
        }
    }

## 2.3 Semaphore 信号量/令牌桶 ##

● **概念**

通过定义令牌数量，让线程进行获取和归还，控制一定时间内的线程通过数量

● **作用**

削峰，控制同一时间执行的线程数量，避免高峰期占用太多线程资源

● **关键方法**

<table> 
 <tbody> 
  <tr> 
   <td><strong>方法</strong></td> 
   <td><strong>作用</strong></td> 
  </tr> 
  <tr> 
   <td>acquire()</td> 
   <td>获取令牌</td> 
  </tr> 
  <tr> 
   <td>release()</td> 
   <td>释放令牌</td> 
  </tr> 
 </tbody> 
</table>

● **示例代码**

public class TestSemphore {
    
        private static SemaphorePool semaphorePool = new SemaphorePool();
    
        private static class TokenThread extends Thread {
            @Override
            public void run() {
                // 获取当前时间
                long start = System.currentTimeMillis();
                // 获取连接，在这里需要获取令牌才能继续往下走
                ConnectBean connectBean = semaphorePool.getConnect();
                connectBean.doSomeThing();
                // 释放连接，归还令牌
                semaphorePool.releaseConnect(connectBean);
                System.out.println("Thread:" + Thread.currentThread().getName() + "归还连接，总计持有时间： " + (System.currentTimeMillis() - start) + "ms");
            }
        }
    
        public static void main(String[] args) {
            for (int i = 0; i < 100; i++) {
                new TokenThread().start();
            }
        }
    }
    
    class ConnectBean {
        public ConnectBean() {
            System.out.println("ConnectBean 生成完成");
        }
    
        public void doSomeThing() {
            try {
                Random r = new Random();
                Thread.sleep(2000 + r.nextInt(100));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    
    class SemaphorePool {
        private static LinkedList<ConnectBean> list = new LinkedList<>();
    
        private static final int POOL_SIZE = 10;
    
        /**
         * 可用的连接数
         */
        private static Semaphore canUse;
    
        public SemaphorePool() {
            canUse = new Semaphore(POOL_SIZE);
        }
    
        /**
         * 初始化线程池
         */
        static {
            for (int i = 0; i < POOL_SIZE; i++) {
                list.add(new ConnectBean());
            }
        }
    
        public ConnectBean getConnect() {
            ConnectBean connectBean = null;
    
            try {
                // 获取许可证
                canUse.acquire();
    
                // 安全取出第一个连接
                synchronized (list) {
                    connectBean = list.removeFirst();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return connectBean;
        }
    
        public void releaseConnect(ConnectBean connectBean) {
            if (connectBean != null) {
                System.out.println("当前有" + canUse.getQueueLength() + "个线程在排队   "
                        + "可用连接数:" + canUse.availablePermits());
    
                synchronized (list) {
                    // 重新添加回队列中
                    list.addLast(connectBean);
                }
                // 释放令牌
                canUse.release();
            }
        }
    }

## 2.4 Exchange 数据交换器 ##

● **概念**

通过在两个线程中定义同步点，当两个线程都到达同步点时，交换数据结构

● **作用**

两个线程之间实现数据交互

● **关键方法**

<table> 
 <tbody> 
  <tr> 
   <td>方法</td> 
   <td>作用</td> 
  </tr> 
  <tr> 
   <td>exchange(V x)</td> 
   <td>交互数据，若只有一个线程到达同步点，则会等待</td> 
  </tr> 
 </tbody> 
</table>

● **示例代码**

public class ExchangerDemo {
    
        private static Exchanger<Set<String>> exchanger = new Exchanger<>();
    
        private static class ExchangerClassO extends Thread {
            private Set<String> obj;
    
            public ExchangerClassO(Set<String> set) {
                this.obj = set;
            }
    
            @Override
            public void run() {
                try {
                    exchange(obj);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    
    
        private static class ExchangerClassT extends Thread {
            private Set<String> set;
    
            public ExchangerClassT(Set<String> set) {
                this.set = set;
            }
    
            @Override
            public void run() {
                try {
                    Thread.sleep(5000);
                    exchange(set);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    
        private static void exchange(Set<String> set) throws InterruptedException {
            //交换数据，阻塞
            System.out.println("线程：" + Thread.currentThread().getName() + "交换前得值....");
            for (String s : set) {
                System.out.println(s);
            }
            exchanger.exchange(set);
            System.out.println("线程：" + Thread.currentThread().getName() + "交换后得值....");
            for (String s : set) {
                System.out.println(s);
            }
        }
    
    
        public static void main(String[] args) {
            Set<String> setA = new HashSet<>();
            Set<String> setB = new HashSet<>();
            setA.add("a1");
            setA.add("b1");
            setA.add("c1");
    
            setB.add("a2");
            setB.add("b2");
    
            new ExchangerClassO(setA).start();
            new ExchangerClassT(setB).start();
        }
    }

# 三、CountDownLatch 和 cyclicBarrier 的区别 #

<table> 
 <tbody> 
  <tr> 
   <td style="width:74px;">&nbsp;</td> 
   <td style="width:242px;"><strong>CountDownLatch</strong></td> 
   <td style="width:588px;"><strong>cyclicBarrier</strong></td> 
  </tr> 
  <tr> 
   <td style="width:74px;"><strong>等待线程数目</strong></td> 
   <td style="width:242px;">一个线程等待，直到其他线程都完成且调用 countDown()，才继续执行</td> 
   <td style="width:588px;">所有线程都等待，直到所有线程都准备好进入 await()后，全部放行</td> 
  </tr> 
  <tr> 
   <td style="width:74px;"><strong>使用次数</strong></td> 
   <td style="width:242px;">只能使用一次</td> 
   <td style="width:588px;">可以调用 reset() 方法重置，能处理更为复杂的业务场景，比如如果计算发生错误，可以重置计数器，并让线程们重新执行一次</td> 
  </tr> 
  <tr> 
   <td style="width:74px;"><strong>拓展性</strong></td> 
   <td style="width:242px;">&nbsp;</td> 
   <td style="width:588px;">提供其他有用的方法，比如getNumberWaiting方法可以获得CyclicBarrier阻塞的线程数量。isBroken方法用来知道阻塞的线程是否被中断。如果被中断返回true，否则返回false</td> 
  </tr> 
 </tbody> 
</table>

# 四、Condition #

[https://www.cnblogs.com/gemine/p/9039012.html][https_www.cnblogs.com_gemine_p_9039012.html]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0NDE2MzMx_size_16_color_FFFFFF_t_70]: /images/20221124/ce958474a2b24fcc812d7156669f9394.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0NDE2MzMx_size_16_color_FFFFFF_t_70 1]: /images/20221124/bccfc91b5a9041b3a3dc267b75a29174.png
[https_www.cnblogs.com_gemine_p_9039012.html]: https://www.cnblogs.com/gemine/p/9039012.html