为了巩固AQS. 看一下CountDownLatch的源码.
大部分都是直接翻译的官方代码注释,嘻嘻
一个同步器, 允许一个或者多个线程等待, 知道其他线程完成一系列操作.
初始化时提供一个数字. await
方法将阻塞,直到别的线程通过调用countDown
,达到给定的数字.
这个类是一个一次性,count数字不能被重新设置. 如果你需要一个可复用的版本,可以考虑使用CyclicBarrier
.
CountDownLatch
可以用于以下目的:
await
等待,直到门被一个线程调用countDown
来打开.CountDownLatch
的一个很有用的特性是:
所有调用countDown
的线程不需要等待计数到达0. 他只是在await
方法上阻塞所有想要通过的线程.
使用实例(来自官方文档):
两个类,使用两个CountDownLatch
来完成以下功能.
CountDownLatch
, 是一个开始信号,告诉所有工作线程,驱动已经就绪,可以开始工作了。CountDownLatch
, 是一个结束信号,允许驱动等待所有工作线程完成,之后进行其他工作. class Driver {
void main() throws InterruptedException {
CountDownLatch startSignal = new CountDownLatch(1);
CountDownLatch doneSignal = new CountDownLatch(N);
for (int i = 0; i < N; ++i) // create and start threads
new Thread(new Worker(startSignal, doneSignal)).start();
doSomethingElse(); // don't let run yet
startSignal.countDown(); // let all threads proceed
doSomethingElse();
doneSignal.await(); // wait for all to finish
}
}
class Worker implements Runnable {
private final CountDownLatch startSignal;
private final CountDownLatch doneSignal;
Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {
this.startSignal = startSignal;
this.doneSignal = doneSignal;
}
public void run() {
try {
startSignal.await();
doWork();
doneSignal.countDown();
} catch (InterruptedException ex) {
} // return;
}
void doWork() {
}
}
另外一个典型应用是,将一个任务分割成N部分,每一个部分封装成一个任务,交给线程池。 然后一个协调线程,调用`await`等到所有的子部分完成. 再通过.
class Driver2 {
void main() throws InterruptedException {
CountDownLatch doneSignal = new CountDownLatch(N);
Executor e = null; // some Executor
for (int i = 0; i < N; ++i) // create and start threads
e.execute(new WorkerRunnable(doneSignal, i));
doneSignal.await(); // wait for all to finish
}
}
class WorkerRunnable implements Runnable {
private final CountDownLatch doneSignal;
private final int i;
WorkerRunnable(CountDownLatch doneSignal, int i) {
this.doneSignal = doneSignal;
this.i = i;
}
public void run() {
try {
doWork(i);
doneSignal.countDown();
} catch (InterruptedException ex) {} // return;
}
void doWork() {}
}
最核心的实现,依然是继承自AQS的一个子类同步器Sync
.
/**
* Synchronization control For CountDownLatch.
* Uses AQS state to represent count.
*/
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;
}
}
}
首先,初始化时传递的Count
值,复用AQS中的状态State
.
实现了AQS的共享模式加锁及共享模式解锁.
共享模式的加锁,锁空闲就返回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;
}
}
递减Count. 如果减1之后为0,就认为解锁成功. 通知.
如果减去1之后不为0. 返回false. 意味着解锁了,但是没有完全解锁成功.
没啥说的, 将传入的Count值传入Sync.复用AQS的状态值来实现Count的控制.
调用Sync同步器的释放共享锁方法,进行一次解锁操作.
调用Sync同步器的获取共享锁方法,进行加锁操作.
CountDownLatch是对AQS的共享模式的比较精巧的应用.
await
方法成功返回,成功通过CountDownLatch
了.为什么CountDownLatch是一次性的?
CountDownLatch中的同步器实现,并不是传统意义上的可以不断加锁或解锁。
只有在初始化时进行了设置State的操作,之后只可以进行读取/递减.
他的加锁操作,不会设置State的值,只是判断State是否大于1.
当解锁完成,State为0. 此时没有渠道去进行更新State的值.
如果重复的调用加锁,会不断的拿到”加锁成功”. 但是State数值并不会改变.
因此此时的”加锁成功”,其实意味着”门已经打开,可以无限进入”. 每一次的”加锁操作”, 约等于判断”门是否开着”的操作.
完.