控制并发流程
1、什么是控制并发流程
在我们不控制的情况下,并发的时候多个线程是尽可能的跑,并且受线程调度器控制,这个时候实际上是不受程序员控制的;
那有的时候要求某一些任务先执行,要求一些任务等某些任务执行完毕在执行的话,这个时候就必须去控制它,依照线程调度器肯定是不靠谱的。所以控制并发流程的工具类,作用就是帮助我们程序员更容易得让线程之间合作。
让线程之间相互配合,来满足业务逻辑。
比如让线程A等待线程B执行完毕之后,在执行等合作策略。
2、CountDownLatch到计时门闩
2.1、CountDownLatch类的作用
并发流程控制的工具:
倒数门闩;
例子:购物拼团;大巴,人满发车;
流程:倒数结束之前,一直处于等待状态,直到倒计时结束了,此线程才继续工作;
2.2、类的主要方法介绍
- CountDownLatch(int count):仅有一个构造函数,参数count为需要倒数的数值。
- await():调用await()方法的线程会被挂起,它会等待直到count值为0才继续执行。
- countDown():将count值减1,直到为0时,等待的线程会被换起。
2.3、两个典型用法
用法一:一个线程等待多个线程都执行完毕,再继续自己的工作。
工厂中有质检环节,为了保证质检的可信性,有5个工人都是检查的所有人都认为通过,才通过;
public class CountDownLatchDemo1 {
public static void main(String[] args) throws InterruptedException {
CountDownLatch latch = new CountDownLatch(5);//5——构造函数中需要等待的次数
//CountDownLatch特点就是多个线程之间可以并行检测的——效率较高
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(); //减1
}
}
};
service.submit(runnable);
}
System.out.println("等待5个人检查完.....");
latch.await();
System.out.println("所有人都完成了工作,进入下一个环节。");
}
}
用法二:多个线程等待某一个线程的信号,同时开始执行。
模拟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() {
@Override
public 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(); //调用完countDown之后,在之前调用await的方法会统一开始
}
}
二者结合:模拟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("所有人到达终点,比赛结束");
}
}
注意点:
a、拓展用法:多个线程等多个线程完成执行后,再同时执行。
b、CountDownLatch是不能够重用的,如果需要重新计数,可以考虑使用CyclicBarrier
或者创建新的CountDownLatch实例。
总结:
a、两个典型用法:一等多和多等一
CountDownLatch类在创建实例的时候,需要传递倒数次数。倒数到0的时候,之前等待的线程会继续运行;
b、CountDownLatch不能回滚重置;
3、Semaphore信号量
Semaphore可以用来限制或管理数量有限的资源的使用情况。
比如对污染工厂的管理,污染物不能太多,假设污染许可证只能发3张;
信号量的作用是维护一个"许可证"的计数,线程可以"获取"许可证,那信号量剩余的许可证就减1,线程也可以"释放"一个许可证,那信号量剩余的许可证就加1,当信号量所拥有的的许可证数量为0时,那么下一个还想获取许可证的线程,就需要等待,直到有另外的线程释放了许可证。
3.1、信号量用法
信号量使用流程:
a、初始化Semaphore并指定许可证的数量
b、在执行一个任务之前,先调用acquire()或者acquireUninterruptibly()方法;
c、在任务执行结束后,调用release()来释放许可证;
主要方法介绍:
acquire();acquireUninterruptibly();release();
3.2、信号量代码演示
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(3);
} 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(3);
}
}
}
3.2、信号量的特殊用法
在获取和释放的时候,可以根据我们的定制数量的;
使用情景:
一个服务器有两个方法:method1和method2,假设method1特别消耗资源,method2不是特别消耗;
此时我们对这两个方法都进行限制,假设我们一共有5个许可证;当方法1运行的时候它会把这5个许可证都拿走,
这个时候方法2就不能同时运行了;但是有的时候,方法1没有在运行,方法2可以每次只拿一个许可证;
所以要么只有1个方法1在运行,要么最多有5个方法2在运行。
注意点:
1)、获取和释放许可证的数量必须一致,否则比如每次都获取2个但是只释放1个或者不释放,随着时间的推移,到最后许可证数量不够用,会导致程序卡死。(虽然信号量并不对是否和获取的数量做规定,但是这是我们的编程规范,否则容易出错);
2)、注意在初始化Semaphore的时候设置公平性,一般设置为true会更合理
3)、并不是必须由获取许可证的线程释放那个许可证,事实上,获取和释放许可证对线程并无要求,也许是A获取了,然后由B释放,只要逻辑合理即可;
4)、信号量的作用,除了控制临界区最多同时拥有N个线程访问外,另一个作用是可以实现"条件等待",例如线程1需要在线程2完成准备工作后才能开始工作,那么就线程1 acquire(),而线程2完成任务后release(),这样的话,相当于是轻量级的CountDownLatch。
4、Condition接口(又称条件对象)
假设线程1需要等待某一个条件的时候,它就去执行condition.await()方法,一旦执行了await()方法,线程就会进入阻塞状态;然后通常会有另外一个线程,假设是线程2,去执行对应的条件,直到这个条件达成的时候,线程2就会去执行condition.signal()方法,这时JVM就会从被阻塞的线程里找到那些等待该condition的线程,当线程1就会收到可执行信号的时候,它的线程状态就会变成Runnable可执行状态。
signalAll()和signal()区别
signalAll()会唤起所有的正在等待的线程;但signal()是公平的,只会唤起那个等待时间最长的线程。
4.1 代码演示
4.1.1 普通示例
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();
}
}
4.1.2 用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();
}
}
}
}
}
注意点:
1)、实际上,如果说Lock用来代替synchronized,那么Condition就是用来代替相对应的Object.wait/notify的,
所以在用法和性质上,几乎都一样;
2)、synchronized中调用wait()方法,就会释放Monitor锁,在这里调用await方法会自动释放持有的Lock锁,和Object.wait一样,不需要自己手动释放锁;
3)、调用await的时候,必须持有锁,否则会抛出异常,和Object.wait一样;
5、CyclicBarrier循环栅栏
CyclicBarrier循环栅栏和CountDownLatch很类似,都能阻塞一组线程;
当有大量线程相互配合,分别计算不同任务,并且需要最后统一汇总的时候,我们可以使用CyclicBarrier。
CyclicBarrier可以构造一个集结点,当某一个线程执行完毕,它就会到集结点等待,直到所有线程都到了集结点,那么该栅栏就被撤销,所有线程再统一出发,继续执行剩下的任务。
5.1、可重用
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 < 5; 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();
}
}
}
}
5.2、CyclicBarrier和CountDownLatch的区别
1)、CyclicBarrier要等固定数量的线程都到了栅栏位置才能继续执行,而CountDownLatch只需等待数字到0,
也就是说,CountDownLatch用于事件,但是CyclicBarrier是用于线程的;
2)、可重用性不同:CountDownLatch在倒数到0并触发门闩打开后,就不能再次使用了,
除非新建新的实例,而CyclicBarrier可以重复使用。