Java多线程-CountDownLatch、Semaphone、CyclicBarrier入门

---

title: CountDownLatch、Semaphone、CyclicBarrier入门
date: 2020-05-27 11:02:00
categories: CountDownLatch,Semaphone,CyclicBarrier
description: Java多线程

• 1. 背景
• 2. CountDownLatch
  • 2.1. 构造函数
  • 2.2. 样例
  • 2.3. 实际应用
    • 2.3.1. 背景
    • 2.3.2. 解决方案
• 3. CyclicBarrier
  • 3.1. 构造函数
  • 3.2. 样例
• 4. Semaphone
  • 4.1. 构造函数
  • 4.2. 样例
• 5. 小结

1. 背景

在使用多线程过程中，

2. CountDownLatch

允许绑定一个或多个线程等待，直到在其他执行中的线程操作完成为止的一种计数器。这种计数器使用的是一种递减
直到当前计数达到零为止，await 方法将阻塞，此后等所有线程将被释放，并且随后的所有await调用将立即返回。

这是一种一次性现象-无法重置计数。如果需要用于重置计数的版本，请考虑使用 CyclicBarrier。

CountDownLatch 是一种多功能的同步工具，可以用于多种目的。以1的计数初始化的 CountDownLatch 用作简单的 on / off锁存器或gate：所有调用线程等待在gate处等待，直到被countDown的线程打开为止。初始化为N的CountDownLatch可以用于使一个线程等待，直到N个线程完成某个动作或某个动作已经完成N次。

CountDownLatch 的一个有用属性是，它不需要调用countDown的线程在继续进行操作之前就无需等待计数达到零，它只是防止任何线程经过等待状态，直到所有线程都可以通过。

用法示例：这是一对类，其中一组工作线程使用两个倒计时锁存器

• 启动信号，可防止任何工人继续前进，直到驾驶员为他们做好准备为止。
• 完成信号，允许驾驶员等到所有工人都完成为止。

2.1. 构造函数

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

CountDownLatch 构造函数只有一个，参数 count 为绑定大小，核心实现时通过一个同步控件(Sync)，维持 AQS 状态表示计数。

AbstractQueuedSynchronizer 是 队列式同步器。

通过图我们可以得知，常用 ReentranLock、Semaphore 、 CountDownLatch等，后面我们会用专门一个篇幅来详细研究下 AbstractQueuedSynchronizer。

2.2. 样例

public class UserCountDownLatch {

    public static void main(String[] args) {
        int size = 3;
        CountDownLatch countDownLatch = new CountDownLatch(size);
        ExecutorService executorService =  Executors.newFixedThreadPool(size);
        String suffix = "t_";
        for (int i = 0; i < size; i++) {
            executorService.execute(new CountDownLatchDemo(countDownLatch,suffix+i));
        }
        try {
            countDownLatch.await();
            System.out.println("执行完成");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        executorService.shutdown();
    }
}

class CountDownLatchDemo implements Runnable{

    private String exeStr;
    private CountDownLatch countDownLatch;

    public CountDownLatchDemo() {
    }

    public CountDownLatchDemo(CountDownLatch countDownLatch,String exeStr) {
        this.countDownLatch = countDownLatch;
        this.exeStr = exeStr;
    }

    @Override
    public void run() {
        try {
            int id = new Random().nextInt(5);
            TimeUnit.SECONDS.sleep(id);
            LocalDateTime localDateTime = LocalDateTime.now();
            String resutl = exeStr+"_"+id;
            System.out.println("当前时间 "+localDateTime.getMinute()+":"+localDateTime.getSecond()+" 当前线程名: "+Thread.currentThread().getName()+ " 结果为: "+ resutl);
            countDownLatch.countDown();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }
}

2.3. 实际应用

2.3.1. 背景

2.3.2. 解决方案

3. CyclicBarrier

谷歌翻译出来意思循环屏障，可以理解为一种屏障机制，让一组线程任务到达该屏障时被阻塞，直到最后一个线程到达该屏障，被阻塞的线程才能继续执行。形象点的比喻，类似田径比赛时的信号枪，当所有运动员都准备好，信号枪发出信号！

3.1. 构造函数

JDK源码中有两个构造函数，分别为：

• CyclicBarrier(int parties)：
• CyclicBarrier(int parties, Runnable barrierAction)：

其中 parties 绑定给定数量的线程或者任务在屏障被释放之前必须调用等待的线程数；参数barrierAction当屏障被绊倒时执行的命令，这是最后一个线程执行结束的时候触发的动作；如果没有动作，则返回null

3.2. 样例

定义一个CyclicBarrier 绑定三个线程，在创建一个线程池，添加三个线程进去，最后别忘了再来一个shutdown，否则线程会一直挂起。例子比较简单，有兴趣同学可以自己看看！

public class CyclicBarrierDemo {

    public static void main(String[] args) throws Exception{
        int cdl = 5;
        ExecutorService executorService =  Executors.newFixedThreadPool(cdl);
        CyclicBarrier cyclicBarrier = new CyclicBarrier(cdl,new BarrierAction());
        for (int i = 0; i < cdl; i++) {
            executorService.submit(new CyclicBarrierRunnable("source"+i,cyclicBarrier));
        }
        System.out.println("执行完成");
        executorService.shutdown();
    }
}

class BarrierAction implements Runnable {

    @Override
    public void run() {
        System.out.println(" Thread Name is: "+ Thread.currentThread().getName() + " ,BarrierAction Task End");
    }
}

public class CyclicBarrierRunnable implements Runnable{

    private String source;

    private CyclicBarrier cyclicBarrier;

    public CyclicBarrierRunnable() {
    }

    public CyclicBarrierRunnable(String source, CyclicBarrier cyclicBarrier) {
        this.source = source;
        this.cyclicBarrier = cyclicBarrier;
    }

    @Override
    public void run() {
        int id = new Random().nextInt(5);
        try {
            TimeUnit.SECONDS.sleep(id);
            LocalDateTime localDateTime = LocalDateTime.now();
            String resutl = source+"_"+id;
            System.out.println("当前时间 "+localDateTime.getMinute()+":"+localDateTime.getSecond()+" 当前线程名: "+Thread.currentThread().getName()+ " 结果为: "+ resutl);
            cyclicBarrier.await();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}

4. Semaphone

Semaphore 用于限制可以访问某些资源（物理或逻辑的）的线程数目，他维护了一个许可证集合，有多少资源需要限制就维护多少许可证集合，假如这里有 N 个资源，那就对应于 N 个许可证，同一时刻也只能有 N 个线程访问。一个线程获取许可证就调用 acquire 方法，用完了释放资源就调用 release 方法。

编写一个样例，只维护一个许可证，限制被访问，从另一方面保护了资源的原子性！

4.1. 构造函数

JDK源码中提供两个构造函数，默认为非公平：

Semaphore(int permits)
Semaphore(int permits, boolean fair)

• int permits：定义许可证
• boolean fair：是否公平

4.2. 样例

public class SemaphoreDemo {

    public static void main(String[] args) {
        int cdl = 2;
        ExecutorService executorService =  Executors.newFixedThreadPool(cdl);
        Semaphore semaphore = new Semaphore(cdl);
        for (int i = 0; i < 6; i++) {
            executorService.submit(new CredentialsRunnable("source"+i,semaphore));
        }
        System.out.println("执行完成");
        executorService.shutdown();
    }
}

public class CredentialsRunnable implements Runnable{

    private String source;
    private Semaphore semaphore;

    public CredentialsRunnable() {
    }

    public CredentialsRunnable(String source, Semaphore semaphore) {
        this.source = source;
        this.semaphore = semaphore;
    }

    @Override
    public void run() {
        try {
            System.out.println("ThreadName is "+ Thread.currentThread().getName()+" 凭据: "+source +" 等待中 准备获取许可");
            semaphore.acquire();
            System.out.println("ThreadName is "+ Thread.currentThread().getName()+" 凭据: "+source +" 获取许可");
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            System.out.println("ThreadName is "+ Thread.currentThread().getName()+" 凭据: "+source +" 许可使用完毕，准备释放");
            semaphore.release();
        }
    }
}

5. 小结