并发工具类
CountDownLatch
它是一个同步辅助类，在计数到达 0 之前，使线程一直处于等待状态，直到计数结束，线程才继续工作
用法
一个线程等待多个线程都执行完毕，再继续自己的工作
多个线程等待某一个线程的信号，同时开始执行
主要方法
- CountDownLatch：仅有一个构造函数，参数 count 为需要倒数的数值
- await：调用 await() 方法的线程会被挂起，它会等待直到 count 值为 0 才继续执行
- countDown：将 count 值减 1，直到为 0 时，等待的线程会被唤起
示例
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 CountDownLatchDemo {
public static void main(String[] args) throws InterruptedException {
CountDownLatch begin = new CountDownLatch(1);
ExecutorService service = Executors.newFixedThreadPool(5);
for (int i = 0; i < 5; i++) {
final int no = i + 1;
Runnable runnable = () -> {
System.out.println("No." + no + "准备完毕，等待发令枪");
try {
begin.await();
System.out.println("No." + no + "开始跑步了");
} catch (InterruptedException e) {
e.printStackTrace();
}
};
service.submit(runnable);
}
Thread.sleep(5000);
System.out.println("发令枪响，比赛开始！");
begin.countDown();
}
}
|
注意
CountDownLatch 是不能够重用的，如果需要重新计数，可以考虑使用 CyclicBarrier 或者创建新的 CountDownLatch 实例
CyclicBarrier
CyclicBarrier 和 CountDownLatch 很类似，都能阻塞一组线程
当有大量线程相互配合，分别计算不同任务，并且需要最后统一汇总的时候，我们可以使用 CyclicBarrier，它可以构造一个集结点，当某一个线程执行完毕，就会到集结点等待，直到所有线程都到了集结点，那么该栅栏就被撤销，所有线程再统一出发，继续执行剩下的任务
示例
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
|
public class CyclicBarrierDemo {
public static void main(String[] args) {
CyclicBarrier cyclicBarrier = new CyclicBarrier(5, () -> System.out.println("所有人都到场了，大家统一出发"));
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();
}
}
}
}
|
和 CountDownLatch 的区别
- 作用不同：CyclicBarrier 要等固定数量的线程都到达了栅栏位置才能继续执行，而 CountDownLatch 只需等待数字到 0，也就是说，CountDownLatch 用于事件，但是 CyclicBarrier 是用于线程的
- 可重用性不同：CountDownLatch 在倒数到 0 并触发门闩打开后，就不能再次使用了，除非新建新的实例，而 CyclicBarrier 可以重复使用
Semaphore
Semaphore（信号量）是用来控制同时访问特定资源的线程数量，通过协调各个线程，以保证合理的使用公共资源，当达到指定数量时，只能等待其他线程释放信号量
主要方法
- Semaphore：创建一个信号量，参数 permits 为许可证数量，fair 是否公平
- tryAcquire：尝试获取许可证，参数 timeout 指定超时时间
- acquire：获得许可证（如果有的话），然后立即返回，使可用许可证的数量减少一个
- release：释放许可证
示例
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
|
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(() -> {
try {
semaphore.acquire();
System.out.println(Thread.currentThread().getName() + "拿到了许可证");
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
System.out.println(Thread.currentThread().getName() + "释放了许可证");
semaphore.release();
}
});
}
service.shutdown();
}
}
|
注意
- 获取和释放的许可证数量必须一致，否则比如每次都获取 2 个但是只释放 1 个甚至不释放，随着时间的推移，到最后许可证数量不够用，会导致程序卡死，虽然信号量类并不对释放和获取的数量做规定，但是这是我们的编程规范，否则容易出错
- 注意在初始化 Semaphore 的时候设置公平性，一般设置为 true 会更合理
- 并不是必须由获取许可证的线程释放那个许可证，事实上，获取和释放许可证对线程并无要求，也许是 A 获取了，然后由 B 释放，只要逻辑合理即可
- 信号量的作用，除了控制临界区最多同时有 N 个线程访问外，另一个作用是可以实现“条件等待”，例如线程 1 需要在线程 2 完成准备工作后才能开始工作，那么就线程 1 获取许可证，而线程 2 完成任务后释放许可证，这样的话，相当于是轻量级的 CountDownLatch
Condition
如果说 Lock 用来代替 synchronized，那么 Condition 就是用来代替相对应的 Object.wait/notify 的，所以在用法和性质上几乎都一样
await 方法会自动释放持有的 Lock 锁，和 Object 的 wait 一样，不需要自己手动释放锁
另外，调用 await 的时候，必须持有锁，否则会抛出异常，和 Object.wait 一样
示例
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
67
68
69
70
71
72
73
|
public class ConditionDemo2 {
private final int queueSize = 10;
private final PriorityQueue<Integer> queue = new PriorityQueue<>(queueSize);
private final Lock lock = new ReentrantLock();
private final Condition notFull = lock.newCondition();
private final Condition notEmpty = lock.newCondition();
public static void main(String[] args) {
ConditionDemo2 conditionDemo2 = new ConditionDemo2();
Producer producer = conditionDemo2.new Producer();
Consumer consumer = conditionDemo2.new Consumer();
producer.start();
consumer.start();
}
class Consumer extends Thread {
@Override
public void run() {
while (true) {
lock.lock();
try {
while (queue.size() == 0) {
System.out.println("队列空，等待数据");
try {
notEmpty.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
queue.poll();
notFull.signalAll();
System.out.println("从队列里取走了一个数据，队列剩余" + queue.size() + "个元素");
} finally {
lock.unlock();
}
}
}
}
class Producer extends Thread {
@Override
public void run() {
while (true) {
lock.lock();
try {
while (queue.size() == queueSize) {
System.out.println("队列满，等待有空余");
try {
notFull.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
queue.offer(1);
notEmpty.signalAll();
System.out.println("向队列插入了一个元素，队列剩余空间" + (queueSize - queue.size()));
} finally {
lock.unlock();
}
}
}
}
}
|