控制并发流程，做好线程间的协调

本是古典何须时尚 2024-03-16 18:33 65阅读 0赞

### 一、概述 ###

#### 1. 什么是控制并发流程？ ####

线程一般是由线程调度器自动控制的，但有些场景需要按照我们程序员的意愿去实现多线程之间相互配合，从而满足业务逻辑。比如：

*  让线程A等待线程B执行完后再执行等一些相互合作的逻辑；
 *  或一系列线程等待一个线程运行完毕或发出信号之后再执行

#### 2. 控制并发流程工具类 ####

![在这里插入图片描述][6d1651d9f1e14ced8b2a3fcd3dc83c95.jpeg]

### 二、 CountDownLatch 倒计时门栓 ###

倒数（向下计数、倒着计数）count为0后，那些执行了 await() 方法陷入阻塞的线程就被唤醒继续执行。就像去做过山车时，等到空余座位为0时，就会发车。

#### 1. 主要方法介绍 ####

*  CountDownLatch(int count)：仅有这一个构造函数，参数count为需要倒数的数值。
 *  await()： 调用 await() 方法的线程会被挂起，它会等待直到count值为0才继续执行。
 *  countDown() ：将count值减1，等到为0时，那些等待的线程会被唤起。

#### 2. 图解 ####

![5f8ed6552011da46330b85448d7a4937.png][]

*  在构造方法中指定倒数count值；
 *  调用await的线程Ta会被挂起；
 *  每调用countDown()，倒数count会减1；但是该线程不会被挂起，依然执行
 *  当倒数count值为0时，之前执行await的线程就被唤醒，开始执行

#### 3、代码演示 ####

##### （1）用法①：一等多 #####

一个线程等待多个线程都发出信号后，再继续自己的工作：

/**
     *      工厂中，质检，5个工人检查，当5个人都认为通过，才认为这个质检通过
     */
    public class CountDownLatchDemo1 {
        
        public static void main(String[] args) throws InterruptedException {
        
            CountDownLatch countDownLatch = new CountDownLatch(5);//指定倒数count为5
            ExecutorService pool = Executors.newFixedThreadPool(5);//线程池创建5个线程
    
            //5次任务质检
            for (int i = 0; i < 5; i++) {
        
                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 {
        
                            countDownLatch.countDown();//倒计count减1
                        }
                    }
                };
    
                pool.submit(runnable);
            }
            System.out.println("等待5个人质检完。。。。。");
            countDownLatch.await();//主线程等待倒时count=0，释放所有挂起线程，并主线进行工作
            System.out.println("质检完成");
    
        }
        
    }

![image-20230615103516182][]

##### (2) 用法②：多等一 #####

多个线程等待某个线程发出信号后，同时开始执行

/**
     *      模拟100m跑步，5名选手都准备好了，只能裁判员一生令下，5人同时跑出
     */
    public class CountDownLatchDemo2 {
        
        public static void main(String[] args) throws InterruptedException {
        
            CountDownLatch countDownLatch = new CountDownLatch(1);
            ExecutorService pool = Executors.newFixedThreadPool(5);//线程池创建5个线程
    
            for (int i = 0; i < 5; i++) {
        
                int no = i+1;
                Runnable runnable = new Runnable() {
        
                    @Override
                    public void run() {
        
                        System.out.println("NO." + no + ":准备完毕，等待发令");
                        try {
        
                            countDownLatch.await();
                            System.out.println("No." + no + ":开始跑步");
                        } catch (InterruptedException e) {
        
                            e.printStackTrace();
                        }
                    }
                };
                pool.submit(runnable);
            }
            //主线程，模拟裁判员
            Thread.sleep(5000);
            System.out.println("发令枪响，比赛开始");
            countDownLatch.countDown();
        }
    }

![47d64ff7c33d66faf5880f14d12934ff.png][]

##### （3）综合用法：多等一 & 一等多 #####

**多等一**：5名运动员等待裁判员打枪开跑；**一等多**：裁判员等5名运动员到达终点；

/**
     *      模拟100m跑步，5名选手都准备好了，只能裁判员一生令下，5人同时跑出，当所有人都到终点后，比赛结束
     */
    public class CountDownLatchDemo3 {
        
        public static void main(String[] args) throws InterruptedException {
        
            CountDownLatch begin = new CountDownLatch(1);
            CountDownLatch end = new CountDownLatch(5);
            ExecutorService pool = Executors.newFixedThreadPool(5);//线程池创建5个线程
    
            for (int i = 0; i < 5; i++) {
        
                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();
                        }
                    }
                };
                pool.submit(runnable);
            }
            //主线程，模拟裁判员
            Thread.sleep(5000);
            System.out.println("发令枪响，比赛开始");
            begin.countDown();
            end.await();
            System.out.println("比赛结束！");
        }
    }

![a042a158bd8464160a066c4f685fd007.png][]

#### 4、注意点 ####

*  他不能`重复使用`，当倒数count=0，该实例就失效了；
 *  如果要再次使用，需要`再实例化新的对象`
 *  可以实现`多等多`的情况

### 三、Semaphore 信号量 ###

#### 1. 作用 ####

对于一些重量级服务，如执行时间长、处理消耗资源大，设置一下同时并发执行任务的线程个数，从而保障服务平稳运行。

*  用来`限制`或管理数量的`有限资源`的使用情况
 *  类似于生活中的"许可证"，许可证数量有限，并且只有拿到“许可证”的线程才允许运行

#### 2. 图解 ####

![5c40ce3e5634133f51af7b9719ad938d.png][]

#### 3. 重要方法 ####

*  new Semaphore(int permits,boolean fair)：初始化Semaphore并指定许可证的数量。这里可以设置是否使用公平策略，如果传入true，那么Semaphore会把之前等待的线程放到FIFO的队列里，以便于当有了新的许可证可以分发给之前等了最长时间的线程；
 *  tryAcquire() ：看看现在有没有空闲的许可证，如果有的话就获取，如果没有的话也没关系，我不会陷入阻塞，我可以去做别的事，过一会再来查看许可证的空闲情况。
 *  tryAcquire(timeout): 和tryAcquire() 一样，但是多了一个超时时间，比如“在3秒内获取不到许可证，我就去做别的事”
 *  acquire()：获取许可证，可响应中断
 *  acquireUninterruptibly()：获取许可证，拒绝响应中断
 *  release()：释放许可证

#### 4、代码演示 ####

##### （1）一般用法： #####

public class SemaphoreDemo {
        
        static Semaphore semaphore = new Semaphore(3,true);
        public static void main(String[] args) {
        
            ExecutorService pool = Executors.newFixedThreadPool(50);
    
            for (int i = 0; i < 100; i++) {
        
                pool.submit(new Task());
            }
            pool.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();
            }
        }
    }

![image-20230615110440704][]

##### (2) 特殊用法 #####

一次性获取或释放多个许可证。

public class SemaphoreDemo {
        
    
        static Semaphore semaphore = new Semaphore(5, 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);
            }
        }
    }

![4fb232ef5b0be5df9ec7afdf1510f876.png][]

**什么时候需要一次性获取多个许可证：**

比如 TaskA 会调用很消耗资源的 method1()，而 TaskB 调用的是不太消耗资源的 method2()，假设我们一共有5个许可证。那么我们就可以要求 TaskA 获取5个许可证才能执行，而 TaskB 只需要获取到一个许可证就能执行，这样就避免了A和B同时运行的情况，我们可以根据自己的需求，通过分配许可证的方式合理分配资源。

#### 5、注意点 ####

*  `获取&释放`的许可证数量要求必须`一致`，否则程序运行到最后，许可证都会被占用，就会都陷入阻塞
 *  根据情况设置`公平性`，一般设置为true，这样可以避免线程饥饿
 *  释放和获取对线程没有要求，可以由`这个线程A`获取，`别的线程B`释放
 *  可以将它实现成一个`轻量级的CountDownLatch`，比如信号量Semaphore的许可证数量为1，线程A获取到，线程B执行 acquire() 再获取时，就会陷入阻塞，线程A 执行release() 之后，线程B才能执行，相当于 CountDownLatch 的唤醒

### 四、Condition条件对象 ###

#### 1. 作用 ####

当线程1需要等待某个条件的时候，它就去执行condition.await0方法，一旦执行了await0方法，线程就会进入阻塞状态。

然后通常会有另外一个线程，假设是线程2，去执行对应的条件，直到这个条件达成的时候，线程2就会去执行condition.signal() 方法，这时JVM就会从被阻塞的线程里，找到那些等待该condition的线程，这时线程1就会收到可执行信号，它的线程状态就会变成Runnable可执行状态

#### 2. 图解 ####

![3ff2ccb757c299853a7de9c4ba17ae8a.png][]

#### 3、signalAll() 和 signal() 的区别 ####

*  signalAll()唤醒`所有`等待的线程，signal()唤醒一个
 *  signal()是`公平`的，会唤醒`等待时间最长的线程`

#### 4、代码演示 ####

##### （1）基本用法 #####

condition一般是绑定在锁lock上面的，基本用法如下：

public class ConditionDemo1 {
        
        private ReentrantLock lock = new ReentrantLock();
        private Condition condtion = lock.newCondition();
    
        void method1() throws InterruptedException {
        
            lock.lock();
            try{
        
                System.out.println("条件不满足，开始wait");
                condtion.await();
                System.out.println("条件满足，开始执行后续的任务");
            }finally {
        
                lock.unlock();
            }
        }
    
        void method2(){
        
            lock.lock();
            try{
        
                System.out.println("准备工作完成，开始唤醒其他线程");
                condtion.signal();
            }finally {
        
                lock.unlock();
            }
        }
    
        //主函数
        public static void main(String[] args) throws InterruptedException {
        
            ConditionDemo1 demo1 = new ConditionDemo1();
            //主线程创建一个线程1
            new Thread(new Runnable() {
        
                @Override
                public void run() {
        
                    try {
        
                        Thread.sleep(1000);
                        demo1.method2();//线程1，1秒后，唤醒主线程
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                }
            }).start();
            demo1.method1();//主线程阻塞
        }
        
    }

![image-20230615112138646][]

##### （2）用Condition实现生产者消费者模式 #####

/**
     *      演示Condition实现生产者消费者模式
     */
    public class ConditionDemo2 {
        
        private static int queueSize = 10;
        private static PriorityQueue<Integer> queue = new PriorityQueue<>(queueSize);
    
        private static ReentrantLock lock = new ReentrantLock();
        private static Condition noFull = lock.newCondition();
        private static Condition notEmpty = lock.newCondition();
    
        //消费者
        static class Consumer extends Thread {
        
            @Override
            public void run() {
        
                comsume();
            }
    
            //消费操作
            void comsume(){
        
                while (true){
        
                    lock.lock();
                    try {
        
                        while (queue.size()==0){
        
                            System.out.println("队列空，等待数据");
                            notEmpty.await();
                        }
                        queue.poll();
                        noFull.signalAll();
                        System.out.println("从队列里取走了一个数据，队列还剩余空间"+(queueSize-queue.size())+"个元素");
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                }
            }
        }
    
        //生产者
        static class Producer extends Thread {
        
            @Override
            public void run() {
        
                produce();
            }
    
            //生产操作
            void produce(){
        
                while (true){
        
                    lock.lock();
                    try {
        
                        while (queue.size()==queueSize){
        
                            System.out.println("队列满，等待消费");
                            noFull.await();
                        }
                        queue.offer(1);
                        notEmpty.signalAll();
                        System.out.println("给队列生成了一个数据，队列有"+queue.size()+"个元素");
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                }
            }
        }
    
        //主函数
        public static void main(String[] args) {
        
            Consumer consumer = new Consumer();
            Producer producer = new Producer();
            consumer.start();
            producer.start();
        }
    
    }

![449a536c8b3c7b093e078cebeb8d07a7.png][]

#### 5. 注意点 ####

*  实际上，如果说Lock可以用来代替synchronized，那么Condition就可以用来代替相对应的Obiect.wait/notify的，所以Condition在用法和性质上，几乎和Obiect.wait/notify都一样
 *  await方法执行后会自动释放持有的Lock锁，和Object.wait一样，不需要自己手动先释放锁
 *  调用await的时候，必须持有锁，否则会抛出异常，和Object.wait一样

### 五、CyclicBarrier循环栅栏 ###

#### 1. 作用 ####

*  CyclicBarrier循环栅栏和CountDownLatch很类似，都能阻塞一组线程
 *  当有大量线程相互配合，分别计算不同任务，并且需要最后统汇总的时候，我们可以使用CyclicBarrier。CvclicBarrier可以构造一个集结点，每一个线程执行完毕后，都会到集结点等待，直到所有线程都到了集结点，那么该栅栏就被撤销，所有线程再统一出发，继续执行剩下的任务
 *  就像我们在生活中聚会时，首先约定咱们3个人明天中午在学校碰面，都到齐后再一起继续接下来的安排。
 *  CyclicBarrier循环栅栏是可以**重复使用**的，这一点和 CountDownLatch 不一样

#### 2、代码演示 ####

public class CyclicBarrierDemo {
        
        public static void main(String[] args) {
        
            //参数1：设置几个等待数
            //参数2：当线程数满足条件后，执行的任务
            CyclicBarrier cyclicBarrier = new CyclicBarrier(5, new Runnable() {
        
                @Override
                public void run() {
        
                    System.out.println("五个到齐，走一波");
                    System.out.println(Thread.currentThread().getName());
                }
            });
    
            //创建10个线程
            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 | BrokenBarrierException e) {
        
                    e.printStackTrace();
                }
            }
        }
    
    }

![ab4fd5107edbcf2b89f2e452ccf9ebcb.png][]

每当凑齐五个之后，就出发一波，cyclicBarrier可以被重复使用。

#### 3、CountDownLatch & CyclicBarrier 的区别 ####

*  作用不同
    
     *  *CountDownLatch*：`对应的是事件`，完成某个事件就可以调用countDown()，一个线程可执行多次countDown()，且该线程不会阻塞
     *  *CyclicBarrier*：`对应的是线程`，每个线程都执行await()，线程执行完await()会阻塞，达到指定的数量才会继续运行，
 *  可重用性不同
    
     *  *CountDownLatch*：倒数count到0之后，该实例就`不能再使用`
     *  *CyclicBarrier*：满足指定的数量条件后，就继续执行，且`可再次使用`
 *  结束后统一工作
    
     *  *CountDownLatch*：结束后就只是唤醒线程继续工作
     *  *CyclicBarrier*：可以在CyclicBarrier的构造函数中自定义 runnable 任务，结束后会执行该任务

#### [点我扫码关注微信公众号][Link 1] ####

#### 文章来源：[控制并发流程，做好线程间的协调][Link 2] ####

--------------------

### 个人微信：CaiBaoDeCai ###

### 微信公众号名称：Java知者 ###

### 微信公众号 ID： JavaZhiZhe ###

### 谢谢关注！ ###

[6d1651d9f1e14ced8b2a3fcd3dc83c95.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/6fdd6519281640cea345970f1fb371b4.png
[5f8ed6552011da46330b85448d7a4937.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/d165151532ff44e08fb75ebc627db3c2.png
[image-20230615103516182]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/8b818d40ea474451b8b783ef2f90ef3e.png
[47d64ff7c33d66faf5880f14d12934ff.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/90f3bdd9ab0b4da29badca6027bfa935.png
[a042a158bd8464160a066c4f685fd007.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/567ddbe66b274bb39e7aed6874d7380e.png
[5c40ce3e5634133f51af7b9719ad938d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/e76dcbbde01a4d00ab9dfa5e2718d216.png
[image-20230615110440704]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/4279746418004850a27d3894c4228f2f.png
[4fb232ef5b0be5df9ec7afdf1510f876.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/2bb3151f29c848738f790d7b5ca04b22.png
[3ff2ccb757c299853a7de9c4ba17ae8a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/d0f25b562067482b85a8897466b2c9ae.png
[image-20230615112138646]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/6f11e08e0e0a4d0ca346034924b138df.png
[449a536c8b3c7b093e078cebeb8d07a7.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/38ba8a11993e4c66ba4fc165a4f1eb19.png
[ab4fd5107edbcf2b89f2e452ccf9ebcb.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/74448cc749f34350b517ad3f3e60f2da.png
[Link 1]: https://docs.qq.com/doc/DSVhXa2JraFpVVE9V
[Link 2]: https://mp.weixin.qq.com/s?__biz=MzU3NDgyMTIxMw==&mid=2247484965&idx=1&sn=b1473328f1bb08410557b99b5adecf7d&chksm=fd2dc017ca5a4901af3c7663e1135e97c5726b4e6048ff651e42ef1ab6fb654e84a6e6d6c000#rd