JUC之AQS之CountDownLatch

133 阅读 0 评论 88 点赞

我是靠谱客的博主慈祥自行车，这篇文章主要介绍JUC之AQS之CountDownLatch，现在分享给大家，希望可以做个参考。

CountDownLatch允许一个或多个线程等待其他线程完成操作。

比如如果要实现一个这样的功能：要让主线程等待所有线程完成自己的操作后再执行，最简单的做法是使用join()方法（可参看https://blog.csdn.net/Dongguabai/article/details/82255331）。这是一个比较简单简单的例子：

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
package dgb.test.concurrent;

/**
 * @author Dongguabai
 * @date 2018/9/24 12:06
 */
public class JoinTest2 {

    public static void main(String[] args) throws InterruptedException {
        System.out.println("主线程开始执行");
        Thread thread1 = new Thread(() -> {
            System.out.println("thread--1开始执行！");
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("thread--1执行完成！");
        });
        Thread thread2 = new Thread(() -> {
            System.out.println("thread--2开始执行！");
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("thread--2执行完成！");
        });
        thread1.start();
        thread2.start();
        thread1.join();
        thread2.join();
        System.out.println("主线程结束执行");
    }
}

join()方法的原理之前也分析过了，本质就是在当前线程中调用了另一个线程的wait()方法，然后不停检查join线程是否存活，如果join线程存活则让当前线程永远等待。其中，wait(0)表示永远等下去。直到join线程中止后，线程的this.notifyAll()方法会被调用，调用notifyAll()方法是在JVM里实现的，所以在JDK里看不到。

其实在JUC中有一个更好用的工具，就是CountDownLatch。

CountDownLatch的构造函数接收一个int类型的参数作为计数器，如果你想等待N个点完成，这里就传入N。当我们调用CountDownLatch的countDown方法时，N就会减1，CountDownLatch的await方法会阻塞当前线程，直到N变成零。由于countDown方法可以用在任何地方，所以这里说的N个点，可以是N个线程，也可以是1个线程里的N个执行步骤。用在多个线程时，只需要把这个CountDownLatch的引用传递到线程里即可。如果有某个操作执行得比较慢，我们不可能让主线程一直等待，所以可以使用另外一个带指定时间的await方法——await（long time，TimeUnit unit），这个方法等待特定时间后，就会不再阻塞当前线程。join也有类似的方法。

要注意的是：计数器必须大于等于0，只是等于0时候，计数器就是零，调用await方法时不会阻塞当前线程。CountDownLatch不可能重新初始化或者修改CountDownLatch对象的内部计数器的值（如果想实现重置功能，可以使用JUC中的另一个工具CyclicBarrier，这个在后面会介绍）。一个线程调用countDown方法happen-before，另外一个线程调用await方法。

下面是几个简单的例子：

复制代码

package dgb.test.concurrent;

import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author Dongguabai
 * @date 2018/9/24 11:46
 */
public class CountDownLaunthDemo implements Runnable{

static int count = 10;

static CountDownLatch end = new CountDownLatch(count);
    static CountDownLaunthDemo demo = new CountDownLaunthDemo();

@Override
    public void run() {
        try {
            Thread.sleep(new Random().nextInt(10)*1000);
            System.out.println("check complete");
            end.countDown();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

public static void main(String[] args) throws InterruptedException {
        ExecutorService exec = Executors.newFixedThreadPool(count);
        for (int i = 0; i < count; i++) {
            exec.submit(demo);
        }
        end.await();
        System.out.println("-------------Fire");
        exec.shutdownNow();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
package dgb.test.concurrent;

import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author Dongguabai
 * @date 2018/9/24 11:46
 */
public class CountDownLaunthDemo implements Runnable{

    static int count = 10;

    static CountDownLatch end = new CountDownLatch(count);
    static CountDownLaunthDemo demo = new CountDownLaunthDemo();

    @Override
    public void run() {
        try {
            Thread.sleep(new Random().nextInt(10)*1000);
            System.out.println("check complete");
            end.countDown();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ExecutorService exec = Executors.newFixedThreadPool(count);
        for (int i = 0; i < count; i++) {
            exec.submit(demo);
        }
        end.await();
        System.out.println("-------------Fire");
        exec.shutdownNow();
    }
}

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
public class TestBingfa {
        //发送请求的url地址
	private final String url = "http://localhost:8085/bda-search";
	//模拟的并发量
	private static final int BINGFA = 199;
	
	private static CountDownLatch cdl = new CountDownLatch(BINGFA);
	
	public static void main(String[] args) {
		for (int i = 0; i < BINGFA; i++) {
			new Thread(new UserRequest()).start();
			cdl.countDown();
		}
	}
	
	public static class UserRequest implements Runnable{
		@Override
		public void run() {
			try {
				cdl.await();
			} catch (Exception e) {
				e.printStackTrace();
			}
	        //使用工具类发送http请求
                String json2 = HttpClientUtil.sendHttpPostJson(url, getJson());
	        System.out.println(new Date().getTime()+"::"+json2);
		}
		
	}
	
	//发送的请求参数
	public static String getJson(){
           return null;
	}
}

复制代码

package dgb.test.concurrent;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @author Dongguabai
 * @date 2018/9/24 11:46
 */
@Slf4j
public class CountDownLaunthDemo {

private final static int threadCount = 200;

public static void main(String[] args) throws Exception {

ExecutorService exec = Executors.newCachedThreadPool();

final CountDownLatch countDownLatch = new CountDownLatch(threadCount);

for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    test(threadNum);
                } catch (Exception e) {
                    log.error("exception", e);
                } finally {
                    countDownLatch.countDown();
                }
            });
        }
        countDownLatch.await(10, TimeUnit.MILLISECONDS);
        log.info("finish");
        exec.shutdown();
    }

private static void test(int threadNum) throws Exception {
        Thread.sleep(100);
        log.info("{}", threadNum);
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
package dgb.test.concurrent;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @author Dongguabai
 * @date 2018/9/24 11:46
 */
@Slf4j
public class CountDownLaunthDemo {

    private final static int threadCount = 200;

    public static void main(String[] args) throws Exception {

        ExecutorService exec = Executors.newCachedThreadPool();

        final CountDownLatch countDownLatch = new CountDownLatch(threadCount);

        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    test(threadNum);
                } catch (Exception e) {
                    log.error("exception", e);
                } finally {
                    countDownLatch.countDown();
                }
            });
        }
        countDownLatch.await(10, TimeUnit.MILLISECONDS);
        log.info("finish");
        exec.shutdown();
    }

    private static void test(int threadNum) throws Exception {
        Thread.sleep(100);
        log.info("{}", threadNum);
    }


}