什么是控制并发流程？
- 控制并发流程的工具类，作用就是帮助我们程序员更容易得让线程之间合作
- 让线程之间相互配合，来满足业务逻辑
- 比如让线程A等待线程B执行完毕后再执行等合作策略
常见的控制并发流程的工具类
CountDownLatch类的作用
倒数门闩
例子：购物拼团；大巴（游乐园坐过山车排队），人满发车。
流程：倒数结束之前，一直处于等待状态，直到倒计时结束了，此线程才继续工作。
开始 ——> 进入等待 ——> 倒数结束 ——> 继续工作
类的主要方法介绍
- CountDownLatch(int count)：仅有一个构造函数，参数count为需要倒数的数值。
- await()：调用await()方法的线程会被挂起，它会等待直到count值为0才继续执行。
- countDown()：将count值减1，直到为0时，等待的线程会被唤起。
两个典型用法
用法一：一个线程等待多个线程都执行完毕，再继续自己的工作。
示例代码：
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* 描述： 工厂中，质检，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() {
@Override
public 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();
System.out.println("所有人都完成了工作，进入下一个环节。");
}
}
执行结果：
五个线程并行执行完后主线程才继续执行。
用法二：多个线程等待某一个线程的化司时开始执行。
情景：模拟100米跑步，5名选手都准备好了，只等裁判员一声令下，所有人同时开始跑步。当所有人都到终点后，比赛结束。
示例代码
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* 描述： 模拟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() {
@Override
public 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("所有人到达终点，比赛结束");
}
}
运行结果：
通过模拟100米跑步案例，很好的体现出CountDownLatch控制并发线程流程的能力。
注意点
- 扩展用法：多个线程等多个线程完成执行后，再同时执行
- CountDownLatch是不能够重用的，如果需要重新计数，可以考虑使用CyclicBarrier或者创建新的CountDownLatch实例。
Semaphore信号量
Semaphore可以用来限制或管理数量有限的资源的使用情况。
信号量使用流程
- 初始化Semaphore并指定许可证的数量
- 在需要被现在的代码前加acquire() 或者acquireUninterruptibly()方法
- 在任务执行结束后，调用release(来释放许可证)
信号量主要方法介绍
- new Semaphore(int permits, boolean fair)：这里可以设置是否要使用公平策略，如果传入true，那么Semaphore会把 之前等待的线程放到FIFO的队列里，以便于当有了新的许可证，可以分发给之前等了最长时间的线程。
- tryAcquire()：看看现在有没有空闲的许可证，如果有的话就获取；如果没有的话也没关系，我不必陷入阻塞，我可以去做别的事，过一会再来查看许可证的空闲情况。
- tryAcquire(timeout)：和tryAcquire()一样，但是多了一个超时时间，比如“在3秒内获取不到许可证，我就去做别的事“。
代码案例
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
/**
* 描述： 演示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 {
@Override
public void run() {
try {
semaphore.acquire();
} 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();
}
}
}
运行结果
信号量特殊用法
一次性获取或释放多个许可证：semaphore.acquire(3)
比如TaskA会调用很消耗资源的method1()，而TaskB调用的是不太消耗资源的method2()，假设我们一共有5个许可证。那么我们就可以要求TaskA获取5个许可证才能执行，而TaskB只需要获取到一个许可证就能执行，这样就避免了A和B同时运行的情况，我们可以根据自己的需求合理分配资源。
注意点
1.获取和释放的许可证数量必须一致，否则比如每次都获取2个但是只释放1个甚至不释放，随着时间的推移，到最后许可证数量不够用，会导致程序卡死。（虽然信号量类并不对是否和获取的数量做规定，但是这是我们的编程规范，否则容易出错）
2.注意在初始化Semaphore的时候设置公平性，一般设置为true会更合理
3.并不是必须由获取许可证的线程释放那个许可证，事实上，获和释放许可证对线程并无要求，也许是A获取了，然后由B释放，只要逻辑合理即可。
4.信号量的作用，除了控制临界区最多同时有N个线程访问外，另一个作用是可以实现"条件等待〞，例如线程1需要在线程2完成准备工作后才能开始工作，那么就线程lacquire()，而线程2完成任务后release()，这样的话，相当于是轻量级的CountDownLatch。
Condition接口（又称条件对象）
Condition作用
- 当线程1需要等待某个条件的时候，它就去执行condition.await()方法，一旦执行了await()方法，线程就会进入阻塞状态
- 然后通常会有另外一个线程，假设是线程2，去执行对应的条件，直到这个条件达成的时候，线程2就会去执行condition.signal()方法，这时JVM就会从被阻塞的线程里找，找到那些等待该condition的线程，当线程1就会收到可执行信号的时候，它的线程状态就会变成Runnable可执行状态
signalAll()和signal()区别
signalAll()会唤起所有的正在等待的线程
但是signal()是公平的，只会唤起那个等待时间最长的线程
Condition接口是和锁绑在一起的，创建Condition的方法：
private ReentrantLock lock = new ReentrantLock();
private Condition condition = lock.newCondition();
代码案例
代码案例一：演示Condition的基本用法
package flowcontrol.condition;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
/**
* 描述： 演示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();
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() {
@Override
public void run() {
try {
Thread.sleep(1000);
conditionDemo1.method2();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
conditionDemo1.method1();
}
}
运行结果：
代码案例二：用Condition实现生产者消费者模式
package flowcontrol.condition;
import java.util.PriorityQueue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
* 描述： 演示用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 {
@Override
public void run() {
consume();
}
private void consume() {
while (true) {
lock.lock();
try {
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 {
@Override
public void run() {
produce();
}
private void produce() {
while (true) {
lock.lock();
try {
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();
}
}
}
}
}
运行结果：
Condition注意点
- 实际上，如果说Lock用来代替synchronized，那么Condition就是用来代替相对应的Object.wait/notify的，所以在用法和性质上，几乎都一样
- await方法会自动释放持有的Lock锁，和Object.wait一样，不
需要自己手动先释放锁 - 调用await的时候，必须持有锁，否则会抛出异常，和Object.wait一样。
Condition的await/notify和锁的Object.wait/notify区别
- 多条件变量支持（关键优势）
一个锁可以创建多个Condition，使控制锁的阻塞和唤醒更灵活，Object.wait/notify只能锁本身进行阻塞和唤醒。
- 使用 Object.wait()/notify()：所有线程（生产者和消费者）都在同一个对象的等待队列中。当缓冲区有空位时，notify() 可能唤醒一个消费者线程，而该线程发现缓冲区仍空，会再次等待，造成效率低下。notifyAll() 会唤醒所有线程，引发“惊群效应”。
- 使用 Condition.await()/signal()：可以创建两个 Condition 对象，例如 notFull（缓冲区未满时生产者等待）和 notEmpty（缓冲区非空时消费者等待）。生产者在 notFull.await() 中等待，消费者在 notEmpty.await() 中等待。当生产者放入一个元素后，只需调用 notEmpty.signal()，就能精准地唤醒一个等待的消费者，而不会干扰生产者线程。
- 功能丰富性
Condition 提供了更丰富的等待方法：
awaitNanos(long nanosTimeout)：等待指定的纳秒数。 await(long time, TimeUnit unit)：等待指定的时间段。 awaitUntil(Date deadline)：等待到指定的截止日期。 awaitUninterruptibly()：等待时不响应中断。
这些方法在 Object.wait() 中均不支持。
- 异常处理与安全性
- 两者在等待时被中断都会抛出 InterruptedException。
- Condition 的优势在于其显式锁机制。虽然需要手动 unlock()，但这使得锁的管理更灵活，例如可以实现公平锁。而 synchronized 锁的获取和释放是自动的，但无法中断或设置超时。
CyclicBarrier循环栅栏
- CyclicBarrier循环栅栏和CountDownLatch很类似，都能阻塞一组线程
- 当有大量线程相互配合，分别计算不同任务，并且需要最后统一汇总的时候，我们可以使用Cyclic Barrier。CyclicBarrier可以构造一个集结点，当某一个线程执行完毕，它就会到集结点等待，直到所有线程都到了集结点，那么该栅栏就被撤销，所有线程再统一出发，继续执行剩下的任务。
生活中的例子：”咱们3个人明天中午在学校碰面，都到齐后，一起讨论下学期的计划。
代码实现：
package flowcontrol.cyclicbarrier;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
/**
* 描述： 演示CyclicBarrier
*/
public class CyclicBarrierDemo {
public static void main(String[] args) {
CyclicBarrier cyclicBarrier = new CyclicBarrier(5, new Runnable() {
@Override
public void run() {
System.out.println("所有人都到场了， 大家统一出发！");
}
});
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;
}
@Override
public 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();
}
}
}
}
运行结果：
CyclicBarrier和CountDownLatch的区别
- 作用不同：CyclicBarrier要等固定数量的线程都到达了栅栏位置才能继续执行，而CountDownLatch只需等待数字到0，也就是说，CountDownLatch用于事件，但是CyclicBarrier是用于线程的。
- 可重用性不同：CountDownLatch在倒数到0并触发后，就不能再次使用了，除非新建新的实例；而CyclicBarrier可以重复使用。
