闭锁CountDownLatch、栅栏CyclicBarrier、信号量Semaphore

Java 5 是Java 历史上非常重要的一个版本,它提供了泛型、for-each、自动装箱和拆箱、枚举、可变参数、静态导入、注解以及java.util.concurrent并发工具包,接下来简单介绍下并发工具包下的闭锁ConutDownLatch、栅栏CyclicBarrier、信号量Semaphore。

1、闭锁CountDownLatch

java.util.concurrent.CountDownLatch 是一个并发构造,它允许一个或多个线程等待一系列指定操作的完成。CountDownLatch 以一个给定的数量初始化。countDown()每被调用一次,这一数量就减1。通过调用 await()方法,线程可以阻塞等待这一数量到达零。

示例代码如下:

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
import java.util.concurrent.CountDownLatch;

public class Test {
public static void main(String[] args) {
final CountDownLatch latch = new CountDownLatch(2);// 2个线程协作
new Thread() {
public void run() {
try {
System.out.println("子线程" + Thread.currentThread().getName()
+ "正在执行");
Thread.sleep(3000);
System.out.println("子线程" + Thread.currentThread().getName()
+ "执行完毕");
latch.countDown();
} catch (InterruptedException e) {
e.printStackTrace();
}
};
}.start();
new Thread() {
public void run() {
try {
System.out.println("子线程" + Thread.currentThread().getName()
+ "正在执行");
Thread.sleep(3000);
System.out.println("子线程" + Thread.currentThread().getName()
+ "执行完毕");
latch.countDown();
} catch (InterruptedException e) {
e.printStackTrace();
}
};
}.start();
try {
System.out.println("等待2个子线程执行完毕…");
latch.await();
System.out.println("2个子线程已经执行完毕");
System.out.println("继续执行主线程");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

运行结果:

子线程Thread-0正在执行
等待2个子线程执行完毕…
子线程Thread-1正在执行
子线程Thread-0执行完毕
子线程Thread-1执行完毕
2个子线程已经执行完毕
继续执行主线程

闭锁还有其他用途,例如,游戏中,需要8名玩家都准备就绪后,游戏才能开始。也可以利用闭锁计算各个线程执行完成所需的时间。

2、栅栏CyclicBarrier

CyclicBarrier 类是一种同步机制,它能对处理一些算法的线程实现同步。换句话说,它就是一个所有线程必须等待的一个栅栏,直到所有线程都到达这里,然后所有线程才可以继续做其他事情。通过调用 CyclicBarrier 对象的 await()方法,两个线程可以实现互相等待。一旦 N 个线程在等待 CyclicBarrier 达成,所有线程将被释放掉去继续执行。

示例代码如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class CyclicBarrierDemo {
public static void main(String[] args) {
ExecutorService service = Executors.newFixedThreadPool(5);
final CyclicBarrier barrier = new CyclicBarrier(5);
for (int i = 0; i < 5; i++) {
service.execute(new Player("玩家" + i, barrier));
}
service.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
import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
public class Player implements Runnable {
private final String name;
private final CyclicBarrier barrier;
public Player(String name, CyclicBarrier barrier) {
this.name = name;
this.barrier = barrier;
}
public void run() {
try {
TimeUnit.SECONDS.sleep(1 + (new Random().nextInt(3)));
System.out.println(name + "已准备,等待其他玩家准备…");
barrier.await();
TimeUnit.SECONDS.sleep(1 + (new Random().nextInt(3)));
System.out.println(name + "已加入游戏");
} catch (InterruptedException e) {
System.out.println(name + "离开游戏");
} catch (BrokenBarrierException e) {
System.out.println(name + "离开游戏");
}
}
}

运行结果:

玩家3已准备,等待其他玩家准备…
玩家4已准备,等待其他玩家准备…
玩家2已准备,等待其他玩家准备…
玩家0已准备,等待其他玩家准备…
玩家1已准备,等待其他玩家准备…
玩家4已加入游戏
玩家1已加入游戏
玩家0已加入游戏
玩家3已加入游戏
玩家2已加入游戏

3、信号量 Semaphore

Semaphore类是一个计数信号量。这就意味着它具备两个主要方法:

● acquire()获取

● release()释放

计数信号量由一个指定数量的“许可”初始化。每调用一次 acquire(),一个许可会被调用线程取走。每调用一次 release(),一个许可会被还给信号量。因此,在没有任何 release()调用时,最多有 N 个线程能够通过 acquire()方法,N 是该信号量初始化时的许可的指定数量。

Semaphore就是一个信号量,它的作用是限制某段代码块的并发数。Semaphore有一个构造函数,可以传入一个int型整数n,表示某段代码最多只有n个线程可以访问,如果超出了n,那么请等待,等到某个线程执行完毕这段代码块,下一个线程再进入。由此可以看出如果Semaphore构造函数中传入的int型整数n=1,相当于变成了一个synchronized了。

示例代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
public class TestSemaphore {
private static final int THREAD_COUNT = 10;
private static ExecutorService threadPool = Executors.newFixedThreadPool(THREAD_COUNT);
private static Semaphore s = new Semaphore(5);
public static void main(String[] args) {
for (int i = 0; i < THREAD_COUNT; i++) {
threadPool.execute(new Employee(String.valueOf(i), s));
}
threadPool.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
import java.util.Random;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

class Employee implements Runnable {
private String id;
private Semaphore semaphore;
private static Random rand = new Random(47);

public Employee(String id, Semaphore semaphore) {
this.id = id;
this.semaphore = semaphore;
}

public void run() {
try {
semaphore.acquire();
System.out.println(this.id + " is using the toilet");
TimeUnit.MILLISECONDS.sleep(rand.nextInt(2000));
System.out.println(this.id + " is leaving");
semaphore.release();
} catch (InterruptedException e) {
}
}
}

运行结果:

1 is using the toilet
0 is using the toilet
2 is using the toilet
6 is using the toilet
3 is using the toilet
1 is leaving
9 is using the toilet
2 is leaving
7 is using the toilet
3 is leaving
4 is using the toilet
9 is leaving
8 is using the toilet
4 is leaving
5 is using the toilet
0 is leaving
5 is leaving
8 is leaving
7 is leaving
6 is leaving

假设有5个semaphore(厕所),有10个线程(员工)取占用,肯定只有5个线程(员工)能够占用,每当一个线程(员工) 使用完(release) ,会有另一个线程占用(acquire) 。可用来控制线程并发数,比如数据库连接等。

4、总结:

1、CountDownLatch和CyclicBarrier都能够实现线程之间的等待,只不过它们侧重点不同:

CountDownLatch一般用于某个线程A等待若干个其他线程执行完任务之后,它才执行;

而CyclicBarrier一般用于一组线程互相等待至某个状态,然后这一组线程再继续执行;

另外,CountDownLatch是不能够重用的,而CyclicBarrier是可以重用的。

2、Semaphore其实和锁有点类似,它一般用于控制对某组资源的访问权限。


闭锁CountDownLatch、栅栏CyclicBarrier、信号量Semaphore
https://river106.cn/posts/e28ade9f.html
作者
river106
发布于
2018年9月15日
许可协议