多线程编程--异步转同步之CountDownLatch

冷不防 2022-05-22 10:59 313阅读 0赞

转载自https://blog.csdn.net/vic\_fang/article/details/24653477

————————————————————————————————————————————————————

在日常开发中，我们经常会碰到这样的情况：一些异步请求我们需要等到接收到请求后再执行下一步动作，这时我们就需要把异步动作转为同步动作。

java中给我们提供了一个CountDownLatch类来实现这个功能。

先看下CountDownLatch的官方定义：

A synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads completes.

意思是：CountDownLatch是一个同步助手，它可以使一个或多个线程等待，直到一系列在其他线程中执行的动作完成（这些线程才被唤醒）

具体如何使用呢？

看下这段文档：

**A `CountDownLatch` is initialized with a givencount. The`await()` methods block until the current count reaches zero due to invocations of the countDown() method, after which all waiting threads are released and any subsequent invocations of`await()` return immediately. **

**A useful property of a`CountDownLatch` is that it doesn't require that threads calling`countDown` wait for the count to reach zero before proceeding, it simply prevents any thread from proceeding past an`await()` until all threads could pass. **

这句翻译出来会有偏差，大家自己体会吧

详细用例：

\[java\] [view plain][] [copy][view plain]

1.  **class** Driver \{ // ...      
2.    **void** main() **throws** InterruptedException \{  
3.        CountDownLatch startSignal = **new** CountDownLatch(1);  
4.        CountDownLatch doneSignal = **new** CountDownLatch(N);  
5.          
6.        **for** (**int** i = 0; i < N; ++i) // create and start threads  
7.          **new** Thread(**new** Worker(startSignal, doneSignal)).start();  
8.          
9.        doSomethingElse();            // don't let run yet  
10.       startSignal.countDown();      // let all threads proceed  
11.       doSomethingElse();  
12.       doneSignal.await();           // wait for all to finish  
13.   \}     
14.   **class** Worker **implements** Runnable \{  
15.     **private** **final** CountDownLatch startSignal;  
16.     **private** **final** CountDownLatch doneSignal;  
17.      
18.     Worker(CountDownLatch startSignal, CountDownLatch doneSignal) \{  
19.       **this**.startSignal = startSignal;  
20.       **this**.doneSignal = doneSignal;  
21.     \}  
22.   
23.         **public** **void** run() \{  
24.       **try** \{  
25.         startSignal.await();  
26.         doWork();  
27.         doneSignal.countDown();  
28.       \} **catch** (InterruptedException ex) \{\} // return;  
29.     \}  
30.        
31.     **void** doWork() \{ ... \}    
32.   \}  
33. \}

——————————————————————-—————————————————————————————

最近项目在执行过程中，有多个 **独立模块** 异步执行，将执行结果统一处理后返回，代码可以顺序调用各个模块执行，然后统一处理，但是效率过低，考虑采用多线程异步处理，但异步执行提交任务后就顺序执行其他代码了，无法统一获取各模块处理结果。采用**countDownLatch**可以等待所有异步线程执行完成再统一处理。

*  countDownLatch jdk里描述   
    `A synchronization aid that allows one or more threads to wait until  a set of operations being performed in other threads completes.  一个允许等待一个或多个其它线程执行完成的同步助手  使用场景：  1.等待一个或多个线程完成再执行其它操作  2.等待一个线程执行多次再执行其它操作 `
 *  **countDownLatch**实现原理

private final Sync sync;
    public CountDownLatch(int count) {
            if (count < 0) throw new IllegalArgumentException("count < 0");
            this.sync = new Sync(count);
        }
    初始化时将任务个数传递给同步控制器Sync，每次调用countDown方法，开始执行异步任务，Sync释放一个资源
    public void countDown() {
            sync.releaseShared(1);
        }
    
    调用await，await方法判断Sycn的状态，如果Sync状态显示未执行完成，则继续分配资源执行。
    当countDown为0时，统一返回异步结果。

*  代码示例

import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.*;
    
    
    public class MultiJobExecutors {
        private CountDownLatch countDownLatch;
        private List<Task> tasks;
        private  static ExecutorService executor = null;
    
        public MultiJobExecutors(List<Task> tasks) {
            countDownLatch = new CountDownLatch(tasks.size());
            executor = Executors.newFixedThreadPool(tasks.size());
            this.tasks = tasks;
        }
    
        public void execute() {
            if (tasks == null|| tasks.isEmpty()) {
                return;
            }
            for (int i=0;i< tasks.size();i++) {
                executor.submit(new Job(countDownLatch,tasks.get(i)));
                System.out.println("xx"+i);
            }
            try {
                //等待所有线程结束
                countDownLatch.await(15, TimeUnit.SECONDS);
                //执行其他操作
                System.out.println("it's over");
                //关闭线程池
                executor.shutdown();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    
        private class Job implements Callable<Object>{
            private CountDownLatch latch;
            private Task task;
    
            public Job(CountDownLatch latch, Task task) {
                this.latch = latch;
                this.task = task;
            }
    
            @Override
            public Object call() throws Exception {
                System.out.println(System.currentTimeMillis());
                //执行线程
                task.execute();
                //countDown自减
                latch.countDown();
                return null;
            }
        }
        private static class Task{
            private String str;
    
            public Task(String str) {
                this.str = str;
            }
    
            public void execute(){
                System.out.println(str);
            }
    
        }
    
        public static void main(String[] args) {
            Task task = new Task("I");
            Task task1 = new Task("love");
            Task task2 = new Task("you");
            Task task3 = new Task(",");
            Task task4 = new Task("its");
            Task task5 = new Task("not");
            Task task6 = new Task("true");
    
            List<Task> tasks = new ArrayList<Task>();
            tasks.add(task);
            tasks.add(task1);
            tasks.add(task2);
            tasks.add(task3);
            tasks.add(task4);
            tasks.add(task5);
            tasks.add(task6);
    
            MultiJobExecutors multiJobExecutors = new MultiJobExecutors(tasks);
            multiJobExecutors.execute();
        }
    }

*  执行结果

xx0
    1472706027197
    I
    xx1
    1472706027197
    love
    xx2
    1472706027197
    you
    xx3
    1472706027197
    ,
    xx4
    1472706027197
    its
    xx5
    1472706027198
    not
    xx6
    1472706027198
    true
    it's over

### CountDownLatch的概念 ###

CountDownLatch是一个同步工具类，用来协调多个线程之间的同步，或者说起到线程之间的通信（而不是用作互斥的作用）。

CountDownLatch能够使一个线程在等待另外一些线程完成各自工作之后，再继续执行。使用一个计数器进行实现。计数器初始值为线程的数量。当每一个线程完成自己任务后，计数器的值就会减一。当计数器的值为0时，表示所有的线程都已经完成了任务，然后在CountDownLatch上等待的线程就可以恢复执行任务。

### CountDownLatch的用法 ###

CountDownLatch典型用法1：某一线程在开始运行前等待n个线程执行完毕。将CountDownLatch的计数器初始化为n `new CountDownLatch(n)` ，每当一个任务线程执行完毕，就将计数器减1 `countdownlatch.countDown()`，当计数器的值变为0时，在CountDownLatch上 `await()` 的线程就会被唤醒。一个典型应用场景就是启动一个服务时，主线程需要等待多个组件加载完毕，之后再继续执行。

CountDownLatch典型用法2：实现多个线程开始执行任务的最大**并行性**。注意是并行性，不是并发，强调的是多个线程在某一时刻同时开始执行。类似于赛跑，将多个线程放到起点，等待发令枪响，然后同时开跑。做法是初始化一个共享的CountDownLatch(1)，将其计数器初始化为1，多个线程在开始执行任务前首先 `coundownlatch.await()`，当主线程调用 `countDown()` 时，计数器变为0，多个线程同时被唤醒。

### CountDownLatch的不足 ###

CountDownLatch是一次性的，计数器的值只能在构造方法中初始化一次，之后没有任何机制再次对其设置值，当CountDownLatch使用完毕后，它不能再次被使用。

参考资料：[http://www.importnew.com/15731.html][http_www.importnew.com_15731.html]

[view plain]: https://blog.csdn.net/vic_fang/article/details/24653477#
[http_www.importnew.com_15731.html]: http://www.importnew.com/15731.html