说到线程同步第一个想到的一定是Thread类的join方法,当需要等待一个线程执行结束之后再接着执行本线程接下去的逻辑,join经常会在这种场景下被使用,但是join方法是Thread类的方法,需要使用Thread类实例进行调用,不够灵活,不能满足不同场景的需求,例如使用线程池来管理线程的时,一般直接添加Runnable到线程池,这个时候就没办法再调用线程的join方法,而接下来要介绍的CountDownLatch就可以满足该需求。
CountDownLatch是一个递减计数器,当计数器的值递减为0时,调用CountDownLatch类的await方法的线程会被唤醒,从而接着执行该线程的后续逻辑,CountDownLatch的计数器是一次性的,也就是等到计数器值变为0后,再调用CountDownLatch的await和countdown方法会立刻返回。下面通过一个简单的需求来介绍下CountDownLatch的使用。
使用线程池执行10个任务,当所有任务执行完之后,主线程打印出“任务完成!”。
final int taskCount = 10;
//(1)初始化一个计数值为10的CountDownLatch。
CountDownLatch countDownLatch = new CountDownLatch(taskCount);
ExecutorService executorService = Executors.newFixedThreadPool(5);
System.out.println("开始执行任务");
for (int i = 0; i < taskCount; i++) {
//(2)每个任务执行完之后都将CountDownLatch的值减1,注意:在finally中进行减操作
executorService.submit(new Runnable() {
@Override
public void run() {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
countDownLatch.countDown();
}
}
});
}
System.out.println("等待任务执行");
try {
//(3)调用CountDownLatch的await方法进行阻塞,等待计数器的值为0后被唤醒。
countDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("任务完成!!");
上面的例子展示了CountDownLatch实现线程间同步,那么CountDownLatch是如何实现线程间同步?CountDownLatch是如何进行计数器值递减?CountDownLatch的计数器值是如何初始化的?CountDownLatch在使用中会有哪些问题?带着这些疑问去CountDownLatch内部进行一探究竟。
CountDownLatch内部使用AQS实现,关于AQS读者可以翻看前面的文章。将初始化的计数值传递给AQS中的state,并通过共享的方式进行state的获取和释放。
CountDownLatch构造函数
CountDownLatch构造函数创建内部类Sync继承自AQS,将初始化的计数器值赋值给AQS中的state。
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}
private static final class Syncextends AbstractQueuedSynchronizer
Sync(int count) {
setState(count);
}
public void await()
阻塞挂起调用线程,直到计数器减到0。调用线程可被中断返回。
public void await() throws InterruptedException {
//(1)调用AQS中共享方式获取锁方法。
sync.acquireSharedInterruptibly(1);
}
//AQS中共享方式获取锁。
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
//(1)对中断响应。被中断则抛出异常。
if (Thread.interrupted())
throw new InterruptedException();
//(2)尝试获取共享锁,如果返回值小于0则将当前线程放到AQS的阻塞队列中并阻塞当前线程。
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
//CountDownLatch内部类Sync类实现的AQS接口。
protected int tryAcquireShared(int acquires) {//(1)判断计数器值是否为0,也就是判断AQS中的state是否为0。
return (getState() == 0) ? 1 : -1;
}
public void countDown()
递减计数器,当减到0的时候,唤醒所有调用CountDownLatch.await()方法而阻塞的线程。如果计数器的值为0则不做任何处理。
public void countDown() {
sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
//(1)如果计数器值减到0返回true,则将所有因为调用CountDownLatch.await()方法而阻塞的线程唤醒。
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
//CountDownLatch内部类Sync类实现的AQS接口。
protected boolean tryReleaseShared(int releases) {// (1)使用CAS+自旋的方式将计数器值减1,也就是将AQS中的state值减1,如果计数器值为0则直接返回false
for (;;) {
int c = getState();
if (c == 0)
return false;
int nextc = c-1;
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
CountDownLatch的原理就是通过内部类Sync实现AQS的tryAcquireShared和tryReleaseShared方法,以共享锁的原理来实现计时器的递减,从而实现线程间的同步。
今天的分享就到这,有看不明白的地方一定是我写的不够清楚,所有欢迎提任何问题以及改善方法。