使用方法
CountDownlatch在初始化的时候,需要赋予一个初始值,用来初始化计数器,然后在子线程里面调用countDown方法来进行减一,当计数器的值为0的时候,调用countdownlatch.await()的线程才能够继续往下走,在值为0之前,调用await的线程会一直被阻塞。
下面的实例代码中,由于在初始化CountDownlatch的时候传入的值是100,但是实际只被减了99次,所以主线程会一直被阻塞住
public class TestCountDownLatch
{
private static CountDownLatch countDownLatch = new CountDownLatch(100);
private static void costCountdownLatch()
{
System.out.println(Thread.currentThread().getName());
countDownLatch.countDown();
}
public static void main(String[] args) throws Exception
{
System.out.println("开始跑");
for (int i = 0; i < 99; i++)
{
new Thread(TestCountDownLatch::costCountdownLatch).start();
}
System.out.println("开始等待");
countDownLatch.await();
System.out.println("结束");
}
}
源码
CountDownLatch里面使用AQS来做线程之间同步的
Sync
使用传入的count来作为AQS中的state进行同步状态的维护
private static final class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 4982264981922014374L;
Sync(int count) {
setState(count);
}
int getCount() {
return getState();
}
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
protected boolean tryReleaseShared(int releases) {
// Decrement count; signal when transition to zero
for (;;) {
int c = getState();
if (c == 0)
return false;
int nextc = c-1;
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
}
初始化函数
根据传入的值,设置AQS的state变量
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}
await
AQS源码分析 await调用了AQS的共享获取,并且使用了上面重写的tryAcquireShared。
能够从上面的代码看到,在state不等于0的时候,一直都是返回-1,也就是获取锁失败
后面的acquire就是AQS里面的acquire方法
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted() ||
(tryAcquireShared(arg) < 0 &&
acquire(null, arg, true, true, false, 0L) < 0))
throw new InterruptedException();
}
countDown
对计数器减一,如果计数器已经等于0了,就通知等待的线程
public void countDown() {
sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
signalNext(head);
return true;
}
return false;
}