并发工具CountDownLatch、CyclicBarrier、Semaphore

学习链接

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();}}}}}