JUC中的同步器三个主要的成员:CountDownLatch
、CyclicBarrier
和Semaphore
。这三个是JUC
中较为常用的同步器,通过它们可以方便地实现很多线程之间协作的功能。
CountDownLatch
是基于AQS
实现,当构建count
对象时,传入的值其实就会赋值给AQS
的关键变量state
,执行countDown()
方法时,其实就是利用CAS
将state-1
,执行await
方法时,其实就是判断state
是否为0,不为0则加入到队列中,将该线程阻塞(除头节点),因为头节点会一直自旋等待state
为0,当state
为0时,头节点把剩余的队列中阻塞的节点一并唤醒。
private static final class Sync extends AbstractQueuedSynchronizer { private static final long serialVersionUID = 4982264981922014374L; /** 构造方法,接收count值,只有count减小为0时,线程才不会被await方法阻塞 */ Sync(int count) { // CountDownLatch利用AQS的方式就是直接让count作为AQS的同步变量state // 所以直接用state记录count值 setState(count); } /** 获取当前的count值 */ int getCount() { return getState(); } /** * 这是AQS的模板方法acquireShared、acquireSharedInterruptibly等方法内部将会调用的方法, * 由子类实现,这个方法的作用是尝试获取一次共享锁,对于AQS来说, * 此方法返回值大于等于0,表示获取共享锁成功,反之则获取共享锁失败, * 而在这里,实际上就是判断count是否等于0,线程能否向下运行 */ protected int tryAcquireShared(int acquires) { // 此处判断state的值是否为0,也就是判断count是否为0, // 若count为0,返回1,表示获取锁成功,此时线程将不会阻塞,正常运行 // 若count不为0,则返回-1,表示获取锁失败,线程将会被阻塞 // 从这里我们已经可以看出CountDownLatch的实现方式了 return (getState() == 0) ? 1 : -1; } /** * 此方法的作用是用来是否AQS的共享锁,返回true表示释放成功,反之则失败 * 此方法将会在AQS的模板方法releaseShared中被调用, * 在CountDownLatch中,这个方法用来减小count值 */ protected boolean tryReleaseShared(int releases) { // 使用死循环不断尝试释放锁 for (;;) { // 首先获取当前state的值,也就是count值 int c = getState(); // 若count值已经等于0,则不能继续减小了,于是直接返回false // 为什么返回的是false,因为等于0表示之前等待的那些线程已经被唤醒了, // 若返回true,AQS会尝试唤醒线程,若返回false,则直接结束,所以 // 在没有线程等待的情况下,返回false直接结束是正确的 if (c == 0) return false; // 若count不等于0,则将其-1 int nextc = c-1; // compareAndSetState的作用是将count值从c,修改为新的nextc // 此方法基于CAS实现,保证了操作的原子性 if (compareAndSetState(c, nextc)) // 若nextc == 0,则返回的是true,表示已经没有锁了,线程可以运行了, // 若nextc > 0,则表示线程还需要继续阻塞,此处将返回false return nextc == 0; } } }
内部类Sync
的实现非常简单,它只实现了AQS
中的两个方法,即tryAcquireShared以及tryReleaseShared,这两个方法是AQS
提供的使用共享锁的接口。这也就表明,CountDownLatch
实际上是一种共享锁机制,即锁可以同时被多个线程获取,这个不难理解,因为一旦count
被减小为0,则所有线程通过await
方法时,都能够顺利通过,不会因为获取不到锁而阻塞。而且从上面的实现中我们可以看到,Sync
直接将count
值作为AQS
的state
的值,只有state
的值为0,线程才能获取锁,也就是获得执行权限。
/** * 只有一个成员变量,就是内部类Sync的一个对象,通过此对象调用AQS的方法,实现线程阻塞和唤醒 */ private final Sync sync; /** * 只有一个构造方法,接收一个count值 */ public CountDownLatch(int count) { // count值不能小于0 if (count < 0) throw new IllegalArgumentException("count < 0"); // 直接创建一个Sync对象,并传入count值,Sync内部将会执行setState(count) this.sync = new Sync(count); }
CountDownLatch
类最最核心的两个方法就是await
以及ountDown
,我们先来看一看await
方法的实现:
// 此方法用来让当前线程阻塞,直到count减小为0才恢复执行 public void await() throws InterruptedException { // 这里直接调用sync的acquireSharedInterruptibly方法,这个方法定义在AQS中 // 方法的作用是尝试获取共享锁,若获取失败,则线程将会被加入到AQS的同步队列中等待 // 直到获取成功为止。且这个方法是会响应中断的,线程在阻塞的过程中,若被其他线程中断, // 则此方法会通过抛出异常的方式结束等待。 sync.acquireSharedInterruptibly(1); }
await
的实现异常简单,只有短短一行代码,调用了AQS
中已经封装好的方法。这就是AQS
的好处,AQS
已经实现了线程的阻塞和唤醒机制,将实现的复杂性隐藏,而其他类只需要简单的使用它即可。为了方便理解,我们还是来看看acquireSharedInterruptibly
方法吧:
/** 此方法是AQS中提供的一个模板方法,用以获取共享锁,并且会响应中断 */ public final void acquireSharedInterruptibly(int arg) throws InterruptedException { // 首先判断当前线程释放被中断,若被中断,则直接抛出异常结束 if (Thread.interrupted()) throw new InterruptedException(); // 调用tryAcquireShared方法尝试获取锁,这个方法被Sycn类重写了, // 若count == 0,则这个方法会返回1,表示获取锁成功,则这里会直接返回,线程不会被阻塞 // 若count < 0,将会执行下面的doAcquireSharedInterruptibly方法, // 此处请去查看Sync中tryAcquireShared方法的实现 if (tryAcquireShared(arg) < 0) // 下面这个方法的作用是,线程获取锁失败,将会加入到AQS的同步队列中阻塞等待, // 直到成功获取到锁,而此处成功获取到锁的条件就是count == 0,若当前线程在等待的过程中, // 成功地获取了锁,则它会继续唤醒在它后面等待的线程,也尝试获取锁, // 这也就是说,只要count == 0了,则所有被阻塞的线程都能恢复运行 doAcquireSharedInterruptibly(arg); }
可以看出CountDownLatch
的实现完全就是依赖于AQS
/** * 此方法的作用就是将count的值-1,如果count等于0了,就唤醒等待的线程 */ public void countDown() { // 这里直接调用sync的releaseShared方法,这个方法的实现在AQS中,也是AQS提供的模板方法, // 这个方法的作用是当前线程释放锁,若释放失败,返回false,若释放成功,则返回false, // 若锁被释放成功,则当前线程会唤醒AQS同步队列中第一个被阻塞的线程,让他尝试获取锁 // 对于CountDownLatch来说,释放锁实际上就是让count - 1,只有当count被减小为0, // 锁才是真正被释放,线程才能继续向下运行 sync.releaseShared(1); }
AQS
的releaseShared
的方法实现
public final boolean releaseShared(int arg) { // 调用tryReleaseShared尝试释放锁,这个方法已经由Sycn重写,请回顾上面对此方法的分析 // 若tryReleaseShared返回true,表示count经过这次释放后,等于0了,于是执行doReleaseShared if (tryReleaseShared(arg)) { // 这个方法的作用是唤醒AQS的同步队列中,正在等待的第一个线程 // 而我们分析acquireSharedInterruptibly方法时已经说过, // 若一个线程被唤醒,检测到count == 0,会继续唤醒下一个等待的线程 // 也就是说,这个方法的作用是,在count == 0时,唤醒所有等待的线程 doReleaseShared(); return true; } return false; }
通过CountDownLatch
的源码会发现,它的实现真的非常简单,除了重写AQS
的两个方法外,其余的基本上就是调用AQS
提供的模板方法而已。
CountDownLatch的一个非常典型的应用场景是:有一个任务想要往下执行,但必须要等到其他的任务执行完毕后才可以继续往下执行。假如我们这个想要继续往下执行的任务调用一个CountDownLatch对象的await()方法,其他的任务执行完自己的任务后调用同一个CountDownLatch对象上的countDown()方法,这个调用await()方法的任务将一直阻塞等待,直到这个CountDownLatch对象的计数值减到0为止。
举个例子,有三个工人在为老板干活,这个老板有一个习惯,就是当三个工人把一天的活都干完了的时候,他就来检查所有工人所干的活。记住这个条件:三个工人先全部干完活,老板才检查。所以在这里用Java代码设计两个类,Worker代表工人,Boss代表老板,具体的代码实现如下:
import java.util.Random; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; public class Worker implements Runnable{ private CountDownLatch downLatch; private String name; public Worker(CountDownLatch downLatch, String name){ this.downLatch = downLatch; this.name = name; } public void run() { this.doWork(); try{ TimeUnit.SECONDS.sleep(new Random().nextInt(10)); }catch(InterruptedException ie){ } System.out.println(this.name + "活干完了!"); this.downLatch.countDown(); } private void doWork(){ System.out.println(this.name + "正在干活!"); } }
import java.util.concurrent.CountDownLatch; public class Boss implements Runnable { private CountDownLatch downLatch; public Boss(CountDownLatch downLatch){ this.downLatch = downLatch; } public void run() { System.out.println("老板正在等所有的工人干完活......"); try { this.downLatch.await(); } catch (InterruptedException e) { } System.out.println("工人活都干完了,老板开始检查了!"); } }
import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; public class CountDownLatchDemo { public static void main(String[] args) { ExecutorService executor = Executors.newCachedThreadPool(); CountDownLatch latch = new CountDownLatch(3); Worker w1 = new Worker(latch,"张三"); Worker w2 = new Worker(latch,"李四"); Worker w3 = new Worker(latch,"王二"); Boss boss = new Boss(latch); executor.execute(w3); executor.execute(w2); executor.execute(w1); executor.execute(boss); executor.shutdown(); } }
当你运行CountDownLatchDemo这个对象的时候,你会发现是等所有的工人都干完了活,老板才来检查,下面是我本地机器上运行的一次结果,可以肯定的每次运行的结果可能与下面不一样,但老板检查永远是在后面的。
王二正在干活! 李四正在干活! 老板正在等所有的工人干完活...... 张三正在干活! 张三活干完了! 王二活干完了! 李四活干完了! 工人活都干完了,老板开始检查了!
CountDownLatch
底层通过AQS
实现,AQS
的一般使用方式就是以内部类的形式继承它,CountDownLatch
就是这么使用它的。在CountDownLatch
内部有一个内部类Sync
,继承自AQS
,并重写了AQS
加锁解锁的方法,通过Sync
的对象,调用AQS
的方法,阻塞线程的运行。创建一个CountDownLatch
对象时,需要传入一个整数值count
,只有当count
被减小为0
时线程才能通过await
方法,否则将被await
阻塞。这里实际上是这样的:当线程运行到await方法时,需要去获取锁(锁由AQS实现),若count不为0,则线程就会获取锁失败,被阻塞;若count为0,则就能顺利通过。CountDownLatch
是一次性的,因为没有方法可以增加count
的值,也就是说,一旦count
被减小为0
,则之后就一直是0
了,也就再也不能阻塞线程了。