文章目录
Java 的并发工具类提供了一系列方便实用的方法,用于简化多线程编程。以下是一些常见的并发工具类:
Semaphore
:实现信号量,控制资源的访问。CountDownLatch
:用于等待一组线程完成。CyclicBarrier
:让一组线程等待彼此到达某个同步点。
一、Semaphore
1.1 作用
synchronized
和 ReentrantLock
都是一次只允许一个线程访问某个资源,而Semaphore
(信号量)可以用来控制同时访问特定资源的线程数量。Semaphore
的使用简单,我们这里假设有 N(N>5) 个线程来获取 Semaphore
中的共享资源,下面的代码表示同一时刻 N 个线程中只有 5 个线程能获取到共享资源,其他线程都会阻塞,只有获取到共享资源的线程才能执行。等到有线程释放了共享资源,其他阻塞的线程才能获取到。
// 初始共享资源数量
final Semaphore semaphore = new Semaphore(5);
// 获取1个许可
semaphore.acquire();
// 释放1个许可
semaphore.release();
当初始的资源个数为 1 的时候,Semaphore
退化为排他锁。
Semaphore
有两种模式:
- 公平模式: 调用 acquire() 方法的顺序就是获取许可证的顺序,遵循 FIFO
- 非公平模式: 抢占式的
Semaphore 对应的两个构造方法如下:
public Semaphore(int permits) {
sync = new NonfairSync(permits);
}
public Semaphore(int permits, boolean fair) {
sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}
这两个构造方法,都必须提供许可的数量,第二个构造方法可以指定是公平模式还是非公平模式,默认非公平模式。Semaphore
通常用于那些资源有明确访问数量限制的场景比如限流
1.2 Semaphore 的原理
Semaphore
是共享锁的一种实现,它默认构造 AQS 的 state
值为 permits
,你可以将 permits
的值理解为许可证的数量,只有拿到许可证的线程才能执行。调用semaphore.acquire()
,线程尝试获取许可证,如果 state >= 0
的话,则表示可以获取成功。如果获取成功的话,使用 CAS 操作去修改 state
的值 state=state-1
。如果 state<0 的话,则表示许可证数量不足。此时会创建一个 Node 节点加入阻塞队列,挂起当前线程。
/**
* 获取1个许可证
*/
public void acquire() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
/**
* 共享模式下获取许可证,获取成功则返回,失败则加入阻塞队列,挂起线程
*/
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
// 尝试获取许可证,arg为获取许可证个数,当可用许可证数减当前获取的许可证数结果小于0,则创建一个节点加入阻塞队列,挂起当前线程。
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
调用semaphore.release();
,线程尝试释放许可证,并使用 CAS 操作去修改 state
的值 state=state+1
。释放许可证成功之后,同时会唤醒同步队列中的一个线程。被唤醒的线程会重新尝试去修改 state
的值 state=state-1
,如果 state>=0
则获取令牌成功,否则重新进入阻塞队列,挂起线程。
// 释放一个许可证
public void release() {
sync.releaseShared(1);
}
// 释放共享锁,同时会唤醒同步队列中的一个线程。
public final boolean releaseShared(int arg) {
//释放共享锁
if (tryReleaseShared(arg)) {
//唤醒同步队列中的一个线程
doReleaseShared();
return true;
}
return false;
}
二、CountDownLatch
2.1 作用
CountDownLatch
允许count
个线程阻塞在一个地方,直至所有线程的任务都执行完毕。CountDownLatch
是一次性的,计数器的值只能在构造方法中初始化一次,之后没有任何机制再次对其设置值,当 CountDownLatch
使用完毕后,它不能再次被使用。
2.2 CountDownLatch 的原理
CountDownLatch
是共享锁的一种实现,它默认构造 AQS 的 state
值为 count
。当线程使用 countDown()
方法时,其实使用了tryReleaseShared
方法以 CAS 的操作来减少 state
,直至 state
为 0 。当调用 await()
方法的时候,如果 state
不为 0,那就证明任务还没有执行完毕,await()
方法就会一直阻塞,也就是说 await()
方法之后的语句不会被执行。直到count
个线程调用了countDown()
使 state
值被减为 0,或者调用await()
的线程被中断,该线程才会从阻塞中被唤醒,await()
方法之后的语句得到执行。
2.3 使用场景
CountDownLatch
的作用就是 允许 count
个线程阻塞在一个地方,直至所有线程的任务都执行完毕。在公司的一个使用线程池做数据导出的业务中就用到CountDownLatch
阻塞多条线程的任务,等待所有任务执行完成后才会接着执行后面的逻辑
三、CyclicBarrier
3.1 作用
CyclicBarrier
和 CountDownLatch
非常类似,它也可以实现线程间的技术等待,但是它的功能比 CountDownLatch
更加复杂和强大。主要应用场景和 CountDownLatch
类似。CountDownLatch
的实现是基于 AQS 的,而 CycliBarrier
是基于 ReentrantLock
(ReentrantLock 也属于 AQS 同步器)和 Condition
的。CyclicBarrier
的字面意思是可循环使用(Cyclic)的屏障(Barrier)。它要做的事情是:让一组线程到达一个屏障(也可以叫同步点)时被阻塞,直到最后一个线程到达屏障时,屏障才会开门,所有被屏障拦截的线程才会继续干活。
3.2 CyclicBarrier 的原理
CyclicBarrier
内部通过一个 count
变量作为计数器,count
的初始值为 parties
属性的初始化值,每当一个线程到了栅栏这里了,那么就将计数器减 1。如果 count
值为 0 了,表示这是这一代最后一个线程到达栅栏,就尝试执行我们构造方法中输入的任务。
//每次拦截的线程数
private final int parties;
//计数器
private int count;
下面我们结合源码来简单看看。
1、CyclicBarrier
默认的构造方法是 CyclicBarrier(int parties)
,其参数表示屏障拦截的线程数量,每个线程调用 await()
方法告诉 CyclicBarrier
我已经到达了屏障,然后当前线程被阻塞。
public CyclicBarrier(int parties) {
this(parties, null);
}
public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties <= 0) throw new IllegalArgumentException();
this.parties = parties;
this.count = parties;
this.barrierCommand = barrierAction;
}
其中,parties
就代表了有拦截的线程的数量,当拦截的线程数量达到这个值的时候就打开栅栏,让所有线程通过。
2、当调用 CyclicBarrier
对象调用 await()
方法时,实际上调用的是 dowait(false, 0L)
方法。await()
方法就像树立起一个栅栏的行为一样,将线程挡住了,当拦住的线程数量达到 parties
的值时,栅栏才会打开,线程才得以通过执行。