AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

【高并发】浅谈AQS中的CountDownLatch、Semaphore与CyclicBarrier

2021-09-30 231
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
全局流量管理 GTM，标准版 1个月
公共DNS（含HTTPDNS解析），每月1000万次HTTP解析
云解析 DNS，旗舰版 1个月
简介： 今天，我们来一起聊聊浅谈AQS中的CountDownLatch、Semaphore与CyclicBarrier这几个重要的类，好啦，不多说啦，咱们直接进入主题吧。

大家好，我是冰河~~

今天，我们来一起聊聊浅谈AQS中的CountDownLatch、Semaphore与CyclicBarrier这几个重要的类，好啦，不多说啦，咱们直接进入主题吧。

CountDownLatch

概述

同步辅助类，通过它可以阻塞当前线程。也就是说，能够实现一个线程或者多个线程一直等待，直到其他线程执行的操作完成。使用一个给定的计数器进行初始化，该计数器的操作是原子操作，即同时只能有一个线程操作该计数器。

调用该类await()方法的线程会一直阻塞，直到其他线程调用该类的countDown()方法，使当前计数器的值变为0为止。每次调用该类的countDown()方法，当前计数器的值就会减1。当计数器的值减为0的时候，所有因调用await()方法而处于等待状态的线程就会继续往下执行。这种操作只能出现一次，因为该类中的计数器不能被重置。如果需要一个可以重置计数次数的版本，可以考虑使用CyclicBarrier类。

CountDownLatch支持给定时间的等待，超过一定的时间不再等待，使用时只需要在await()方法中传入需要等待的时间即可。此时，await()方法的方法签名如下：

public boolean await(long timeout, TimeUnit unit)

使用场景

在某些业务场景中，程序执行需要等待某个条件完成后才能继续执行后续的操作。典型的应用为并行计算：当某个处理的运算量很大时，可以将该运算任务拆分成多个子任务，等待所有的子任务都完成之后，父任务再拿到所有子任务的运算结果进行汇总。

代码示例

调用ExecutorService类的shutdown()方法，并不会第一时间内把所有线程全部都销毁掉，而是让当前已有的线程全部执行完，之后，再把线程池销毁掉。

示例代码如下：

package io.binghe.concurrency.example.aqs;
 
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@Slf4j
public class CountDownLatchExample {
    private static final int threadCount = 200;
 
    public static void main(String[] args) throws InterruptedException {
 
        ExecutorService exec = Executors.newCachedThreadPool();
        final CountDownLatch countDownLatch = new CountDownLatch(threadCount);
        for (int i = 0; i < threadCount; i++){
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    test(threadNum);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    countDownLatch.countDown();
                }
            });
        }
        countDownLatch.await();
        log.info("finish");
        exec.shutdown();
    }
 
    private static void test(int threadNum) throws InterruptedException {
        Thread.sleep(100);
        log.info("{}", threadNum);
        Thread.sleep(100);
    }
}

支持给定时间等待的示例代码如下：

package io.binghe.concurrency.example.aqs;
 
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@Slf4j
public class CountDownLatchExample {
    private static final int threadCount = 200;
 
    public static void main(String[] args) throws InterruptedException {
        ExecutorService exec = Executors.newCachedThreadPool();
        final CountDownLatch countDownLatch = new CountDownLatch(threadCount);
        for (int i = 0; i < threadCount; i++){
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    test(threadNum);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    countDownLatch.countDown();
                }
            });
        }
        countDownLatch.await(10, TimeUnit.MICROSECONDS);
        log.info("finish");
        exec.shutdown();
    }
 
    private static void test(int threadNum) throws InterruptedException {
        Thread.sleep(100);
        log.info("{}", threadNum);
    }
}

Semaphore

概述

控制同一时间并发线程的数目。能够完成对于信号量的控制，可以控制某个资源可被同时访问的个数。

提供了两个核心方法——acquire()方法和release()方法。acquire()方法表示获取一个许可，如果没有则等待，release()方法则是在操作完成后释放对应的许可。Semaphore维护了当前访问的个数，通过提供同步机制来控制同时访问的个数。Semaphore可以实现有限大小的链表。

使用场景

Semaphore常用于仅能提供有限访问的资源，比如：数据库连接数。

代码示例

每次获取并释放一个许可，示例代码如下：

package io.binghe.concurrency.example.aqs;
 
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
@Slf4j
public class SemaphoreExample {
    private static final int threadCount = 200;
 
    public static void main(String[] args) throws InterruptedException {
 
        ExecutorService exec = Executors.newCachedThreadPool();
        final Semaphore semaphore  = new Semaphore(3);
 
        for (int i = 0; i < threadCount; i++){
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    semaphore.acquire();  //获取一个许可
                    test(threadNum);
                    semaphore.release();  //释放一个许可
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        exec.shutdown();
    }
 
    private static void test(int threadNum) throws InterruptedException {
        log.info("{}", threadNum);
        Thread.sleep(1000);
    }
}

每次获取并释放多个许可，示例代码如下：

package io.binghe.concurrency.example.aqs;
 
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
@Slf4j
public class SemaphoreExample {
    private static final int threadCount = 200;
 
    public static void main(String[] args) throws InterruptedException {
 
        ExecutorService exec = Executors.newCachedThreadPool();
        final Semaphore semaphore  = new Semaphore(3);
 
        for (int i = 0; i < threadCount; i++){
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    semaphore.acquire(3);  //获取多个许可
                    test(threadNum);
                    semaphore.release(3);  //释放多个许可
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        log.info("finish");
        exec.shutdown();
    }
 
    private static void test(int threadNum) throws InterruptedException {
        log.info("{}", threadNum);
        Thread.sleep(1000);
    }
}

假设有这样一个场景，并发太高了，即使使用Semaphore进行控制，处理起来也比较棘手。假设系统当前允许的最高并发数是3，超过3后就需要丢弃，使用Semaphore也能实现这样的场景，示例代码如下：

package io.binghe.concurrency.example.aqs;
 
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
@Slf4j
public class SemaphoreExample {
    private static final int threadCount = 200;
 
    public static void main(String[] args) throws InterruptedException {
 
        ExecutorService exec = Executors.newCachedThreadPool();
        final Semaphore semaphore  = new Semaphore(3);
 
        for (int i = 0; i < threadCount; i++){
            final int threadNum = i;
            exec.execute(() -> {
                try {
                //尝试获取一个许可，也可以尝试获取多个许可，
                    //支持尝试获取许可超时设置，超时后不再等待后续线程的执行
                    //具体可以参见Semaphore的源码
                    if (semaphore.tryAcquire()) { 
                        test(threadNum);
                        semaphore.release();  //释放一个许可
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        log.info("finish");
        exec.shutdown();
    }
    private static void test(int threadNum) throws InterruptedException {
        log.info("{}", threadNum);
        Thread.sleep(1000);
    }
}

CyclicBarrier

概述

是一个同步辅助类，允许一组线程相互等待，直到到达某个公共的屏障点，通过它可以完成多个线程之间相互等待，只有当每个线程都准备就绪后，才能各自继续往下执行后面的操作。

与CountDownLatch有相似的地方，都是使用计数器实现，当某个线程调用了CyclicBarrier的await()方法后，该线程就进入了等待状态，而且计数器执行加1操作，当计数器的值达到了设置的初始值，调用await()方法进入等待状态的线程会被唤醒，继续执行各自后续的操作。CyclicBarrier在释放等待线程后可以重用，所以，CyclicBarrier又被称为循环屏障。

使用场景

可以用于多线程计算数据，最后合并计算结果的场景

CyclicBarrier与CountDownLatch的区别

  • CountDownLatch的计数器只能使用一次，而CyclicBarrier的计数器可以使用reset()方法进行重置，并且可以循环使用
  • CountDownLatch主要实现1个或n个线程需要等待其他线程完成某项操作之后，才能继续往下执行，描述的是1个或n个线程等待其他线程的关系。而CyclicBarrier主要实现了多个线程之间相互等待，直到所有的线程都满足了条件之后，才能继续执行后续的操作，描述的是各个线程内部相互等待的关系。
  • CyclicBarrier能够处理更复杂的场景，如果计算发生错误，可以重置计数器让线程重新执行一次。
  • CyclicBarrier中提供了很多有用的方法，比如：可以通过getNumberWaiting()方法获取阻塞的线程数量，通过isBroken()方法判断阻塞的线程是否被中断。

代码示例

示例代码如下。

package io.binghe.concurrency.example.aqs;
 
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@Slf4j
public class CyclicBarrierExample {
 
    private static CyclicBarrier cyclicBarrier = new CyclicBarrier(5);
 
    public static void main(String[] args) throws Exception {
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < 10; i++){
            final int threadNum = i;
            Thread.sleep(1000);
            executorService.execute(() -> {
                try {
                    race(threadNum);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
        executorService.shutdown();
    }
    private static void race(int threadNum) throws Exception{
        Thread.sleep(1000);
        log.info("{} is ready", threadNum);
        cyclicBarrier.await();
        log.info("{} continue", threadNum);
    }
}

设置等待超时示例代码如下：

package io.binghe.concurrency.example.aqs;
 
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.*;
@Slf4j
public class CyclicBarrierExample {
 
    private static CyclicBarrier cyclicBarrier = new CyclicBarrier(5);
 
    public static void main(String[] args) throws Exception {
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < 10; i++){
            final int threadNum = i;
            Thread.sleep(1000);
            executorService.execute(() -> {
                try {
                    race(threadNum);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
        executorService.shutdown();
    }
    private static void race(int threadNum) throws Exception{
        Thread.sleep(1000);
        log.info("{} is ready", threadNum);
        try{
            cyclicBarrier.await(2000, TimeUnit.MILLISECONDS);
        }catch (BrokenBarrierException | TimeoutException e){
            log.warn("BarrierException", e);
        }
        log.info("{} continue", threadNum);
    }
}

在声明CyclicBarrier的时候，还可以指定一个Runnable，当线程达到屏障的时候，可以优先执行Runnable中的方法。
示例代码如下：

package io.binghe.concurrency.example.aqs;
 
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@Slf4j
public class CyclicBarrierExample {
 
    private static CyclicBarrier cyclicBarrier = new CyclicBarrier(5, () -> {
        log.info("callback is running");
    });
 
    public static void main(String[] args) throws Exception {
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < 10; i++){
            final int threadNum = i;
            Thread.sleep(1000);
            executorService.execute(() -> {
                try {
                    race(threadNum);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
        executorService.shutdown();
    }
    private static void race(int threadNum) throws Exception{
        Thread.sleep(1000);
        log.info("{} is ready", threadNum);
        cyclicBarrier.await();
        log.info("{} continue", threadNum);
    }
}

推荐阅读

《实践出真知：全网最强秒杀系统架构解密，不是所有的秒杀都是秒杀！！》

《注意：线程的执行顺序与你想的并不一样！！》

《深入解析Callable接口 》

《两种异步模型与深度解析Future接口！》

《解密SimpleDateFormat类的线程安全问题和六种解决方案！》

《不得不说的线程池与ThreadPoolExecutor类浅析》

《深度解析线程池中那些重要的顶层接口和抽象类》

《从源码角度分析创建线程池究竟有哪些方式》

《通过源码深度解析ThreadPoolExecutor类是如何保证线程池正确运行的》

《通过ThreadPoolExecutor类的源码深度解析线程池执行任务的核心流程》

《通过源码深度分析线程池中Worker线程的执行流程》

《从源码角度深度解析线程池是如何实现优雅退出的》

《ScheduledThreadPoolExecutor与Timer的区别和简单示例》

《深度解析ScheduledThreadPoolExecutor类的源代码》

《由InterruptedException异常引发的思考》

好了，今天就到这儿吧，我是冰河，我们下期见~~

文章标签：
云解析DNS
Java
并行计算
数据库
关键词：
高并发cyclicbarrier
高并发countdownlatch semaphore
高并发aqs countdownlatch
冰河001
目录
相关文章
杰哥在此
|
4月前
|
设计模式 安全 Java
Java面试题：如何实现一个线程安全的单例模式，并确保其在高并发环境下的内存管理效率？如何使用CyclicBarrier来实现一个多阶段的数据处理任务，确保所有阶段的数据一致性？
Java面试题：如何实现一个线程安全的单例模式，并确保其在高并发环境下的内存管理效率？如何使用CyclicBarrier来实现一个多阶段的数据处理任务，确保所有阶段的数据一致性？
杰哥在此
57 0
xindoo
|
安全 Java
Java高并发之CyclicBarrier简介
CyclicBarrier 是 Java 中的一种同步工具，它可以让多个线程在一个屏障点处等待，直到所有线程都到达该点后，才能继续执行。CyclicBarrier 可以用于协调多个线程的执行，以便它们可以在某个点上同步执行。
xindoo
135 0
张起灵-小哥
|
Java
Java——多线程高并发系列之JUC三大辅助类（CountDownLatch、CyclicBarrier、Semaphore）
Java——多线程高并发系列之JUC三大辅助类（CountDownLatch、CyclicBarrier、Semaphore）
张起灵-小哥
145 0
Java——多线程高并发系列之JUC三大辅助类（CountDownLatch、CyclicBarrier、Semaphore）
码农小达人
|
并行计算 Java 数据库
高并发之——浅谈AQS中的CountDownLatch、Semaphore与CyclicBarrier
同步辅助类，通过它可以阻塞当前线程。也就是说，能够实现一个线程或者多个线程一直等待，直到其他线程执行的操作完成。使用一个给定的计数器进行初始化，该计数器的操作是原子操作，即同时只能有一个线程操作该计数器。
码农小达人
116 0
曦阳x
|
Java
java高并发之CountDownLatch，CyclicBarrier和join
晚上打车回家，在车上看到一篇文章《22岁大学生获谷歌天价Offer，年薪千万！》，讲的是印度一个22岁大学生多次参加ACM大赛，开源多个项目，以非常牛逼的履历通过了谷歌的AI测试，斩获谷歌仅有的50个顶尖offer之一。
曦阳x
1817 0
老板这功能得加钱
|
6月前
|
消息中间件 Java Linux
2024年最全BATJ真题突击：Java基础+JVM+分布式高并发+网络编程+Linux(1)，2024年最新意外的惊喜
2024年最全BATJ真题突击：Java基础+JVM+分布式高并发+网络编程+Linux(1)，2024年最新意外的惊喜
老板这功能得加钱
68 0
Jack_hrx
|
5月前
|
缓存 NoSQL Java
Java高并发实战：利用线程池和Redis实现高效数据入库
Java高并发实战：利用线程池和Redis实现高效数据入库
Jack_hrx
479 0
龙大吉
|
3月前
|
监控 算法 Java
企业应用面临高并发等挑战，优化Java后台系统性能至关重要
随着互联网技术的发展，企业应用面临高并发等挑战，优化Java后台系统性能至关重要。本文提供三大技巧：1）优化JVM，如选用合适版本（如OpenJDK 11）、调整参数（如使用G1垃圾收集器）及监控性能；2）优化代码与算法，减少对象创建、合理使用集合及采用高效算法（如快速排序）；3）数据库优化，包括索引、查询及分页策略改进，全面提升系统效能。
龙大吉
46 0
东方睿赢
|
5月前
|
存储 NoSQL Java
探索Java分布式锁：在高并发环境下的同步访问实现与优化
【6月更文挑战第30天】Java分布式锁在高并发下确保数据一致性，通过Redis的SETNX、ZooKeeper的临时节点、数据库操作等方式实现。优化策略包括锁超时重试、续期、公平性及性能提升，关键在于平衡同步与效率，适应大规模分布式系统的需求。
东方睿赢
165 1
萝卜丝丸子
|
4月前
|
算法 Java 调度
高并发架构设计三大利器：缓存、限流和降级问题之使用Java代码实现令牌桶算法问题如何解决
高并发架构设计三大利器：缓存、限流和降级问题之使用Java代码实现令牌桶算法问题如何解决
萝卜丝丸子
46 0

热门文章

最新文章

  • 1
    高并发架构系列：如何从0到1设计一个类Dubbo的RPC框架
  • 2
    不了解 QPS、TPS、RT、并发数、吞吐量，劝你简历别写熟悉高并发
  • 3
    MySQL 5.7在高并发下性能劣化问题的详细剖析
  • 4
    从服务端视角看高并发难题
  • 5
    高并发应用场景下的负载均衡与故障转移实践，AgileEAS.NET SOA 负载均衡介绍与实践
  • 6
    究竟啥才是互联网架构“高并发”
  • 7
    Flask 高并发部署方案详细教程！
  • 8
    Serverless时代Rust即将称王，详解C、Java、Go、Rust高并发背后的技术？
  • 9
    Redis 的高并发实战：抢购系统 --浅奕
  • 10
    PostgreSQL 高并发应用数据库设计与优化实践 PPT
  • 1
    TiDB适用场景解析：海量数据存储与高并发读写的利器
    878
  • 2
    优惠券秒杀案例 - CAS、Redis+Lua脚本解决高并发并行
    303
  • 3
    【三十】springboot项目上高并发解决示例
    292
  • 4
    一文让你深入了解 Java-Netty高性能高并发
    466
  • 5
    高并发的场景下，不能不说的限流算法
    88
  • 6
    【Spring云原生】Spring Batch：海量数据高并发任务处理！数据处理纵享新丝滑！事务管理机制+并行处理+实例应用讲解
    233
  • 7
    【Redis项目实战】使用Springcloud整合Redis分布式锁+RabbitMQ技术实现高并发预约管理处理系统
    171
  • 8
    【Spring boot实战】Springboot＋对话ai模型整体框架+高并发线程机制处理优化+提示词工程效果展示（按照框架自己修改可对接市面上百分之99的模型）
    648
  • 9
    阿里巴巴内部高并发Nginx性能调优实战解密
    137
  • 10
    面试官：你会如何设计一个高并发系统的架构设计？
    185

    • 相关电子书

    更多
  • 徐雷-Java为王，互联网高并发架构设计与选型之路6.0
  • Redis 的高并发实战：抢购系统
  • MySQL高并发场景实战

    • 相关实验场景

    更多
  • RDS MySQL高并发访问下的系统稳定性
    • 下一篇
    无影云桌面