AQS实现之Semaphore、CountDownLatch、CyclicBarrier分析_走进高并发:aqs的countdownlatch、semaphore与cyclicbarrier揭秘

作者：Gausst松鼠会 | 2024-02-27 01:31:09

踩

走进高并发:aqs的countdownlatch、semaphore与cyclicbarrier揭秘.

对基于AQS的一些线程并发工具的分析

Semaphore
CountDownLatch
CyclicBarrier
- CyclicBarrier等待入口方法：await

Semaphore

Semaphore维持一组信号量，信号量的大小代表线程并发上限，超过上限后线程将被阻塞直到其它线程释放信号，该类可用于控制并发大小，详见下面例子，例子中定义了一个持有3个信号量的Semaphore,同时启动12个线程去竞争这个三个信号量，竞争成功后打印提示信息

package com.test.sync;

import lombok.SneakyThrows;

import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

public class SemaphoreTest {
    static class MyTask implements  Runnable{
        private Semaphore semaphore;
        private DateFormat df = new SimpleDateFormat("HH:mm:ss");

        public MyTask(Semaphore semaphore) {
            this.semaphore = semaphore;
        }

        @SneakyThrows
        @Override
        public void run() {
            try {
                semaphore.acquire();
                System.out.println("时间："+df.format(new Date())+"，任务："+Thread.currentThread().getName()+"执行中···");
                TimeUnit.SECONDS.sleep(2);
            }finally {
                semaphore.release();
            }
        }
    }

    public static void main(String[] args) {
        Semaphore semaphore = new Semaphore(3);
        ExecutorService es = Executors.newFixedThreadPool(12);
        for (int i = 0; i < 12; i++) {
            es.submit(new MyTask(semaphore));
        }
        es.shutdown();
    }
}
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

测试结果如下：可发现每隔两秒钟有3个线程完成打印工作，12个线程前后间隔了6秒钟完成所有打印任务

时间：17:52:58，任务：pool-1-thread-1执行中···
时间：17:52:58，任务：pool-1-thread-2执行中···
时间：17:52:58，任务：pool-1-thread-3执行中···
时间：17:53:00，任务：pool-1-thread-4执行中···
时间：17:53:00，任务：pool-1-thread-5执行中···
时间：17:53:00，任务：pool-1-thread-6执行中···
时间：17:53:02，任务：pool-1-thread-7执行中···
时间：17:53:02，任务：pool-1-thread-9执行中···
时间：17:53:02，任务：pool-1-thread-8执行中···
时间：17:53:04，任务：pool-1-thread-10执行中···
时间：17:53:04，任务：pool-1-thread-11执行中···
时间：17:53:04，任务：pool-1-thread-12执行中···
1
2
3
4
5
6
7
8
9
10
11
12

源码逻辑相对较为简单：主要是构造器初始化了锁数量，每次acquire将锁数量-1，如果大于等于0，则标识获取锁成功，否则标识锁已用完进入阻塞状态，当有线程调用release时将释放锁并将所数量+1，并通知唤醒队列线程。

CountDownLatch

直译是倒计时启动：实现多个线程间执行同步，假设有n个线程各自再执行任务，但是他们之间需要在执行到指定地方时等待其它线程也到达各自指定的地方后再一起执行，可以使用该类实现步调控制，参见下面例子：

public class CountDownLatchTest {
    public static void main(String[] args) throws InterruptedException {
        ExecutorService es = Executors.newFixedThreadPool(5);
        CountDownLatch cdl = new CountDownLatch(1);
        for (int i = 0; i < 5; i++) {
            es.submit(()->{
                try {
                    //随机休眠500~2500模拟各线程任务耗时
                    TimeUnit.MILLISECONDS.sleep((int)(Math.random()*2000+500));
                    System.out.println("线程："+Thread.currentThread().getName()+"已到达指定地点，等待其它线程准备各自指定地点后再执行");
                    cdl.await();
                    System.out.println("线程："+Thread.currentThread().getName()+"已越过指定地点继续执行");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        TimeUnit.SECONDS.sleep(5);
        cdl.countDown();
        es.shutdown();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

测试结果：可以看到所有线程都到达指定点后开始等待其它线程到达，主线程通过countDown让所有线程都继续执行，这里主线程为保证所有子线程都已到达指定点通过休眠5s实现，实际上完全可以让子线程自己来判断各自线程是否已经到达指定点，参见下面例子。

线程：pool-1-thread-4已到达指定地点，等待其它线程准备各自指定地点后再执行
线程：pool-1-thread-1已到达指定地点，等待其它线程准备各自指定地点后再执行
线程：pool-1-thread-5已到达指定地点，等待其它线程准备各自指定地点后再执行
线程：pool-1-thread-2已到达指定地点，等待其它线程准备各自指定地点后再执行
线程：pool-1-thread-3已到达指定地点，等待其它线程准备各自指定地点后再执行
线程：pool-1-thread-4已越过指定地点继续执行
线程：pool-1-thread-3已越过指定地点继续执行
线程：pool-1-thread-2已越过指定地点继续执行
线程：pool-1-thread-5已越过指定地点继续执行
线程：pool-1-thread-1已越过指定地点继续执行
1
2
3
4
5
6
7
8
9
10

相比第一种控制方式，下面这种初始化计数器为线程总数量，每个线程再到达指点地点后执行countDown将计数器-1后执行await进行等待，最后一个线程到达后计数值-1为0后将通知所有被阻塞的线程继续执行，同时由于计数器为0，因此await将不在阻塞本本线程。

public class CountDownLatchTest {
    public static void main(String[] args) throws InterruptedException {
        ExecutorService es = Executors.newFixedThreadPool(5);
        CountDownLatch cdl = new CountDownLatch(5);
        for (int i = 0; i < 5; i++) {
            es.submit(()->{
                try {
                    //随机休眠500~2500模拟各线程任务耗时
                    TimeUnit.MILLISECONDS.sleep((int)(Math.random()*2000+500));
                    System.out.println("线程："+Thread.currentThread().getName()+"已到达指定地点，等待其它线程准备各自指定地点后再执行");
                    cdl.countDown();
                    cdl.await();
                    System.out.println("线程："+Thread.currentThread().getName()+"已越过指定地点继续执行");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        es.shutdown();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

测试结果如下

线程：pool-1-thread-1已到达指定地点，等待其它线程准备各自指定地点后再执行
线程：pool-1-thread-3已到达指定地点，等待其它线程准备各自指定地点后再执行
线程：pool-1-thread-2已到达指定地点，等待其它线程准备各自指定地点后再执行
线程：pool-1-thread-5已到达指定地点，等待其它线程准备各自指定地点后再执行
线程：pool-1-thread-4已到达指定地点，等待其它线程准备各自指定地点后再执行
线程：pool-1-thread-4已越过指定地点继续执行
线程：pool-1-thread-1已越过指定地点继续执行
线程：pool-1-thread-5已越过指定地点继续执行
线程：pool-1-thread-3已越过指定地点继续执行
线程：pool-1-thread-2已越过指定地点继续执行
1
2
3
4
5
6
7
8
9
10

对比第一种的区别在于第二种能再第一时间知晓是否所有线程都已经到达指定点并及时唤醒所有阻塞线程继续执行，这是自动的，而第一种则必须要通过一些检测才能发现是否全部线程已到达，但它可以控制其它线程何时唤醒执行。

循环壁垒：实现的功能和CountDownLatch很相似，都是让一组线程达到各自指定地点后相互等待，待最后一个线程到达后再同时执行，与CountDownLatch不同的是CyclicBarrier的可以指定多处地点让一组线程相互等待，同时CyclicBarrier构造器提供了一个Runnable参数barrierAction，可以让最后一个到达指定地点的线程执行该barrierAction。参见如下例子：3个线程完成相同的任务，任务分为3部分，要求每个线程完成第一部分任务后等待其它线程完成第一部分任务后再执行第二部分任务；同理要求先完成第二部分任务的线程等待其它线程完成第二部分任务，最后完成第三部分任务

public class CyclicBarrierTest {
    public static void main(String[] args) {
        CyclicBarrier cb = new CyclicBarrier(3, () -> {
            System.out.println(Thread.currentThread().getName() + "last barrier passed!");
        });
        ExecutorService es = Executors.newFixedThreadPool(3);
        for (int i = 0; i < 3; i++) {
            es.submit(() -> {
                try {
                    System.out.println(Thread.currentThread().getName() + "执行完成任务第一部分，等待其它线程完成第一部分");
                    cb.await();
                    System.out.println(Thread.currentThread().getName() + "执行完成任务第二部分，等待其它线程完成第二部分");
                    cb.await();
                    System.out.println(Thread.currentThread().getName() + "执行完成任务第三部分，等待其它线程完成第三部分");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            });
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

测试结果：各个线程按照要求再完成任务一部分后等待其它线程，重测试结果可以发现，每次最后一个完成部分的线程执行barrierAction并会最新执行下个部分任务，说明最后一个到达指定点的任务负责执行barrierAction并负责唤醒等待的任务

pool-1-thread-1执行完成任务第一部分，等待其它线程完成第一部分
pool-1-thread-2执行完成任务第一部分，等待其它线程完成第一部分
pool-1-thread-3执行完成任务第一部分，等待其它线程完成第一部分
pool-1-thread-3last barrier passed!
pool-1-thread-3执行完成任务第二部分，等待其它线程完成第二部分
pool-1-thread-2执行完成任务第二部分，等待其它线程完成第二部分
pool-1-thread-1执行完成任务第二部分，等待其它线程完成第二部分
pool-1-thread-1last barrier passed!
pool-1-thread-1执行完成任务第三部分，等待其它线程完成第三部分
pool-1-thread-3执行完成任务第三部分，等待其它线程完成第三部分
pool-1-thread-2执行完成任务第三部分，等待其它线程完成第三部分
1
2
3
4
5
6
7
8
9
10
11

虽然与CountDownLatch很相似，但实际上二者逻辑完全不一样。

CyclicBarrier等待入口方法：await

    public int await() throws InterruptedException, BrokenBarrierException {
        try {
            return dowait(false, 0L);
        } catch (TimeoutException toe) {
            throw new Error(toe); // cannot happen
        }
    }
1
2
3
4
5
6
7

dowait源码如下：

    private int dowait(boolean timed, long nanos)
        throws InterruptedException, BrokenBarrierException,
               TimeoutException {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
        	//引用当前head，在后续过程中由于当前线程释放锁或所有线程已到达会更新generation导致g!=generation
            final Generation g = generation;
            //检测是否broken,阻塞中断或barrierAction异常将导致此处broken=true;
            if (g.broken)
                throw new BrokenBarrierException();
			//响应中断并抛出中断异常，其它阻塞线程将抛出BrokenBarrierException
            if (Thread.interrupted()) {
                breakBarrier();
                throw new InterruptedException();
            }

            int index = --count;
            if (index == 0) {  // 如果最后一个线程到达指定点，如果存在barrierCommand将执行该barrierCommand
                boolean ranAction = false;
                try {
                    final Runnable command = barrierCommand;
                    if (command != null)
                        command.run();
                    ranAction = true;
                    //进入下一个指定点进行相互等待
                    nextGeneration();
                    return 0;
                } finally {
                	//执行barrierCommand抛出异常则将breakBarrier
                    if (!ranAction)
                        breakBarrier();
                }
            }
            for (;;) {
                try {
                	//非超时等待
                    if (!timed)
                        trip.await();
                    //超时等待
                    else if (nanos > 0L)
                        nanos = trip.awaitNanos(nanos);
                } catch (InterruptedException ie) {
                	//阻塞中断将broken并抛出中断异常，其它线程将抛出BrokenBarrierException
                    if (g == generation && ! g.broken) {
                        breakBarrier();
                        throw ie;
                    } else {
                        Thread.currentThread().interrupt();
                    }
                }
                if (g.broken)
                    throw new BrokenBarrierException();

                if (g != generation)
                    return index;

                if (timed && nanos <= 0L) {
                    breakBarrier();
                    throw new TimeoutException();
                }
            }
        } finally {
            lock.unlock();
        }
    }
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
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

上述流程分以下几个分支(不讨论parties!=threads的情况)

没有中断请求
- 如果barrierAction没有发生异常，则将按照正常流程工作，不做说明
- 如果barrierAction发生异常，这件导致执行barrierAction的线程抛出该异常并导致其它阻塞线程抛出BrokenBarrierException，测试代码如下：

public static void main(String[] args) {
        CyclicBarrier cb = new CyclicBarrier(3, () -> {
            System.out.println(Thread.currentThread().getName() + "last barrier passed!");
            throw new Error("error");
        });
        ExecutorService es = Executors.newFixedThreadPool(3);
        for (int i = 0; i < 3; i++) {
            es.submit(() -> {
                try {
                    System.out.println(Thread.currentThread().getName() + "执行完成任务第一部分，等待其它线程完成第一部分");
                    cb.await();
                    System.out.println(Thread.currentThread().getName() + "执行完成任务第二部分，等待其它线程完成第二部分");
                    cb.await();
                    System.out.println(Thread.currentThread().getName() + "执行完成任务第三部分，等待其它线程完成第三部分");
                } catch (Throwable e) {
                    e.printStackTrace();
                    throw new Error(e);
                }
            });
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

测试结果：BrokenBarrierException的个数为总线数-1个，剩余一个异常是barrierAction抛出的运行时异常。

pool-1-thread-1执行完成任务第一部分，等待其它线程完成第一部分
pool-1-thread-3执行完成任务第一部分，等待其它线程完成第一部分
pool-1-thread-2执行完成任务第一部分，等待其它线程完成第一部分
pool-1-thread-1last barrier passed!
java.util.concurrent.BrokenBarrierException
	···省略异常堆栈信息
	at java.lang.Thread.run(Thread.java:748)
java.util.concurrent.BrokenBarrierException
	···省略异常堆栈信息
	at java.lang.Thread.run(Thread.java:748)
java.lang.Error: error
	···省略异常堆栈信息
	at java.lang.Thread.run(Thread.java:748)
1
2
3
4
5
6
7
8
9
10
11
12
13

存在中断请求
- 在进入await前发生中断请求或阻塞期间发生中断请求，则该线程进入await后或唤醒后将抛出中断异常，被阻塞线程将抛出BrokenBarrierException，还未到达await地点的线程将也将抛出BrokenBarrierException异常，测试代码如下：

    public static void main(String[] args) {
        CyclicBarrier cb = new CyclicBarrier(3, () -> {
            System.out.println(Thread.currentThread().getName() + "last barrier passed!");
        });
        ExecutorService es = Executors.newFixedThreadPool(3);
        AtomicInteger ai = new AtomicInteger(0);
        for (int i = 0; i < 3; i++) {
            es.submit(() -> {
                try {
                    if(ai.incrementAndGet() == 2){
                        //模拟第二个线程进入wait前发生中断
                        Thread.currentThread().interrupt();
                        System.out.println(Thread.currentThread().getName() + "进入wait前发生中断请求");
                    }
                    System.out.println(Thread.currentThread().getName() + "执行完成任务第一部分，等待其它线程完成第一部分");
                    cb.await();
                    System.out.println(Thread.currentThread().getName() + "执行完成任务第二部分，等待其它线程完成第二部分");
                    cb.await();
                    System.out.println(Thread.currentThread().getName() + "执行完成任务第三部分，等待其它线程完成第三部分");
                } catch (Throwable e) {
                    e.printStackTrace();
                    throw new Error(e);
                }
            });
        }
    }
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

测试结果：和预期一致

pool-1-thread-1进入wait前发生中断请求
pool-1-thread-1执行完成任务第一部分，等待其它线程完成第一部分
pool-1-thread-3执行完成任务第一部分，等待其它线程完成第一部分
pool-1-thread-2执行完成任务第一部分，等待其它线程完成第一部分
java.util.concurrent.BrokenBarrierException
	···省略异常堆栈信息
	at java.lang.Thread.run(Thread.java:748)
java.util.concurrent.BrokenBarrierException
	···省略异常堆栈信息
	at java.lang.Thread.run(Thread.java:748)
java.lang.InterruptedException
	···省略异常堆栈信息
	at java.lang.Thread.run(Thread.java:748)
1
2
3
4
5
6
7
8
9
10
11
12
13

CyclicBarrier使用总结：关于CyclicBarrier的两个异常InterruptedException，BrokenBarrierException，前者是线程交互主动要抛出的异常，后者是barrierAction抛出异常或InterruptedException导致其它线程被动抛出的异常。如果想要无视这两个异常(barrierAction中抛出的异常不包括在内)则可以捕获这两个异常后调用reset方法重置barrier信息并循环检测，代码如下：将能正常完成工作。

public class CyclicBarrierTest {
    public static void main(String[] args) {
        CyclicBarrier cb = new CyclicBarrier(3, () -> {
            System.out.println(Thread.currentThread().getName() + "last barrier passed!");
        });
        ExecutorService es = Executors.newFixedThreadPool(3);
        AtomicInteger ai = new AtomicInteger(0);

        for (int i = 0; i < 3; i++) {
            es.submit(() -> {
                if (ai.incrementAndGet() == 2) {
                    //模拟第二个线程进入wait前发生中断
                    Thread.currentThread().interrupt();
                    System.out.println(Thread.currentThread().getName() + "进入wait前发生中断请求");
                }
                System.out.println(Thread.currentThread().getName() + "执行完成任务第一部分，等待其它线程完成第一部分");
                wait(cb);
                System.out.println(Thread.currentThread().getName() + "执行完成任务第二部分，等待其它线程完成第二部分");
                wait(cb);
                System.out.println(Thread.currentThread().getName() + "执行完成任务第三部分，等待其它线程完成第三部分");
            });
        }
        es.shutdown();
    }

    static void wait(CyclicBarrier cb) {
        boolean ex = true;
        do {
            try {
                cb.await();
                ex = false;
            } catch (InterruptedException | BrokenBarrierException e) {
                cb.reset();
            }
        } while (ex);
    }
}
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

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/Gausst松鼠会/article/detail/148972