并发工具CountDownLatch、CyclicBarrier、Semaphore
文章目录
- 学习链接
- CountDownLatch
- CountDownLatch类的作用
- 类的主要方法介绍
- 图解await和countDown方法
- 两个典型用法
- 注意点
- 总结
- 示例
- CountDownLatchDemo1
- CountDownLatchDemo2
- CountDownLatchDemo1And2
- CyclicBarrier
- CyclicBarrier循环栅栏
- CyclicBarrier和CountDownLatch的区别
- 示例
- CyclicBarrierDemo
- Semaphore
- Semaphore信号量
- Semaphore应用实例
- 信号量的使用流程
- 信号量主要方法介绍
- 信号量的特殊用法
- 注意点
- 示例
- SemaphoreDemo
- Condition
- Condition的作用
- signalAll和signal的区别
- Condition注意点
- 示例
- Condition1
- Condition2
学习链接
Java并发编程第9讲——CountDownLatch、CyclicBarrier和Semaphore(万字详解)
CountDownLatch
CountDownLatch类的作用
类的主要方法介绍
图解await和countDown方法
两个典型用法
注意点
总结
示例
CountDownLatchDemo1
/*** 描述:工厂中,质检,5个工人检查,所有人都认为通过,才通过*/
public class CountDownLatchDemo1 {public static void main(String[] args) throws InterruptedException {CountDownLatch latch = new CountDownLatch(5);ExecutorService service = Executors.newFixedThreadPool(5);for (int i = 0; i < 5; i++) {final int no = i + 1;Runnable runnable = new Runnable() {@Overridepublic void run() {try {Thread.sleep((long) (Math.random() * 10000));System.out.println("No." + no + "完成了检查。");} catch (InterruptedException e) {e.printStackTrace();} finally {latch.countDown();}}};service.submit(runnable);}System.out.println("等待5个人检查完.....");latch.await(); // 这里也可以设置超时等待时间, 如: latch.await(5, TimeUnit.SECONDS);System.out.println("所有人都完成了工作,进入下一个环节。");}
}
CountDownLatchDemo2
/*** 描述:模拟100米跑步,5名选手都准备好了,只等裁判员一声令下,所有人同时开始跑步。*/
public class CountDownLatchDemo2 {public static void main(String[] args) throws InterruptedException {CountDownLatch begin = new CountDownLatch(1);ExecutorService service = Executors.newFixedThreadPool(5);for (int i = 0; i < 5; i++) {final int no = i + 1;Runnable runnable = new Runnable() {@Overridepublic void run() {System.out.println("No." + no + "准备完毕,等待发令枪");try {begin.await();System.out.println("No." + no + "开始跑步了");} catch (InterruptedException e) {e.printStackTrace();}}};service.submit(runnable);}//裁判员检查发令枪...Thread.sleep(5000);System.out.println("发令枪响,比赛开始!");begin.countDown();}
}
CountDownLatchDemo1And2
/*** 描述:模拟100米跑步,5名选手都准备好了,只等裁判员一声令下,所有人同时开始跑步。当所有人都到终点后,比赛结束。*/
public class CountDownLatchDemo1And2 {public static void main(String[] args) throws InterruptedException {CountDownLatch begin = new CountDownLatch(1);CountDownLatch end = new CountDownLatch(5);ExecutorService service = Executors.newFixedThreadPool(5);for (int i = 0; i < 5; i++) {final int no = i + 1;Runnable runnable = new Runnable() {@Overridepublic void run() {System.out.println("No." + no + "准备完毕,等待发令枪");try {begin.await();System.out.println("No." + no + "开始跑步了");Thread.sleep((long) (Math.random() * 10000));System.out.println("No." + no + "跑到终点了");} catch (InterruptedException e) {e.printStackTrace();} finally {end.countDown();}}};service.submit(runnable);}//裁判员检查发令枪...Thread.sleep(5000);System.out.println("发令枪响,比赛开始!");begin.countDown();end.await();System.out.println("所有人到达终点,比赛结束");}
}
CyclicBarrier
CyclicBarrier循环栅栏
CyclicBarrier和CountDownLatch的区别
示例
CyclicBarrierDemo
/*** 描述: 演示CyclicBarrier*/
public class CyclicBarrierDemo {public static void main(String[] args) {// 5个线程都调用await时,才执行Runnable,然后5个线程接着执行CyclicBarrier cyclicBarrier = new CyclicBarrier(5, new Runnable() {@Overridepublic void run() {System.out.println("所有人都到场了, 大家统一出发!");}});// 这里循环10次,目的是验证CyclicBarrier可重用for (int i = 0; i < 10; i++) {new Thread(new Task(i, cyclicBarrier)).start();}}static class Task implements Runnable {private int id;private CyclicBarrier cyclicBarrier;public Task(int id, CyclicBarrier cyclicBarrier) {this.id = id;this.cyclicBarrier = cyclicBarrier;}@Overridepublic void run() {System.out.println("线程" + id + "现在前往集合地点");try {Thread.sleep((long) (Math.random() * 10000));System.out.println("线程" + id + "到了集合地点,开始等待其他人到达");cyclicBarrier.await();System.out.println("线程" + id + "出发了");} catch (InterruptedException e) {e.printStackTrace();} catch (BrokenBarrierException e) {e.printStackTrace();}}}
}
Semaphore
Semaphore信号量
Semaphore应用实例
信号量的使用流程
信号量主要方法介绍
信号量的特殊用法
注意点
示例
SemaphoreDemo
/*** 描述: 演示Semaphore用法*/
public class SemaphoreDemo {static Semaphore semaphore = new Semaphore(3, true);public static void main(String[] args) {ExecutorService service = Executors.newFixedThreadPool(50);for (int i = 0; i < 100; i++) {service.submit(new Task());}service.shutdown();}static class Task implements Runnable {@Overridepublic void run() {try {semaphore.acquire(1);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(Thread.currentThread().getName() + "拿到了许可证");try {Thread.sleep(2000);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(Thread.currentThread().getName() + "释放了许可证");semaphore.release(1);}}
}
Condition
Condition的作用
signalAll和signal的区别
Condition注意点
示例
Condition1
/*** 描述:演示Condition的基本用法*/
public class ConditionDemo1 {private ReentrantLock lock = new ReentrantLock();private Condition condition = lock.newCondition();void method1() throws InterruptedException {lock.lock();try{System.out.println("条件不满足,开始await");condition.await(); // 可设置超时时间进入条件等待: condition.await(1, TimeUnit.SECONDS);System.out.println("条件满足了,开始执行后续的任务");}finally {lock.unlock();}}void method2() {lock.lock();try{System.out.println("准备工作完成,唤醒其他的线程");condition.signal();}finally {lock.unlock();}}public static void main(String[] args) throws InterruptedException {ConditionDemo1 conditionDemo1 = new ConditionDemo1();new Thread(new Runnable() {@Overridepublic void run() {try {Thread.sleep(1000);conditionDemo1.method2();} catch (InterruptedException e) {e.printStackTrace();}}}).start();conditionDemo1.method1();}
}
Condition2
/*** 描述: 演示用Condition实现生产者消费者模式*/
public class ConditionDemo2 {private int queueSize = 10;private PriorityQueue<Integer> queue = new PriorityQueue<Integer>(queueSize);private Lock lock = new ReentrantLock();// 当队列不满足 不满的条件时,生产者进入等待// (当队列不满时,唤醒在此条件中等待的线程)private Condition notFull = lock.newCondition();// 当队列不满足 不空的条件时,消费者进入等待// (当队列不空时,唤醒在此条件中等待的线程)private Condition notEmpty = lock.newCondition();public static void main(String[] args) {ConditionDemo2 conditionDemo2 = new ConditionDemo2();Producer producer = conditionDemo2.new Producer();Consumer consumer = conditionDemo2.new Consumer();producer.start();consumer.start();}class Consumer extends Thread {@Overridepublic void run() {consume();}private void consume() {while (true) {lock.lock();try {// 使用while, 防止了虚假唤醒while (queue.size() == 0) {System.out.println("队列空,等待数据");try {// 当队列为空时,进入等待// 当队列不为空时,被唤醒notEmpty.await(); // 进入条件队列中等待,并释放锁;此处被唤醒时,仍然需要获取到锁,才能往下运行} catch (InterruptedException e) {e.printStackTrace();}}// 走到这里说明队列肯定不为空了// (因为先持有了锁, 所以保证了操作的原子性。其它此时尝试从该队列中获取数据的线程将进入锁等待环节)queue.poll();// 因为拿到了锁, 并且从队列中获取了数据, 队列此时肯定不为满了, 所以可以唤醒生产者线程去生产notFull.signalAll();System.out.println("从队列里取走了一个数据,队列剩余" + queue.size() + "个元素");} finally {lock.unlock();}}}}class Producer extends Thread {@Overridepublic void run() {produce();}private void produce() {while (true) {lock.lock();try {// 使用while, 防止了虚假唤醒while (queue.size() == queueSize) {System.out.println("队列满,等待有空余");try {// 当队列为满时,进入等待// 当队列为不满时,被唤醒notFull.await(); // 进入条件队列中等待,并释放锁;此处被唤醒时,仍然需要获取到锁,才能往下运行} catch (InterruptedException e) {e.printStackTrace();}}// 走到这里说明队列肯定不为满了// (因为先持有了锁, 所以保证了操作的原子性。其它此时尝试往该队列中添加数据的线程将进入锁等待环节)queue.offer(1);// 因为拿到了锁, 并且往该队列中添加了数据, 队列此时肯定不为空了, 所以可以唤醒消费者者线程去消费notEmpty.signalAll();System.out.println("向队列插入了一个元素,队列剩余空间" + (queueSize - queue.size()));} finally {lock.unlock();}}}}}