【多线程并发编程】十三 CountDownLatch和CyclicBarrier

你的名字 2023-07-10 15:40 29阅读 0赞

> 程序猿学社的GitHub，欢迎**Star**  
> [https://github.com/ITfqyd/cxyxs][https_github.com_ITfqyd_cxyxs]  
> 本文已记录到github，形成对应专题。

# 前言 #

> jdk1.5以后，增加了不少内容，我们就来看一看**CountDownLatch**和CyclicBarrier，实际上，很多的方法，如果实际开发过程中，没有用到过，我们几乎都不怎么熟悉。一切的一切，都感觉十分的陌生。

# 1.CountDownLatch #

## 概念 ##

允许一个或多个线程等待直到在其他线程执行完毕后再执行。

*  CountDownLatch用给定的计数初始化
 *  await方法阻塞
 *  直到countDown()方法的调用而导致当前计数达到零，之后所有等待线程被释放，并且任何后续的await 调用立即返回。

## 常用api接口 ##

<table> 
 <thead> 
  <tr> 
   <th>方法</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>await()</td> 
   <td>导致当前线程等到锁存器计数到零，除非线程是 interrupted 。线程处于休眠状态，直到计数为0</td> 
  </tr> 
  <tr> 
   <td>countDown()</td> 
   <td>减少锁存器的计数(调用一次，计数减去1)，如果计数达到零，释放所有等待的线程。</td> 
  </tr> 
 </tbody> 
</table>

## 应用场景 ##

现在是疫情期间，为了保证员工的安全，减少与他们接触的机会，某某公司，实行车接车送。所有人员就位后，司机需要向HR汇报，人员上车情况。

## 实战 ##

### demo案例 ###

package com.cxyxs.thread.thirteen;
    
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.TimeUnit;
    
    /** * Description：转发请注明来源 程序猿学社 - https://ithub.blog.csdn.net/ * Author: 程序猿学社 * Date: 2020/3/3 16:24 * Modified By: */
    public class Emp { 
          //初始化为3，标识有3个员工需要等待
          public static CountDownLatch countDownLatch = new CountDownLatch(3);
    
        /** * 员工正在上车中 */
        public void toAddress(){ 
           System.out.println(Thread.currentThread().getName()+"正在步行去固定的上车地点...");
           //模拟去固定地点上车的过程
            try { 
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
            countDownLatch.countDown();
            System.out.println(Thread.currentThread().getName()+"已上车");
        }
    }

**调用类**

package com.cxyxs.thread.thirteen;
    
    /** * Description：转发请注明来源 程序猿学社 - https://ithub.blog.csdn.net/ * Author: 程序猿学社 * Date: 2020/3/3 16:22 * Modified By: */
    public class Demo1 { 
        public static void main(String[] args) { 
            Emp emp = new Emp();
            //使用lambda方式实现一个线程
            new Thread(()->{ 
                emp.toAddress();
            },"张三").start();
            new Thread(()->{ 
                emp.toAddress();
            },"李四").start();
            new Thread(()->{ 
                emp.toAddress();
            },"王五").start();
    
    
            new Thread(()->{ 
                try { 
                    emp.countDownLatch.await();
                    System.out.println(Thread.currentThread().getName()+"向HR汇报上车情况!");
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            },"黄师傅").start();
        }
    }

![20200304205353964.png][]

*  初始化计数为3，调用countDown方法后，每次减去1，一直到零，就会主动释放，继续await后面的代码。
 *  在实际使用过程中，一定要记得调用一次，就调用countDown方法，不然，线程会一直堵塞在哪里。

### 源码分析 ###

#### CountDownLatch初始化 ####

public CountDownLatch(int count) { 
            if (count < 0) throw new IllegalArgumentException("count < 0");
            this.sync = new Sync(count);
    }

*  初始化一个CountDownLatch对象，需要给定计数的初始值，如果<0,则会抛出异常。
 *  创建了一个Sync对象。

/** * Synchronization control For CountDownLatch. * Uses AQS state to represent count. */
        private static final class Sync extends AbstractQueuedSynchronizer { 
            private static final long serialVersionUID = 4982264981922014374L;
    
            Sync(int count) { 
                setState(count);
            }
            protected final void setState(int newState) { 
            state = newState;
            }
        }
        private volatile int state;
        private final Sync sync;

*  首先调用Sync的构造方法。该构造方法会调用setState方法。
 *  **sync继承AbstractQueuedSynchronizer抽象类，其中有一个变量state**，就是为了同步状态的。注意，他的前面有**volatile关键字**，表示线程之间是可见的，只要值有变动，其他的项目都会知道。类似于开发过程中，某某提交了代码，但是，不说明，其他的人，只有更新代码的时候，发现起冲突才知道，改了同一个文件，而可见性，就是，某人一改代码，马上，就全网通知，说我改代码了，你们都拉取一下最新的代码。

#### countDown方法 ####

public void countDown() { 
            sync.releaseShared(1);
        }

*  调用AbstractQueuedSynchronizer方法

public final boolean releaseShared(int arg) { 
            if (tryReleaseShared(arg)) { 
                doReleaseShared();
                return true;
            }
            return false;
        }

*  调用了两个方法tryReleaseShared和doReleaseShared

protected boolean tryReleaseShared(int releases) { 
                // Decrement count; signal when transition to zero
                for (;;) { 
                    int c = getState();
                    if (c == 0)
                        return false;
                    int nextc = c-1;
                    if (compareAndSetState(c, nextc))
                        return nextc == 0;
                }
            }

*  获取state的数量，也就是计数器的值，如果为0直接返回false。表示计数已经完成。
 *  compareAndSetState方法我们一步步跟踪下去，发现调用的compareAndSwapInt被native关键字修饰，说明是调用的事c或者c++的方法。利用了CAS，这里不过多阐述，可以理解为乐观锁, 但是，可以保证该变量是原子的。代码块，就无法保证了。说到原子，我们就可以联想起多个线程操作i++线程不安全的问题。这也是一到面试经常会的题目。
 *  \-1完后，再次检查是否为0，如果为0表示计数已完成。

#### await方法 ####

public void await() throws InterruptedException { 
            sync.acquireSharedInterruptibly(1);
    }

*  实际上是调用父类的acquireSharedInterruptibly方法

public final void acquireSharedInterruptibly(int arg)
                throws InterruptedException { 
            if (Thread.interrupted())
                throw new InterruptedException();
            if (tryAcquireShared(arg) < 0)
                doAcquireSharedInterruptibly(arg);
        }

*  判断当前线程是否中断，如果中断则抛出InterruptedException异常。
 *  tryAcquireShared的返回值小于才会调用doAcquireSharedInterruptibly方法

##### tryAcquireShared代码 #####

protected int tryAcquireShared(int acquires) { 
                return (getState() == 0) ? 1 : -1;
            }

*  判断state的计算是否为0，如果为0返回1.

# 2 CyclicBarrier #

## 概念 ##

实现让所有线程全部等待彼此达到共同屏障点之后，才能同时执行。循环阻塞在涉及固定大小的线程方的程序中很有用，这些线程必须等待彼此，屏障又被称为循环 ，因为它可以在等待的线程被释放之后重新使用。

*  可以简单的理解为CyclicBarrier是一个减1操作，CountDownLatch是加一操作。

## 常用api接口 ##

<table> 
 <thead> 
  <tr> 
   <th>方法</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>CyclicBarrier(int parties)</td> 
   <td>创建一个新的 CyclicBarrier ，当给定数量的线程（线程）等待它，达到这个数量时，所有达到共同屏障昨天的线程都会继续运行</td> 
  </tr> 
  <tr> 
   <td>CyclicBarrier(int parties, Runnable barrierAction)</td> 
   <td>类似于上面这种，只是多一个一个参数，可以理解为行为，也就是说，达到固定线程数后，就会执行runnable里面的行为代码</td> 
  </tr> 
  <tr> 
   <td>await()</td> 
   <td>等待所有线程都已调用await方法</td> 
  </tr> 
 </tbody> 
</table>

*  这里列举一些常用的api接口，详情的api接口，请查看jdk的api文档

## 应用场景 ##

*  大部分的公司，每年都有一次聚会，公司会包一辆大巴车，而大巴车，需要等所有人都到达后，才能出发。

## 实战 ##

package com.cxyxs.thread.thirteen;
    
    import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CyclicBarrier;
    
    /** * Description： * Author: 程序猿学社 * Date: 2020/3/6 8:18 * Modified By: */
    public class CyclicBarrierTest { 
        public static void main(String[] args) { 
            final CyclicBarrier barrier = new CyclicBarrier(3, new Runnable() { 
                @Override
                public void run() { 
                    //达到固定屏障数量后，就会触发
                    System.out.println("人员已到期，司机开始发车!");
                }
            });
    
            for (int i = 0; i < 3; i++) { 
                final  int index =i;
                //使用lambda
                new Thread(()->{ 
                    System.out.println(Thread.currentThread().getName()+",开始上车");
                    try { 
                        barrier.await();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    } catch (BrokenBarrierException e) { 
                        e.printStackTrace();
                    }
                    System.out.println(Thread.currentThread().getName()+",上车完毕,当前时间:"+System.currentTimeMillis());
                },"员工"+i).start();
            }
        }
    }

![20200306083424530.png][]

*  模拟3个员工，他们公司旅游，大巴司机发现人齐，就发车的一个场景
 *  final关键字，注意Thread中是无法直接使用外面的i，所以引入一个中间变量，用final修饰。
 *  使用了lambda表达式，实现一个线程，让代码更简洁。
 *  发现每个员工上车后，就调用await方法，在等待其他的员工，人员一齐，司机就马上发车。
 *  后面，还有一句话，上车完毕和当前时间，实际上也就是在调用await方法后，打印的一句话，**为什么这样设计？** 实际上，就是更直观的，让我们知道，达到固定屏障数量后，所有的线程就会被唤醒，继续运行，这也是我们概念里面写的**同时执行**。

[https_github.com_ITfqyd_cxyxs]: https://github.com/ITfqyd/cxyxs
[20200304205353964.png]: https://img-blog.csdnimg.cn/20200304205353964.png
[20200306083424530.png]: https://img-blog.csdnimg.cn/20200306083424530.png